fix(checkpoint): unwrap Overwrite first writes into empty channels (#4383)

* 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>

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 (`<think>...</think>`, 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 `<think>` 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`

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:

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()

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

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()