"""Subgraph stream frames must not impersonate root-graph frames (#4399).

The gateway worker drives ``agent.astream(subgraphs=...)`` and publishes each
frame to the StreamBridge. Delegated subagent graphs inherit the parent's
checkpoint namespace (``subagents/executor.py``), so with ``subgraphs=True``
their values snapshots and token chunks arrive interleaved with root frames.
Publishing them under bare event names lets a subagent's values snapshot
replace the whole thread view in SDK clients and floods the parent message
stream with the subagent's token chunks. The namespace must ride the SSE event
name (LangGraph Platform style ``mode|ns1|ns2``) and namespaced frames must
bypass the root-only consumers (file-tool chunk batcher, subagent event
persistence).
"""

import asyncio
import importlib
import sys
from importlib.metadata import version as package_version
from types import SimpleNamespace

import pytest
from langchain_core.messages import AIMessage, HumanMessage, ToolMessage
from packaging.version import Version

from deerflow.runtime.runs import worker
from deerflow.runtime.runs.manager import RunRecord, RunStartOutcome
from deerflow.runtime.runs.schemas import DisconnectMode, RunStatus
from deerflow.runtime.runs.worker import (
    _compose_sse_event,
    _publish_stream_item,
    _unpack_stream_item,
)
from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge

SUBAGENT_NS = ("tools:call_subagent_1",)

# Delegated graphs inherit the parent checkpoint namespace (and therefore
# stream as subgraphs) only on LangGraph >= 1.2.6 — same gate as
# tests/test_subagent_executor.py::TestSubagentCheckpointLineage.
_LANGGRAPH_INHERITS_SUBGRAPH_NAMESPACE = Version(package_version("langgraph")) >= Version("1.2.6")


class _FakeBridge:
    def __init__(self) -> None:
        self.published: list[tuple[str, str, object]] = []

    async def publish(self, run_id: str, event: str, payload: object) -> None:
        self.published.append((run_id, event, payload))


class _FakeSubagentEvents:
    def __init__(self) -> None:
        self.added: list[object] = []

    async def add(self, chunk: object) -> None:
        self.added.append(chunk)


class _SpyBatcher:
    """Observable stand-in for _LargeFileToolChunkBatcher."""

    def __init__(self) -> None:
        self.pushed: list[object] = []
        self.finish_calls = 0
        self.flush_calls = 0

    def push(self, chunk: object) -> list[object]:
        self.pushed.append(chunk)
        return [chunk]

    def finish(self) -> list[object]:
        self.finish_calls += 1
        return []

    def flush(self) -> list[object]:
        self.flush_calls += 1
        return []


class TestUnpackStreamItem:
    def test_root_frame_with_subgraphs_has_empty_namespace(self):
        mode, chunk, namespace = _unpack_stream_item(((), "values", {"messages": []}), ["values"], True)
        assert mode == "values"
        assert namespace == ()

    def test_subgraph_frame_preserves_namespace(self):
        mode, chunk, namespace = _unpack_stream_item((SUBAGENT_NS, "values", {"messages": []}), ["values"], True)
        assert mode == "values"
        assert namespace == SUBAGENT_NS

    def test_nested_subgraph_namespace_is_preserved_in_order(self):
        ns = ("tools:call_a", "model_request:xyz")
        _mode, _chunk, namespace = _unpack_stream_item((ns, "messages", object()), ["messages"], True)
        assert namespace == ns

    def test_two_tuple_under_subgraphs_is_root(self):
        mode, _chunk, namespace = _unpack_stream_item(("custom", {"type": "task_started"}), ["custom"], True)
        assert mode == "custom"
        assert namespace == ()

    def test_without_subgraphs_frames_are_root(self):
        mode, _chunk, namespace = _unpack_stream_item(("values", {}), ["values"], False)
        assert mode == "values"
        assert namespace == ()

    def test_single_mode_fallback_is_root(self):
        mode, chunk, namespace = _unpack_stream_item({"messages": []}, ["values"], False)
        assert mode == "values"
        assert chunk == {"messages": []}
        assert namespace == ()

    def test_unparsable_item_under_subgraphs(self):
        mode, chunk, namespace = _unpack_stream_item("garbage", ["values"], True)
        assert mode is None
        assert chunk is None
        assert namespace == ()


