diff --git a/backend/AGENTS.md b/backend/AGENTS.md index 88677c7fa..22f08a28a 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -432,7 +432,7 @@ Localhost persistence deliberately reads the direct request `Host` and ignores ` | **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_ready` (whether `lark-cli` will actually be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants | | **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data | | **Uploads** (`/api/threads/{id}/uploads`) | `POST /` - upload files (auto-converts PDF/PPT/Excel/Word); `GET /list` - list; `DELETE /{filename}` - delete | -| **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - create a new main-thread branch from a completed assistant turn checkpoint and, when an addressable pre-user replay checkpoint exists, materialize it into the branch namespace so the inherited response remains regeneratable. Workspace files are not checkpointed, so the branch only best-effort copies the current workspace when branching from the **latest** turn (`workspace_clone_mode="current_thread_best_effort"`); branching from an older/historical turn skips the copy (`workspace_clone_mode="skipped_historical_turn"`) so the branch never inherits files that only exist in a later timeline. Branch creation also seeds the new thread's run-event feed from the branch checkpoint's visible messages (`history_seed_mode` in the response): the thread feed reads run_events, not checkpoints, so without the seed the inherited history disappears from the UI after the branch's first run (#4380). Seeded rows are grouped into one synthetic run per inherited turn (`branch-seed-{thread_id}-{n}`, a new turn opening at every persisted human message, including an allowlisted hidden `ask_clarification` reply) because `run_id` is a turn identity to the feed's consumers, not a provenance tag: regenerating an inherited answer supersedes that row's whole `run_id` in `GET /messages/page`, so one shared id for the entire seed deleted the complete inherited history on a branch's first regenerate (#4458); `GET /goal`, `PUT /goal`, `DELETE /goal` - read, set, and clear the active thread goal; `POST /compact` - manually summarize older active context into `summary_text` and retain the recent message window, blocked while a run is in flight; unexpected failures are logged server-side and return a generic 500 detail | +| **Threads** (`/api/threads/{id}`) | `DELETE /` - remove DeerFlow-managed local thread data after LangGraph thread deletion; `POST /branches` - create a new main-thread branch from a completed assistant turn checkpoint and, when an addressable pre-user replay checkpoint exists, materialize it into the branch namespace so the inherited response remains regeneratable. Workspace files are not checkpointed, so the branch only best-effort copies the current workspace when branching from the **latest** turn (`workspace_clone_mode="current_thread_best_effort"`); branching from an older/historical turn skips the copy (`workspace_clone_mode="skipped_historical_turn"`) so the branch never inherits files that only exist in a later timeline. Thread-scoped runtime channels (`sandbox`, `thread_data`) are not copied onto the branch: the parent's `sandbox_id` binds path mappings and the release lifecycle to the parent's workspace, so the branch lazily acquires its own sandbox instead. Branch creation also seeds the new thread's run-event feed from the branch checkpoint's visible messages (`history_seed_mode` in the response): the thread feed reads run_events, not checkpoints, so without the seed the inherited history disappears from the UI after the branch's first run (#4380). Seeded rows are grouped into one synthetic run per inherited turn (`branch-seed-{thread_id}-{n}`, a new turn opening at every persisted human message, including an allowlisted hidden `ask_clarification` reply) because `run_id` is a turn identity to the feed's consumers, not a provenance tag: regenerating an inherited answer supersedes that row's whole `run_id` in `GET /messages/page`, so one shared id for the entire seed deleted the complete inherited history on a branch's first regenerate (#4458); `GET /goal`, `PUT /goal`, `DELETE /goal` - read, set, and clear the active thread goal; `POST /compact` - manually summarize older active context into `summary_text` and retain the recent message window, blocked while a run is in flight; unexpected failures are logged server-side and return a generic 500 detail | | **Artifacts** (`/api/threads/{id}/artifacts`) | `GET /{path}` - serve artifacts; active content types (`text/html`, `application/xhtml+xml`, `image/svg+xml`) are always forced as download attachments to reduce XSS risk; `?download=true` still forces download for other file types | | **Suggestions** (`/api/suggestions`) | `GET /config` - returns global suggestions config boolean; `POST /threads/{id}/suggestions` - generate follow-up questions; rich list/block model content is normalized and inline reasoning (`...`, including unclosed/truncated blocks from reasoning models like MiniMax-M3) is stripped before JSON parsing | | **Input Polish** (`/api/input-polish`) | `POST /` - rewrite a composer draft before it is sent. This is a short authenticated `runs:create` LLM request using `input_polish` config; it does not create a LangGraph run, persist a message, or modify thread state. Shares the non-graph one-shot LLM path (`deerflow.utils.oneshot_llm.run_oneshot_llm`) with the suggestions route so model build + Langfuse metadata + invoke stay in one place; validates the same stripped view of the draft it sends to the model, and preserves literal `` substrings in the rewrite (`strip_think_blocks(truncate_unclosed=False)`) | @@ -872,7 +872,7 @@ Checkpointer storage runs in one of two channel modes, selected by `checkpoint_c **Where things live**: - `runtime/checkpoint_mode.py` — mode freeze, marker injection, delta detection, compatibility gate, both error types - `runtime/checkpoint_state.py` — `CheckpointStateAccessor`, `build_state_mutation_graph`, `RollbackPoint` -- `checkpoint_patches.py` (package root) — saver patches: delta-history folding for `InMemorySaver` (delegating to the base walk), stable message IDs across materialization, upstream first-write drop fix +- `checkpoint_patches.py` (package root) — checkpoint-machinery patches: delta-history folding for `InMemorySaver` (delegating to the base walk), stable message IDs across materialization, upstream first-write drop fix, and `BinaryOperatorAggregate` unwrapping an `Overwrite` first write into an empty (MISSING) channel — Union-typed reducer channels (`sandbox`/`goal`/`todos`/`promoted`) have no constructible default, so a replace-style write into a fresh branch thread or a never-written channel stored the wrapper literally and crashed the next consumer (#4380; probe-guarded, stands down if upstream fixes it) - `agents/thread_state.py` — `ThreadState`/`DeltaThreadState`, `DELTA_MESSAGES_FIELD` (`DeltaChannel` with `snapshot_frequency=1000`), schema adaptation helpers - `runtime/context_compaction.py` — compaction via accessor + mutation graph (reference consumer) - Tests: `tests/test_checkpoint_mode.py` (freeze/detect/gate), `tests/test_checkpoint_state.py` (accessor/mutation graph), `tests/test_delta_channel_checkpointers.py` (saver parity), `tests/test_threads_checkpoint_mode.py`, `tests/test_gateway_checkpoint_mode.py` (dual-mode e2e parity), `tests/test_context_compaction.py` (mutation-graph write, no scheduling), `tests/test_run_worker_rollback.py` diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index 0e323a18b..b3cf2d5da 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -97,6 +97,14 @@ def _checkpoint_mode_http_error(exc: Exception, thread_id: str) -> HTTPException _SERVER_RESERVED_METADATA_KEYS: frozenset[str] = frozenset({"owner_id", "user_id"}) _SIDECAR_METADATA_KEY = "deerflow_sidecar" _BRANCH_METADATA_KEY = "deerflow_branch" +# Thread-scoped runtime channels a branch must NOT inherit from its parent: +# ``sandbox.sandbox_id`` binds path mappings and the release lifecycle to the +# *parent* thread, so copying it would make the branch read/write the parent's +# workspace (bypassing the per-branch user-data clone) and release the +# parent's sandbox after its first run; the branch lazily acquires its own +# sandbox keyed by its own thread_id instead. ``thread_data`` is recomputed +# from the branch's thread_id by ThreadDataMiddleware on every run. +_BRANCH_EXCLUDED_CHANNELS = frozenset({"sandbox", "thread_data"}) _BRANCH_HISTORY_SCAN_LIMIT = 200 _BRANCH_HISTORY_RAW_SCAN_LIMIT = _BRANCH_HISTORY_SCAN_LIMIT * 2 @@ -830,6 +838,8 @@ async def branch_thread(thread_id: str, body: ThreadBranchRequest, request: Requ def branch_values(source_snapshot: Any) -> dict[str, Any]: values: dict[str, Any] = {} for key, value in dict(source_snapshot.values).items(): + if key in _BRANCH_EXCLUDED_CHANNELS: + continue if key in branch_reducer_fields: values[key] = Overwrite(list(value) if key == "messages" and isinstance(value, list) else value) else: diff --git a/backend/packages/harness/deerflow/checkpoint_patches.py b/backend/packages/harness/deerflow/checkpoint_patches.py index c82cc573c..180bbcea2 100644 --- a/backend/packages/harness/deerflow/checkpoint_patches.py +++ b/backend/packages/harness/deerflow/checkpoint_patches.py @@ -1,4 +1,4 @@ -"""Compatibility patches for third-party checkpoint savers. +"""Compatibility patches for third-party checkpoint machinery. Lives at the top-level package (not ``deerflow.runtime``) so it can be imported from ``deerflow.agents.thread_state`` without pulling in the heavy @@ -12,10 +12,14 @@ from __future__ import annotations import importlib.metadata import logging +from collections.abc import Sequence from typing import Any +from langgraph.channels.binop import BinaryOperatorAggregate from langgraph.checkpoint.base import BaseCheckpointSaver from langgraph.checkpoint.memory import InMemorySaver +from langgraph.errors import ErrorCode, InvalidUpdateError, create_error_message +from langgraph.types import Overwrite from packaging.version import Version logger = logging.getLogger(__name__) @@ -91,9 +95,98 @@ def ensure_inmemory_delta_history_patch() -> None: return InMemorySaver.get_delta_channel_history = _get_delta_channel_history_via_base # type: ignore[method-assign] InMemorySaver.aget_delta_channel_history = _aget_delta_channel_history_via_base # type: ignore[method-assign] - InMemorySaver._deerflow_delta_history_patched = True # type: ignore[attr-defined] + setattr(InMemorySaver, _PATCH_FLAG, True) except (AttributeError, TypeError): logger.warning("Failed to apply the InMemorySaver delta-history patch; leaving the upstream implementation untouched.", exc_info=True) +_BINOP_PATCH_FLAG = "_deerflow_overwrite_first_write_patched" +_unpatched_binop_update = BinaryOperatorAggregate.update + + +def _as_overwrite(value: Any) -> tuple[bool, Any]: + """Local stand-in for langgraph's private ``_get_overwrite``. + + Matches only the public ``Overwrite`` *class* form - the sole form + DeerFlow's write paths produce for these Union channels (the branch and + ``/state`` routes wrap replace-style writes in ``Overwrite(...)``). + Avoiding the underscored ``_get_overwrite`` import keeps an upstream + refactor that drops it - plausibly the very release that fixes the bug - + from failing this module's import and crashing startup before the probe + can stand the patch down. The dict sentinel form upstream also accepts is + an internal serialization detail DeerFlow never emits into these channels. + """ + if isinstance(value, Overwrite): + return True, value.value + return False, None + + +def _binop_first_write_stores_overwrite_wrapper() -> bool: + """Probe whether upstream still stores an Overwrite first write literally. + + Uses a Union-typed channel (no constructible default, so it starts + MISSING) - the same shape as ``ThreadState``'s ``sandbox`` / ``goal`` / + ``todos`` / ``promoted`` channels. + """ + channel = BinaryOperatorAggregate(dict | None, lambda existing, new: new) + channel.key = "deerflow-overwrite-probe" + channel.update([Overwrite({"probe": True})]) + return isinstance(channel.get(), Overwrite) + + +def _binop_update_unwrapping_empty_channel(self: Any, values: Sequence[Any]) -> bool: + """``BinaryOperatorAggregate.update`` that unwraps an Overwrite first write. + + Only intercepts the empty-channel + leading-Overwrite case; everything + else delegates to the upstream implementation. The intercepted case + mirrors upstream's own post-Overwrite batch semantics: later plain values + are skipped and a second Overwrite raises ``InvalidUpdateError``. + """ + if not self.is_available() and values: + is_overwrite, overwrite_value = _as_overwrite(values[0]) + if is_overwrite: + self.value = overwrite_value + for value in values[1:]: + if _as_overwrite(value)[0]: + msg = create_error_message( + message="Can receive only one Overwrite value per super-step.", + error_code=ErrorCode.INVALID_CONCURRENT_GRAPH_UPDATE, + ) + raise InvalidUpdateError(msg) + return True + return _unpatched_binop_update(self, values) + + +def ensure_binop_overwrite_first_write_patch() -> None: + """Fix ``Overwrite`` first writes being stored literally on empty channels. + + Upstream ``BinaryOperatorAggregate.update`` seeds an empty channel + (``self.value is MISSING``) with ``values[0]`` verbatim - without the + Overwrite unwrapping the rest of the method applies. 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) then persists the ``Overwrite`` wrapper itself into the + checkpoint, and the next consumer crashes with ``TypeError: 'Overwrite' + object is not subscriptable`` (#4380). ``DeltaChannel.update`` already + unwraps in the same situation, so this also removes a behavioral + inconsistency between the two reducer channel types. + + Idempotent. Guarded by a behavioral probe instead of a version pin: if a + future LangGraph unwraps the first write itself, the probe reports the + bug as absent and the patch stands down. + """ + if getattr(BinaryOperatorAggregate, _BINOP_PATCH_FLAG, False): + return + try: + if not _binop_first_write_stores_overwrite_wrapper(): + # Upstream unwraps the first write itself: nothing to patch. + return + BinaryOperatorAggregate.update = _binop_update_unwrapping_empty_channel # type: ignore[method-assign] + setattr(BinaryOperatorAggregate, _BINOP_PATCH_FLAG, True) + except Exception: + logger.warning("Failed to apply the BinaryOperatorAggregate Overwrite first-write patch; leaving the upstream implementation untouched.", exc_info=True) + + ensure_inmemory_delta_history_patch() +ensure_binop_overwrite_first_write_patch() diff --git a/backend/tests/test_checkpoint_patches.py b/backend/tests/test_checkpoint_patches.py new file mode 100644 index 000000000..0a25bdf95 --- /dev/null +++ b/backend/tests/test_checkpoint_patches.py @@ -0,0 +1,91 @@ +"""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 diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index 99a937a26..bc751f12f 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -2054,6 +2054,124 @@ def test_branch_history_seed_failure_keeps_branch_usable(monkeypatch) -> None: assert branch_response.json()["history_seed_mode"] == "failed" +async def _seed_union_channel_source(checkpointer, custom_factory, mode, source_thread_id): + """Seed a completed turn plus Union-typed reducer channels (sandbox/goal/todos).""" + accessor = CheckpointStateAccessor.bind(custom_factory(), checkpointer, mode=mode) + config = {"configurable": {"thread_id": source_thread_id, "checkpoint_ns": ""}} + await accessor.aupdate( + config, + {"messages": [HumanMessage(id="h1", content="question")], "goal": {"objective": "ship the fix"}}, + as_node="model", + ) + await accessor.aupdate( + config, + { + "messages": [AIMessage(id="a1", content="answer")], + "todos": [{"content": "write tests", "status": "pending"}], + "sandbox": {"sandbox_id": "local:parent-thread"}, + "thread_data": {"workspace_path": "/parent/workspace"}, + }, + as_node="model", + ) + + +def _branch_union_channel_thread(monkeypatch, mode): + """Drive POST /branches on a source seeded with Union-typed channels; return branch values.""" + app, _store, checkpointer = _build_thread_app() + custom_factory = _wire_extension_agent(monkeypatch, app, checkpointer, mode) + source_thread_id = f"union-branch-source-{mode}" + + with TestClient(app) as client: + created = client.post( + "/api/threads", + json={"thread_id": source_thread_id, "metadata": {}, "assistant_id": "extension-agent"}, + ) + assert created.status_code == 200, created.text + + asyncio.run(_seed_union_channel_source(checkpointer, custom_factory, mode, source_thread_id)) + + branch_response = client.post( + f"/api/threads/{source_thread_id}/branches", + json={"message_id": "a1", "message_ids": ["a1"]}, + ) + assert branch_response.status_code == 200, branch_response.text + branch_thread_id = branch_response.json()["thread_id"] + + async def materialize(): + accessor = CheckpointStateAccessor.bind(custom_factory(), checkpointer, mode=mode) + snapshot = await accessor.aget({"configurable": {"thread_id": branch_thread_id, "checkpoint_ns": ""}}) + return snapshot.values + + return asyncio.run(materialize()) + + +@pytest.mark.parametrize("mode", ["full", "delta"]) +def test_branch_copies_union_typed_reducer_channels_as_plain_values(monkeypatch, mode) -> None: + """Branching must not persist Overwrite wrappers into the fresh thread (#4380). + + Union-typed reducer channels (``goal``, ``todos``, ``promoted``, + ``sandbox``) have no constructible default, so they start MISSING on the + branch thread; an ``Overwrite`` first write that isn't unwrapped is stored + literally and the next consumer crashes with ``TypeError: 'Overwrite' + object is not subscriptable``. + """ + branch_values = _branch_union_channel_thread(monkeypatch, mode) + + # The exact crash shape from #4380: subscripting the copied channel value. + assert branch_values["goal"]["objective"] == "ship the fix" + assert branch_values["todos"] == [{"content": "write tests", "status": "pending"}] + assert not any(isinstance(value, Overwrite) for value in branch_values.values()) + + +@pytest.mark.parametrize("mode", ["full", "delta"]) +def test_branch_does_not_inherit_thread_scoped_channels(monkeypatch, mode) -> None: + """The branch must acquire its own sandbox and thread paths, not the parent's. + + ``sandbox.sandbox_id`` binds path mappings and the release lifecycle to + the parent thread, so inheriting it would make the branch read/write the + parent's workspace and release the parent's sandbox after its first run; + ``thread_data`` is recomputed from the branch's own thread_id by + ThreadDataMiddleware on every run. + """ + branch_values = _branch_union_channel_thread(monkeypatch, mode) + + assert branch_values.get("sandbox") is None + assert branch_values.get("thread_data") is None + + +@pytest.mark.parametrize("mode", ["full", "delta"]) +def test_update_thread_state_overwrite_into_never_written_channel(monkeypatch, mode) -> None: + """POST /state must store a plain value when the reducer channel was never written. + + Same mechanism as the branch case (#4380): ``goal`` starts MISSING on a + thread that never wrote it, and the endpoint's replace-style ``Overwrite`` + wrapping must not be persisted literally. + """ + app, _store, checkpointer = _build_thread_app() + custom_factory = _wire_extension_agent(monkeypatch, app, checkpointer, mode) + source_thread_id = f"never-written-goal-{mode}" + + with TestClient(app) as client: + created = client.post( + "/api/threads", + json={"thread_id": source_thread_id, "metadata": {}, "assistant_id": "extension-agent"}, + ) + assert created.status_code == 200, created.text + + asyncio.run(_seed_extension_source(checkpointer, custom_factory, mode, source_thread_id)) + + update_response = client.post( + f"/api/threads/{source_thread_id}/state", + json={"values": {"goal": {"objective": "finish"}}}, + ) + assert update_response.status_code == 200, update_response.text + assert update_response.json()["values"]["goal"] == {"objective": "finish"} + + read_response = client.get(f"/api/threads/{source_thread_id}/state") + assert read_response.status_code == 200, read_response.text + assert read_response.json()["values"]["goal"] == {"objective": "finish"} + + def test_update_thread_state_rejects_unknown_state_fields(monkeypatch) -> None: """Unknown fields fail 422 instead of a false-success 200.""" app, _store, checkpointer = _build_thread_app()