mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-26 16:07:53 +00:00
d1aeea2c3e
* fix(gateway): stop persisting Overwrite wrappers into empty reducer channels on branch Thread branching (and POST /state on a never-written channel) wraps copied reducer values in Overwrite. Upstream BinaryOperatorAggregate.update seeds an empty (MISSING) channel with values[0] verbatim without unwrapping, so Union-typed channels (sandbox/goal/todos/promoted) stored the wrapper literally and the next consumer crashed with TypeError: 'Overwrite' object is not subscriptable (#4380). Patch the channel to unwrap the first write (mirroring DeltaChannel semantics), and stop copying thread-scoped channels (sandbox, thread_data) into branches: the parent's sandbox_id would bind the branch to the parent's workspace and release lifecycle. * refactor(checkpoint): drop private _get_overwrite import for a local Overwrite check Importing langgraph's underscored _get_overwrite at module top level meant an upstream refactor that drops it - plausibly the same release that fixes the bug - would fail this module's import and crash startup before the probe can stand the patch down. Replace it with a local helper on the public Overwrite type, and fix two test docstring nits. * refactor(checkpoint): write patch flags via their constants to avoid drift Both saver patches read their "already patched" idempotence flag through a module constant (_PATCH_FLAG / _BINOP_PATCH_FLAG) but wrote it as a hard-coded attribute literal, so renaming the constant would silently break the guard and double-apply the patch. Write via the same constant (setattr), dropping the now-unneeded attr-defined ignores. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
92 lines
3.8 KiB
Python
92 lines
3.8 KiB
Python
"""Guards for the BinaryOperatorAggregate first-write Overwrite patch (#4380).
|
|
|
|
Upstream ``BinaryOperatorAggregate.update`` has a fast path for empty channels
|
|
(``self.value is MISSING``) that stores the first incoming value without
|
|
unwrapping ``Overwrite``. Reducer channels whose type is a Union
|
|
(``SandboxState | None``, ``GoalState | None``, ...) have no constructible
|
|
default, so they start MISSING — a replace-style write into a fresh thread
|
|
(thread branching) or a never-written channel (state update) persists the
|
|
wrapper literally, and the next consumer crashes with ``TypeError:
|
|
'Overwrite' object is not subscriptable``. ``DeltaChannel`` already unwraps in
|
|
the same situation; the patch aligns ``BinaryOperatorAggregate`` with it.
|
|
"""
|
|
|
|
import pytest
|
|
from langgraph.channels.binop import BinaryOperatorAggregate
|
|
from langgraph.errors import InvalidUpdateError
|
|
from langgraph.types import Overwrite
|
|
|
|
import deerflow.agents.thread_state # noqa: F401 - applies checkpoint patches at import
|
|
from deerflow import checkpoint_patches
|
|
|
|
|
|
def _replace(existing, new):
|
|
return new
|
|
|
|
|
|
def _empty_union_channel() -> BinaryOperatorAggregate:
|
|
"""A channel shaped like ``sandbox``/``goal``/``todos``: Union type, starts MISSING."""
|
|
channel = BinaryOperatorAggregate(dict | None, _replace)
|
|
channel.key = "probe"
|
|
return channel
|
|
|
|
|
|
def test_first_write_overwrite_unwraps_into_empty_channel() -> None:
|
|
channel = _empty_union_channel()
|
|
channel.update([Overwrite({"sandbox_id": "local:thread-2"})])
|
|
|
|
stored = channel.get()
|
|
assert not isinstance(stored, Overwrite)
|
|
# The exact read that crashed in #4380 (SandboxMiddleware.aafter_agent).
|
|
assert stored["sandbox_id"] == "local:thread-2"
|
|
|
|
|
|
def test_overwrite_into_populated_channel_still_replaces() -> None:
|
|
channel = _empty_union_channel()
|
|
channel.update([{"sandbox_id": "old"}])
|
|
channel.update([Overwrite({"sandbox_id": "new"})])
|
|
|
|
assert channel.get() == {"sandbox_id": "new"}
|
|
|
|
|
|
def test_plain_first_write_still_seeds_then_reduces() -> None:
|
|
channel = _empty_union_channel()
|
|
channel.update([{"sandbox_id": "first"}])
|
|
assert channel.get() == {"sandbox_id": "first"}
|
|
|
|
channel.update([{"sandbox_id": "second"}])
|
|
assert channel.get() == {"sandbox_id": "second"}
|
|
|
|
|
|
def test_values_after_first_write_overwrite_are_ignored() -> None:
|
|
"""Upstream semantics: after an Overwrite, later plain values in the batch are skipped."""
|
|
channel = _empty_union_channel()
|
|
channel.update([Overwrite({"sandbox_id": "kept"}), {"sandbox_id": "ignored"}])
|
|
|
|
assert channel.get() == {"sandbox_id": "kept"}
|
|
|
|
|
|
def test_double_overwrite_in_one_batch_raises_on_empty_channel() -> None:
|
|
channel = _empty_union_channel()
|
|
with pytest.raises(InvalidUpdateError):
|
|
channel.update([Overwrite({"sandbox_id": "a"}), Overwrite({"sandbox_id": "b"})])
|
|
|
|
|
|
def test_binop_overwrite_patch_is_active() -> None:
|
|
"""The compatibility patch must be applied in every test/app process."""
|
|
assert getattr(BinaryOperatorAggregate, checkpoint_patches._BINOP_PATCH_FLAG, False) is True
|
|
assert BinaryOperatorAggregate.update is checkpoint_patches._binop_update_unwrapping_empty_channel
|
|
|
|
|
|
def test_binop_overwrite_patch_stands_down_when_upstream_fixed(monkeypatch) -> None:
|
|
"""If upstream unwraps the first write itself, the patch must not reinstall."""
|
|
monkeypatch.setattr(checkpoint_patches, "_binop_first_write_stores_overwrite_wrapper", lambda: False)
|
|
monkeypatch.delattr(BinaryOperatorAggregate, checkpoint_patches._BINOP_PATCH_FLAG, raising=False)
|
|
sentinel = object()
|
|
monkeypatch.setattr(BinaryOperatorAggregate, "update", sentinel)
|
|
|
|
checkpoint_patches.ensure_binop_overwrite_first_write_patch()
|
|
|
|
assert getattr(BinaryOperatorAggregate, checkpoint_patches._BINOP_PATCH_FLAG, False) is False
|
|
assert BinaryOperatorAggregate.update is sentinel
|