class TestComposeSseEvent:
    def test_root_frame_keeps_bare_event_name(self):
        assert _compose_sse_event("values", ()) == "values"

    def test_subgraph_frame_gets_namespace_qualified_name(self):
        assert _compose_sse_event("values", SUBAGENT_NS) == "values|tools:call_subagent_1"

    def test_nested_namespace_joins_all_segments(self):
        assert _compose_sse_event("messages", ("tools:call_a", "model_request:xyz")) == "messages|tools:call_a|model_request:xyz"


class TestPublishStreamItem:
    @pytest.mark.asyncio
    async def test_subagent_values_snapshot_is_never_published_as_bare_values(self):
        # The #4399 regression: a delegated subagent's values snapshot published
        # as bare "values" replaces the whole thread view in SDK clients.
        bridge = _FakeBridge()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": [{"type": "human", "content": "subagent task prompt"}]},
            namespace=SUBAGENT_NS,
            file_tool_chunk_batcher=None,
            subagent_events=_FakeSubagentEvents(),
        )
        assert [event for _run, event, _payload in bridge.published] == ["values|tools:call_subagent_1"]

    @pytest.mark.asyncio
    async def test_root_values_snapshot_keeps_bare_event_name(self):
        bridge = _FakeBridge()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": []},
            namespace=(),
            file_tool_chunk_batcher=None,
            subagent_events=_FakeSubagentEvents(),
        )
        assert [event for _run, event, _payload in bridge.published] == ["values"]

    @pytest.mark.asyncio
    async def test_subagent_message_chunks_are_namespaced(self):
        bridge = _FakeBridge()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="messages",
            chunk=({"content": "token"}, {"langgraph_node": "model"}),
            namespace=SUBAGENT_NS,
            file_tool_chunk_batcher=_SpyBatcher(),
            subagent_events=_FakeSubagentEvents(),
        )
        assert [event for _run, event, _payload in bridge.published] == ["messages|tools:call_subagent_1"]

    @pytest.mark.asyncio
    async def test_root_custom_event_is_persisted_for_subagent_history(self):
        bridge = _FakeBridge()
        subagent_events = _FakeSubagentEvents()
        chunk = {"type": "task_started", "task_id": "call_1"}
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="custom",
            chunk=chunk,
            namespace=(),
            file_tool_chunk_batcher=None,
            subagent_events=subagent_events,
        )
        assert [event for _run, event, _payload in bridge.published] == ["custom"]
        assert subagent_events.added == [chunk]

    @pytest.mark.asyncio
    async def test_subgraph_custom_event_is_not_persisted(self):
        bridge = _FakeBridge()
        subagent_events = _FakeSubagentEvents()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="custom",
            chunk={"type": "noise"},
            namespace=SUBAGENT_NS,
            file_tool_chunk_batcher=None,
            subagent_events=subagent_events,
        )
        assert [event for _run, event, _payload in bridge.published] == ["custom|tools:call_subagent_1"]
        assert subagent_events.added == []

    @pytest.mark.asyncio
    async def test_only_root_frames_drive_the_file_tool_batcher(self):
        bridge = _FakeBridge()
        batcher = _SpyBatcher()
        # A subagent values frame must not finish() a pending root batch...
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": []},
            namespace=SUBAGENT_NS,
            file_tool_chunk_batcher=batcher,
            subagent_events=_FakeSubagentEvents(),
        )
        assert batcher.finish_calls == 0
        assert batcher.pushed == []
        # ...while a root values frame does.
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": []},
            namespace=(),
            file_tool_chunk_batcher=batcher,
            subagent_events=_FakeSubagentEvents(),
        )
        assert batcher.finish_calls == 1

    @pytest.mark.asyncio
    async def test_root_message_chunks_go_through_the_batcher(self):
        bridge = _FakeBridge()
        batcher = _SpyBatcher()
        chunk = ({"content": "token"}, {"langgraph_node": "model"})
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="messages",
            chunk=chunk,
            namespace=(),
            file_tool_chunk_batcher=batcher,
            subagent_events=_FakeSubagentEvents(),
        )
        assert batcher.pushed == [chunk]
        assert [event for _run, event, _payload in bridge.published] == ["messages"]


# ---------------------------------------------------------------------------
# Production-shaped integration: SubagentExecutor -> astream(subgraphs=...)
# -> run_agent stream loop -> StreamBridge. The namespace must originate from
# LangGraph's own delegation routing (checkpoint-namespace inheritance), not
# be hand-fed to the publishing helper — the #4399 regression lived in that
# interaction, not in any helper in isolation.
# ---------------------------------------------------------------------------

_CHILD_MESSAGE_IDS = frozenset(
    {
        "child-task-sentinel",
        "child-ai-sentinel",
        "child-tool-sentinel",
        "child-final-sentinel",
    }
)
_PARENT_FINAL_ID = "parent-final-sentinel"
_THREAD_ID = "thread-subgraph-stream-integration"


@pytest.fixture
def real_executor_module():
    """Swap the conftest MagicMock for the real subagent executor module.

    conftest.py mocks ``deerflow.subagents.executor`` to break a package-init
    import cycle; by the time this fixture runs every other deerflow module is
    already imported, so a fresh import of the real module is safe.
    """
    original = sys.modules.get("deerflow.subagents.executor")
    sys.modules.pop("deerflow.subagents.executor", None)
    subagents_pkg = sys.modules.get("deerflow.subagents")
    if subagents_pkg is not None and hasattr(subagents_pkg, "executor"):
        delattr(subagents_pkg, "executor")

    module = importlib.import_module("deerflow.subagents.executor")
    # Hermetic in CI (no config.yaml) — same defaults as test_subagent_executor.
    module.get_app_config = lambda: SimpleNamespace(tool_search=SimpleNamespace(enabled=False))
    module.build_tracing_callbacks = lambda: []
    yield module

    if original is not None:
        sys.modules["deerflow.subagents.executor"] = original
    else:
        sys.modules.pop("deerflow.subagents.executor", None)
    subagents_pkg = sys.modules.get("deerflow.subagents")
    if subagents_pkg is not None and hasattr(subagents_pkg, "executor"):
        delattr(subagents_pkg, "executor")


class _RecordingStreamBridge(MemoryStreamBridge):
    """Real in-memory bridge that also records (event, payload) pairs."""

    def __init__(self) -> None:
        super().__init__()
        self.published: list[tuple[str, object]] = []

    async def publish(self, run_id: str, event: str, payload: object) -> None:
        self.published.append((event, payload))
        await super().publish(run_id, event, payload)


class _IntegrationRunManager:
    def __init__(self, record: RunRecord) -> None:
        self._record = record

    async def try_start(self, _run_id):
        self._record.status = RunStatus.running
        return RunStartOutcome.started

    async def wait_for_prior_finalizing(self, *_args, **_kwargs):
        return None

    async def set_status(self, _run_id, status, **_kwargs):
        self._record.status = status

    async def set_status_if_not_cancelled(self, _run_id, status, **kwargs):
        await self.set_status(_run_id, status, **kwargs)
        return None

    async def update_model_name(self, *_args, **_kwargs):
        return None

    async def update_run_completion(self, *_args, **_kwargs):
        return None

    async def has_later_started_run(self, *_args, **_kwargs):
        return False

    async def set_finalizing(self, *_args, **_kwargs):
        return None

    async def cleanup(self, *_args, **_kwargs):
        return None


def _collect_ids(payload: object) -> set[str]:
    """All string ``id`` values anywhere in a serialized stream payload."""
    ids: set[str] = set()

    def walk(node: object) -> None:
        if isinstance(node, dict):
            node_id = node.get("id")
            if isinstance(node_id, str):
                ids.add(node_id)
            for value in node.values():
                walk(value)
        elif isinstance(node, (list, tuple)):
            for value in node:
                walk(value)

    walk(payload)
    return ids


def _build_delegating_parent_graph(executor_module, monkeypatch, *, child_emits_error_fallback: bool = False):
    """Real parent graph whose node delegates a scripted child through the
    real ``SubagentExecutor`` and emits ``task_*`` custom events the way the
    production task tool does (root-graph ``get_stream_writer``).

    With ``child_emits_error_fallback`` the child stream contains an assistant
    message carrying the ``deerflow_error_fallback`` marker (not as its final
    message, so the delegation itself still completes) — the shape whose leak
    would mark the *parent* run as errored (#4399).
    """
    from langgraph.config import get_stream_writer
    from langgraph.graph import END, START, MessagesState, StateGraph

    from deerflow.subagents.config import SubagentConfig

    child_builder = StateGraph(MessagesState)
    child_builder.add_node(
        "child_model",
        lambda _state: {
            "messages": [
                AIMessage(
                    content="",
                    id="child-ai-sentinel",
                    tool_calls=[{"name": "child_tool", "args": {}, "id": "child-tool-call", "type": "tool_call"}],
                )
            ]
        },
    )
    child_builder.add_node(
        "child_tool",
        lambda _state: {"messages": [ToolMessage(content="child tool output", name="child_tool", tool_call_id="child-tool-call", id="child-tool-sentinel")]},
    )
    child_builder.add_node(
        "child_fallback",
        lambda _state: {
            "messages": [
                AIMessage(
                    content="child provider failed after retries",
                    id="child-fallback-sentinel",
                    additional_kwargs={"deerflow_error_fallback": True},
                )
            ]
        },
    )
    child_builder.add_node(
        "child_final",
        lambda _state: {"messages": [AIMessage(content="child final answer", id="child-final-sentinel")]},
    )
    child_builder.add_edge(START, "child_model")
    child_builder.add_edge("child_model", "child_tool")
    if child_emits_error_fallback:
        child_builder.add_edge("child_tool", "child_fallback")
        child_builder.add_edge("child_fallback", "child_final")
    else:
        child_builder.add_edge("child_tool", "child_final")
    child_builder.add_edge("child_final", END)
    child_graph = child_builder.compile(checkpointer=False)

    executor = executor_module.SubagentExecutor(
        config=SubagentConfig(
            name="general-purpose",
            description="Namespace integration test agent",
            system_prompt="You are a namespace integration test agent.",
            max_turns=5,
            timeout_seconds=30,
        ),
        tools=[],
        parent_model="test-model",
        thread_id=_THREAD_ID,
        trace_id="trace-namespace-integration",
    )

    async def build_initial_state(task):
        return ({"messages": [HumanMessage(content=task, id="child-task-sentinel")]}, [], None)

    monkeypatch.setattr(executor, "_build_initial_state", build_initial_state)
    monkeypatch.setattr(executor, "_create_agent", lambda *_args, **_kwargs: child_graph)

    async def delegate(_state):
        writer = get_stream_writer()
        task_id = executor.execute_async("run the delegated child graph")
        writer({"type": "task_started", "task_id": task_id})
        try:
            deadline = asyncio.get_running_loop().time() + 10
            while True:
                result = executor_module.get_background_task_result(task_id)
                if result is not None and result.status.is_terminal:
                    break
                if asyncio.get_running_loop().time() >= deadline:
                    pytest.fail("delegated subagent did not complete")
                await asyncio.sleep(0.001)
            assert result.status.value == "completed", f"delegation failed: {result.error}"
        finally:
            executor_module.cleanup_background_task(task_id)
        writer({"type": "task_completed", "task_id": task_id})
        return {"messages": [AIMessage(content="parent final answer", id=_PARENT_FINAL_ID)]}

    parent_builder = StateGraph(MessagesState)
    parent_builder.add_node("delegate", delegate)
    parent_builder.add_edge(START, "delegate")
    parent_builder.add_edge("delegate", END)
    return parent_builder.compile()


async def _run_delegation_through_worker(executor_module, monkeypatch, *, stream_subgraphs: bool, child_emits_error_fallback: bool = False) -> tuple[RunRecord, _RecordingStreamBridge]:
    from langgraph.checkpoint.memory import InMemorySaver

    parent_graph = _build_delegating_parent_graph(executor_module, monkeypatch, child_emits_error_fallback=child_emits_error_fallback)
    bridge = _RecordingStreamBridge()
    record = RunRecord(
        run_id=f"run-ns-int-{int(stream_subgraphs)}",
        thread_id=_THREAD_ID,
        assistant_id="lead-agent",
        status=RunStatus.pending,
        on_disconnect=DisconnectMode.cancel,
        model_name=None,
    )
    record.abort_event = asyncio.Event()

    await worker.run_agent(
        bridge,
        _IntegrationRunManager(record),
        record,
        ctx=worker.RunContext(checkpointer=InMemorySaver()),
        agent_factory=lambda config: parent_graph,
        graph_input={"messages": [HumanMessage(content="delegate to the subagent")]},
        config={"configurable": {"thread_id": _THREAD_ID}},
        stream_modes=["values", "messages-tuple", "custom"],
        stream_subgraphs=stream_subgraphs,
    )
    return record, bridge


@pytest.mark.skipif(
    not _LANGGRAPH_INHERITS_SUBGRAPH_NAMESPACE,
    reason="delegated graphs stream as namespaced subgraphs only on LangGraph >= 1.2.6",
)
class TestWorkerSubgraphStreamIntegration:
    @pytest.mark.asyncio
    async def test_stream_subgraphs_publishes_delegated_frames_namespaced_never_bare(self, real_executor_module, monkeypatch):
        record, bridge = await _run_delegation_through_worker(real_executor_module, monkeypatch, stream_subgraphs=True)
        assert record.status == RunStatus.success, f"run failed: {bridge.published}"

        events = bridge.published
        bare_values = [payload for event, payload in events if event == "values"]
        bare_messages = [payload for event, payload in events if event == "messages"]
        namespaced_values = [(event, payload) for event, payload in events if event.startswith("values|")]
        namespaced_messages = [(event, payload) for event, payload in events if event.startswith("messages|")]

        # The #4399 takeover: a delegated values snapshot must never be
        # published as bare "values" (SDK clients replace the thread view).
        for payload in bare_values:
            assert not (_collect_ids(payload) & _CHILD_MESSAGE_IDS), f"delegated messages leaked into a bare values frame: {payload}"
        for payload in bare_messages:
            assert not (_collect_ids(payload) & _CHILD_MESSAGE_IDS), f"delegated message chunk leaked into the bare messages stream: {payload}"

        # The delegated frames must actually arrive — namespaced by LangGraph,
        # not silently dropped (guards against a vacuous pass).
        assert any(_collect_ids(payload) & _CHILD_MESSAGE_IDS for _event, payload in namespaced_values), f"expected namespaced delegated values frames, got events: {[event for event, _ in events]}"
        assert any(_collect_ids(payload) & _CHILD_MESSAGE_IDS for _event, payload in namespaced_messages), f"expected namespaced delegated message chunks, got events: {[event for event, _ in events]}"
        for event, _payload in namespaced_values + namespaced_messages:
            segments = event.split("|")[1:]
            assert segments and all(segments), f"namespaced event name has empty namespace segments: {event}"

        # Root frames stay bare and intact.
        assert any(_PARENT_FINAL_ID in _collect_ids(payload) for payload in bare_values)
        custom_types = [payload.get("type") for event, payload in events if event == "custom" and isinstance(payload, dict)]
        assert "task_started" in custom_types and "task_completed" in custom_types

    @pytest.mark.asyncio
    async def test_delegated_error_fallback_does_not_mark_the_parent_run_as_error(self, real_executor_module, monkeypatch):
        record, bridge = await _run_delegation_through_worker(real_executor_module, monkeypatch, stream_subgraphs=True, child_emits_error_fallback=True)

        # A delegated subagent's LLM error fallback is the executor's to map
        # (task_failed); it must not decide the parent run's status.
        assert record.status == RunStatus.success, "delegated error fallback leaked into the parent run status"
        assert not [payload for event, payload in bridge.published if event == "error"]

        # Non-vacuous: the marked child message really rode the stream —
        # namespaced, where the root-only fallback detector must ignore it.
        namespaced_payloads = [payload for event, payload in bridge.published if event.startswith(("values|", "messages|"))]
        assert any("child-fallback-sentinel" in _collect_ids(payload) for payload in namespaced_payloads)

    @pytest.mark.asyncio
    async def test_without_stream_subgraphs_delegated_frames_stay_out_while_task_events_remain(self, real_executor_module, monkeypatch):
        record, bridge = await _run_delegation_through_worker(real_executor_module, monkeypatch, stream_subgraphs=False)
        assert record.status == RunStatus.success, f"run failed: {bridge.published}"

        events = bridge.published
        # No delegated frame of any mode reaches the parent stream...
        for event, payload in events:
            assert not (_collect_ids(payload) & _CHILD_MESSAGE_IDS), f"delegated messages leaked into event {event!r}: {payload}"
        assert not [event for event, _payload in events if "|" in event]

        # ...while the parent's own frames and the task_* progress contract
        # (what the web frontend relies on instead of the flag) still hold.
        bare_values = [payload for event, payload in events if event == "values"]
        assert any(_PARENT_FINAL_ID in _collect_ids(payload) for payload in bare_values)
        custom_types = [payload.get("type") for event, payload in events if event == "custom" and isinstance(payload, dict)]
        assert "task_started" in custom_types and "task_completed" in custom_types


class TestMessageSeqStamping:
    """A values frame carries the feed seq of messages already persisted.

    The checkpoint has no seq of its own and loses messages to summarization,
    so a client merging it with the seq-ordered feed cannot place a surviving
    old message once the feed's loaded page window no longer reaches back to it
    (#4666). The worker already holds the event store, so it attaches the seq
    that store assigned. Nothing is written back to the checkpoint.
    """

    @staticmethod
    async def _seeded_store():
        from deerflow.runtime.events.store.memory import MemoryRunEventStore

        store = MemoryRunEventStore()
        await store.put(
            thread_id="t1",
            run_id="r1",
            event_type="llm.human.input",
            category="message",
            content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"},
        )
        return store

    @pytest.mark.asyncio
    async def test_root_values_frame_stamps_a_persisted_message(self):
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        bridge = _FakeBridge()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]},
            namespace=(),
            file_tool_chunk_batcher=None,
            subagent_events=_FakeSubagentEvents(),
            seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"),
        )

        _run, _event, payload = bridge.published[0]
        assert payload["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1

    @pytest.mark.asyncio
    async def test_a_message_not_in_the_feed_is_left_unstamped(self):
        """A message still streaming has no seq yet — and needs none: appending
        it at the tail is already its correct position."""
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        bridge = _FakeBridge()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": [{"type": "ai", "id": "not-persisted-yet", "content": "…"}]},
            namespace=(),
            file_tool_chunk_batcher=None,
            subagent_events=_FakeSubagentEvents(),
            seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"),
        )

        _run, _event, payload = bridge.published[0]
        assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {})

    @pytest.mark.asyncio
    async def test_subgraph_frames_are_not_stamped(self):
        """A subagent frame does not belong to the thread feed's ordering."""
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        bridge = _FakeBridge()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]},
            namespace=SUBAGENT_NS,
            file_tool_chunk_batcher=None,
            subagent_events=_FakeSubagentEvents(),
            seq_stamper=_MessageSeqStamper(await self._seeded_store(), "t1"),
        )

        _run, _event, payload = bridge.published[0]
        assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {})

    @pytest.mark.asyncio
    async def test_no_stamper_publishes_the_frame_unchanged(self):
        bridge = _FakeBridge()
        await _publish_stream_item(
            bridge=bridge,
            run_id="run-1",
            mode="values",
            chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]},
            namespace=(),
            file_tool_chunk_batcher=None,
            subagent_events=_FakeSubagentEvents(),
        )

        _run, _event, payload = bridge.published[0]
        assert "deerflow_seq" not in (payload["messages"][0].get("additional_kwargs") or {})

    @pytest.mark.asyncio
    async def test_a_resolved_identity_is_not_looked_up_twice(self):
        """Only a frame carrying messages it has not seen costs a query — in a
        real run that is the compaction frame, not every frame."""
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        store = await self._seeded_store()
        calls: list[list[str]] = []
        original = store.get_message_seqs

        async def counting(thread_id, identities, **kwargs):
            calls.append(list(identities))
            return await original(thread_id, identities, **kwargs)

        store.get_message_seqs = counting  # type: ignore[method-assign]
        stamper = _MessageSeqStamper(store, "t1")
        frame = {"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}

        for _ in range(3):
            await _publish_stream_item(
                bridge=_FakeBridge(),
                run_id="run-1",
                mode="values",
                chunk=dict(frame),
                namespace=(),
                file_tool_chunk_batcher=None,
                subagent_events=_FakeSubagentEvents(),
                seq_stamper=stamper,
            )

        assert len(calls) == 1

    @pytest.mark.asyncio
    async def test_a_miss_is_retried_once_the_feed_advances(self):
        """A miss is provisional: the same message can be persisted later in the run.

        A message this run produces reaches a ``values`` frame before
        ``RunJournal`` flushes it, so its first lookup misses. The journal then
        writes it and it does have a feed seq. Caching that miss permanently
        would leave it unstamped for the rest of the run — and a long run that
        afterwards rolls past the history page and compacts is exactly the
        misplacement this stamper exists to prevent (#4696 review).
        """
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        store = await self._seeded_store()
        generation = 0
        stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: generation)
        frame = {"messages": [{"type": "ai", "id": "a1", "content": "…"}]}

        first = await stamper.stamp(dict(frame))
        assert "deerflow_seq" not in (first["messages"][0].get("additional_kwargs") or {})

        await store.put(
            thread_id="t1",
            run_id="r1",
            event_type="llm.ai.output",
            category="message",
            content={"type": "ai", "id": "a1", "content": "…"},
        )
        generation += 1

        second = await stamper.stamp(dict(frame))
        assert second["messages"][0]["additional_kwargs"]["deerflow_seq"] == 2

    @pytest.mark.asyncio
    async def test_a_miss_is_not_retried_while_the_feed_is_unchanged(self):
        """Retrying is bounded by feed writes, not by frames.

        Without the generation gate the retry would run on every frame that
        carries a streaming message — a query per frame on exactly the long
        threads this stamper is careful to cost one query in.
        """
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        store = await self._seeded_store()
        calls: list[list[str]] = []
        original = store.get_message_seqs

        async def counting(thread_id, identities, **kwargs):
            calls.append(list(identities))
            return await original(thread_id, identities, **kwargs)

        store.get_message_seqs = counting  # type: ignore[method-assign]
        stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: 7)
        frame = {"messages": [{"type": "ai", "id": "never-persisted", "content": "…"}]}

        for _ in range(3):
            await stamper.stamp(dict(frame))

        assert len(calls) == 1

    @pytest.mark.asyncio
    async def test_a_failed_lookup_is_retried_once_the_feed_advances(self):
        """A transient store error must not disable stamping for the whole run.

        The except clause degrades the frame to "no seq"; treating that answer
        as final would make one failed query as permanent as a real miss.
        """
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        store = await self._seeded_store()
        generation = 0
        original = store.get_message_seqs
        failed = False

        async def failing_once(thread_id, identities, **kwargs):
            nonlocal failed
            if not failed:
                failed = True
                raise RuntimeError("transient store failure")
            return await original(thread_id, identities, **kwargs)

        store.get_message_seqs = failing_once  # type: ignore[method-assign]
        stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: generation)
        frame = {"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}

        first = await stamper.stamp(dict(frame))
        assert "deerflow_seq" not in (first["messages"][0].get("additional_kwargs") or {})

        generation += 1
        second = await stamper.stamp(dict(frame))
        assert second["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1

    @pytest.mark.asyncio
    async def test_a_resolved_identity_survives_a_feed_advance(self):
        """Only misses are provisional — a resolved seq is never looked up again.

        The feed's earliest-seq-wins rule makes a resolved answer final, so an
        advancing feed must not turn the positive cache into a per-write query.
        """
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        store = await self._seeded_store()
        calls: list[list[str]] = []
        original = store.get_message_seqs

        async def counting(thread_id, identities, **kwargs):
            calls.append(list(identities))
            return await original(thread_id, identities, **kwargs)

        store.get_message_seqs = counting  # type: ignore[method-assign]
        generation = 0
        stamper = _MessageSeqStamper(store, "t1", feed_generation=lambda: generation)
        frame = {"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]}

        await stamper.stamp(dict(frame))
        generation += 1
        stamped = await stamper.stamp(dict(frame))

        assert len(calls) == 1
        assert stamped["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1

    @pytest.mark.asyncio
    async def test_the_run_stamper_re_asks_after_the_journal_writes(self):
        """The wiring itself: the run's stamper reads the journal's feed writes.

        Miss-retrying is only reachable if the stamper the worker builds is the
        one connected to the journal, and a lambda pointing at the wrong object
        fails silently — the stamper simply keeps every miss.
        """
        from deerflow.runtime.journal import RunJournal
        from deerflow.runtime.runs.worker import _build_seq_stamper

        store = await self._seeded_store()
        journal = RunJournal("r1", "t1", store, flush_threshold=100)
        stamper = _build_seq_stamper(store, "t1", journal)
        frame = {"messages": [{"type": "ai", "id": "a1", "content": "…"}]}

        first = await stamper.stamp(dict(frame))
        assert "deerflow_seq" not in (first["messages"][0].get("additional_kwargs") or {})

        journal._put(
            event_type="llm.ai.response",
            category="message",
            content={"type": "ai", "id": "a1", "content": "…"},
        )
        await journal.flush()

        second = await stamper.stamp(dict(frame))
        assert second["messages"][0]["additional_kwargs"]["deerflow_seq"] == 2

    @pytest.mark.asyncio
    async def test_a_run_without_a_journal_still_builds_a_stamper(self):
        """No writer to report feed growth is not a reason to stop stamping."""
        from deerflow.runtime.runs.worker import _build_seq_stamper

        stamper = _build_seq_stamper(await self._seeded_store(), "t1", None)

        stamped = await stamper.stamp({"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]})
        assert stamped["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1

    @pytest.mark.asyncio
    @pytest.mark.no_auto_user
    async def test_stamping_survives_a_launch_path_without_user_context(self, tmp_path):
        """A launch path that never inherits the auth contextvar (e.g. a
        null-owner scheduled task) must not silently disable stamping.

        The db store's ``user_id=AUTO`` default raises without a user in
        context, and ``stamp``'s except clause would swallow that into a
        per-frame warning — a graceful degrade that turns the fix off in
        exactly the background runs that need it. The stamper therefore
        soft-resolves the id once, when it is built, the same way the
        worker's write paths beside it do (unset → no filter)."""
        from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
        from deerflow.runtime.events.store.db import DbRunEventStore
        from deerflow.runtime.runs.worker import _MessageSeqStamper

        url = f"sqlite+aiosqlite:///{tmp_path / 'seqs.db'}"
        await init_engine("sqlite", url=url, sqlite_dir=str(tmp_path))
        try:
            store = DbRunEventStore(get_session_factory())
            # Written without a user in context, as the worker's own
            # human_message put does on such a path: the row's user_id is NULL.
            await store.put(
                thread_id="t1",
                run_id="r1",
                event_type="llm.human.input",
                category="message",
                content={"type": "human", "id": "u1__user", "content": "MARK-FIRST"},
            )

            bridge = _FakeBridge()
            await _publish_stream_item(
                bridge=bridge,
                run_id="run-1",
                mode="values",
                chunk={"messages": [{"type": "human", "id": "u1__user", "content": "MARK-FIRST"}]},
                namespace=(),
                file_tool_chunk_batcher=None,
                subagent_events=_FakeSubagentEvents(),
                seq_stamper=_MessageSeqStamper(store, "t1"),
            )

            _run, _event, payload = bridge.published[0]
            assert payload["messages"][0]["additional_kwargs"]["deerflow_seq"] == 1
        finally:
            await close_engine()