From 70fb91654daf4d26619173b3dbad618c529357ee Mon Sep 17 00:00:00 2001 From: Aari Date: Thu, 23 Jul 2026 13:57:32 +0800 Subject: [PATCH] fix(gateway): seed branch run-events so inherited history survives forking (#4385) * fix(gateway): seed branch run-events so inherited history survives (#4380) The thread feed (GET /messages, /messages/page) reads the run-event store, but branch creation only wrote checkpoint state - a fresh branch had no message rows, so the parent history vanished from the UI as soon as the branch's first run refreshed the feed. Seed the branch's run_events from the same checkpoint snapshot the branch was created from, mirroring RunJournal's message-event contract (event types, hidden-message rules, original-user-text restoration). Best-effort: a seeding failure degrades to the old behavior and is reported as history_seed_mode=failed. * docs(gateway): correct branch-seed docstring on RunJournal divergences The "consumers cannot tell a seeded row from a journaled one" claim was overstated for AI rows: seeded rows omit run-scoped enrichment (usage / latency_ms / llm_call_index) and stamp caller=lead_agent rather than the message's original caller, neither recoverable from a checkpoint message. Rewrite the docstring to state these divergences explicitly and note they are display-invisible today (no consumer indexes those keys; per-message caller drives no attribution). Also add a code comment marking the hide_from_ui filter as intentionally stricter than the live paths. * fix(gateway): seed dict-shaped checkpoint messages + persist hidden AI/tool rows Two review-driven fixes to build_branch_history_seed_events: 1. Checkpoint messages can arrive as model_dump()-shaped dicts (the branch-matching helpers in threads.py already handle both BaseMessage and dict). The seed only handled BaseMessage, so a dict-backed checkpoint seeded nothing and the branch reported skipped_empty while history existed. Coerce dicts back to BaseMessage via messages_from_dict (faithful: tool_calls / tool_call_id / additional_kwargs survive); unparseable dicts are dropped best-effort. 2. RunJournal.on_llm_end and _persist_tool_result_message persist hide_from_ui AI/tool rows unconditionally (the frontend hides them client-side); the hide check only gates the reconciliation pass. The seed dropped them, so a hidden turn vanished from a forked feed and seeded rows diverged from journaled ones. Match RunJournal and write them, restoring true row-level parity. Adds tests for dict deserialization, the unparseable-dict drop, and the hidden AI/tool persistence contract. --- backend/AGENTS.md | 2 +- backend/app/gateway/routers/threads.py | 30 ++- .../harness/deerflow/runtime/journal.py | 100 +++++++++- backend/tests/test_branch_history_seed.py | 181 ++++++++++++++++++ backend/tests/test_threads_router.py | 105 ++++++++++ 5 files changed, 414 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_branch_history_seed.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index aaad859a7..146871502 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -416,7 +416,7 @@ Localhost persistence deliberately reads the direct request `Host` and ignores ` | **Skills** (`/api/skills`) | `GET /` - list skills; `GET /{name}` - details; `PUT /{name}` - update enabled; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`); `POST /reload` - admin-only process-local prompt-cache invalidation after trusted external filesystem changes | | **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. 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; `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. 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); `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)`) | diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index 2567a526c..42c823775 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -25,7 +25,7 @@ from pydantic import BaseModel, Field, field_validator from sqlalchemy.exc import IntegrityError from app.gateway.authz import require_permission -from app.gateway.deps import get_checkpointer, get_run_manager +from app.gateway.deps import get_checkpointer, get_run_event_store, get_run_manager from app.gateway.internal_auth import get_trusted_internal_owner_user_id from app.gateway.services import ( build_checkpoint_state_accessor, @@ -54,6 +54,7 @@ from deerflow.runtime.goal import ( read_thread_goal, write_thread_goal, ) +from deerflow.runtime.journal import build_branch_history_seed_events from deerflow.runtime.runs.worker import valid_duration_entry from deerflow.runtime.user_context import get_effective_user_id from deerflow.utils.file_io import run_file_io @@ -415,6 +416,9 @@ class ThreadBranchResponse(BaseModel): parent_checkpoint_id: str branched_from_message_id: str workspace_clone_mode: str + # "seeded" | "skipped_empty" | "failed" — whether the parent history was + # copied into the branch's run-event feed (see branch_thread). + history_seed_mode: str # --------------------------------------------------------------------------- @@ -768,6 +772,29 @@ async def branch_thread(thread_id: str, body: ThreadBranchRequest, request: Requ logger.exception("Failed to write branch thread_meta for %s", sanitize_log_param(new_thread_id)) raise HTTPException(status_code=500, detail="Failed to create branch") from None + # The thread feed (GET /messages, /messages/page) reads the run-event + # store, not checkpoints, and a fresh branch has no run_events — so the + # inherited history would vanish from the UI as soon as the branch's + # first run refreshes the feed (#4380 problem 2). Seed the branch's + # run_events from the same checkpoint snapshot the branch was created + # from. Best-effort: on failure the branch stays usable, with history + # visible only through the checkpoint overlay until it is re-branched. + try: + seed_events = build_branch_history_seed_events( + _checkpoint_messages(snapshot), + thread_id=new_thread_id, + run_id=f"branch-seed-{new_thread_id}", + parent_thread_id=thread_id, + ) + if seed_events: + await get_run_event_store(request).put_batch(seed_events) + history_seed_mode = "seeded" + else: + history_seed_mode = "skipped_empty" + except Exception: + logger.exception("Failed to seed branch history run-events for thread %s", sanitize_log_param(new_thread_id)) + history_seed_mode = "failed" + if branch_from_latest_turn: workspace_clone_mode = await _copy_branch_user_data(thread_id, new_thread_id) else: @@ -778,6 +805,7 @@ async def branch_thread(thread_id: str, body: ThreadBranchRequest, request: Requ parent_checkpoint_id=parent_checkpoint_id, branched_from_message_id=body.message_id, workspace_clone_mode=workspace_clone_mode, + history_seed_mode=history_seed_mode, ) diff --git a/backend/packages/harness/deerflow/runtime/journal.py b/backend/packages/harness/deerflow/runtime/journal.py index abe5c87cf..9c0cdb57a 100644 --- a/backend/packages/harness/deerflow/runtime/journal.py +++ b/backend/packages/harness/deerflow/runtime/journal.py @@ -20,13 +20,13 @@ from __future__ import annotations import asyncio import logging import time -from collections.abc import Awaitable, Callable, Mapping +from collections.abc import Awaitable, Callable, Mapping, Sequence from datetime import UTC, datetime from typing import TYPE_CHECKING, Any, cast from uuid import UUID from langchain_core.callbacks import BaseCallbackHandler -from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage, ToolMessage +from langchain_core.messages import AIMessage, AnyMessage, BaseMessage, HumanMessage, ToolMessage, messages_from_dict from langgraph.types import Command from deerflow.agents.human_input import read_human_input_response @@ -53,6 +53,102 @@ def _should_persist_human_input_message(message: BaseMessage) -> bool: return response is not None and response["source"] in _PERSISTED_HIDDEN_HUMAN_INPUT_RESPONSE_SOURCES +def _coerce_seed_message(message: Any) -> Any: + """Return ``message`` as a ``BaseMessage``, deserializing dict form if needed. + + ``_checkpoint_messages`` (threads.py) returns whatever the snapshot holds, + and its sibling branch-matching helpers all handle a message being either a + ``BaseMessage`` or a ``model_dump()``-shaped dict (serde differences across + checkpoint backends/modes). The seed path must handle both too — otherwise a + dict-backed checkpoint seeds nothing and the branch silently reports + ``skipped_empty`` while history exists. Unparseable dicts fall through + unchanged and are dropped by the ``isinstance(BaseMessage)`` guard. + """ + if isinstance(message, BaseMessage): + return message + if isinstance(message, Mapping): + msg_type = message.get("type") + if isinstance(msg_type, str) and msg_type: + try: + return messages_from_dict([{"type": msg_type, "data": dict(message)}])[0] + except Exception: + logger.warning("branch seed: could not deserialize checkpoint message dict (type=%s)", msg_type) + return message + + +def build_branch_history_seed_events( + messages: Sequence[Any], + *, + thread_id: str, + run_id: str, + parent_thread_id: str, +) -> list[dict]: + """Serialize a branch checkpoint's messages into run-event message rows. + + Thread branching copies checkpoint state, but the thread feed + (``list_messages`` / ``GET /threads/{id}/messages/page``) reads the + run-event store — which a fresh branch has no rows in, so the inherited + history vanishes from the UI as soon as the branch's first run refreshes + the feed (#4380). Seeding the branch's run_events from the same + checkpoint snapshot the branch was created from keeps the feed + consistent with what the branch actually contains. + + Mirrors RunJournal's message-event contract so seeded rows are + indistinguishable from journaled ones except by the ``branch_seed`` + marker: same event types, ``category="message"``, ``content= + message.model_dump()``, the human-input persistence rule + (``_should_persist_human_input_message``), the original-user-text + restoration, and the same treatment of ``hide_from_ui`` AI/tool rows — + RunJournal persists them (``on_llm_end`` / ``_persist_tool_result_message`` + do not filter) and the frontend hides them client-side, so the seed writes + them too rather than dropping them. + + The one deliberate divergence, because a checkpoint message carries no run + scope: AI rows omit RunJournal's run-scoped enrichment (``usage`` / + ``latency_ms`` / ``llm_call_index``), and ``caller`` is stamped + ``lead_agent`` rather than the message's original caller (unrecoverable + here). Neither is observable today — no consumer indexes those metadata + keys, and per-message ``caller`` drives no attribution (the ``by_caller`` + usage panel is run-scoped, not fed from the message feed). + """ + events: list[dict] = [] + created_at = datetime.now(UTC).isoformat() + seed_metadata = {"branch_seed": True, "branch_parent_thread_id": parent_thread_id} + for raw_message in messages: + message = _coerce_seed_message(raw_message) + if not isinstance(message, BaseMessage): + continue + if isinstance(message, HumanMessage): + if not _should_persist_human_input_message(message): + continue + event_type = "llm.human.input" + content = restore_original_human_message(message).model_dump() + metadata: dict[str, Any] = {"caller": "lead_agent", **seed_metadata} + elif isinstance(message, AIMessage): + event_type = "llm.ai.response" + content = message.model_dump() + metadata = {"caller": "lead_agent", **seed_metadata} + elif isinstance(message, ToolMessage): + event_type = "llm.tool.result" + content = message.model_dump() + metadata = dict(seed_metadata) + else: + # System / remove / summary artifacts never enter the thread feed. + continue + events.append( + { + "thread_id": thread_id, + "run_id": run_id, + "event_type": event_type, + "category": "message", + "content": content, + "metadata": metadata, + "created_at": created_at, + } + ) + return events + + class RunJournal(BaseCallbackHandler): """LangChain callback handler that captures events to RunEventStore.""" diff --git a/backend/tests/test_branch_history_seed.py b/backend/tests/test_branch_history_seed.py new file mode 100644 index 000000000..4d391910d --- /dev/null +++ b/backend/tests/test_branch_history_seed.py @@ -0,0 +1,181 @@ +"""Unit tests for branch-history run-event seeding (#4380 problem 2). + +``build_branch_history_seed_events`` must mirror RunJournal's message-event +contract exactly: same event types, ``category="message"``, +``content=message.model_dump()``, the same hidden-message rules, and the same +original-user-text restoration — so the thread feed cannot tell a seeded row +from a journaled one. +""" + +import asyncio + +from langchain_core.messages import AIMessage, HumanMessage, SystemMessage, ToolMessage + +from deerflow.runtime.events.store.memory import MemoryRunEventStore +from deerflow.runtime.journal import build_branch_history_seed_events + + +def _seed(messages): + return build_branch_history_seed_events( + messages, + thread_id="branch-thread", + run_id="branch-seed-branch-thread", + parent_thread_id="parent-thread", + ) + + +def test_seed_serializes_visible_history_in_order() -> None: + events = _seed( + [ + HumanMessage(id="h1", content="question"), + AIMessage(id="a1", content="answer"), + ToolMessage(id="t1", content="tool output", tool_call_id="call-1"), + ] + ) + + assert [event["event_type"] for event in events] == ["llm.human.input", "llm.ai.response", "llm.tool.result"] + assert all(event["category"] == "message" for event in events) + assert all(event["thread_id"] == "branch-thread" for event in events) + assert all(event["run_id"] == "branch-seed-branch-thread" for event in events) + assert [event["content"]["id"] for event in events] == ["h1", "a1", "t1"] + # Human/AI rows carry the journal's caller tag; every row carries the + # seed provenance marker. + assert events[0]["metadata"]["caller"] == "lead_agent" + assert events[1]["metadata"]["caller"] == "lead_agent" + assert "caller" not in events[2]["metadata"] + assert all(event["metadata"]["branch_seed"] is True for event in events) + assert all(event["metadata"]["branch_parent_thread_id"] == "parent-thread" for event in events) + + +def test_seed_skips_system_summary_and_nonmessage() -> None: + """System prompts, summary-named/hidden human turns, and non-messages never + enter the thread feed — same as RunJournal's message path.""" + events = _seed( + [ + SystemMessage(id="s1", content="system prompt"), + HumanMessage(id="h-hidden", content="internal", additional_kwargs={"hide_from_ui": True}), + HumanMessage(id="h-summary", content="compacted", name="summary"), + HumanMessage(id="h1", content="question"), + AIMessage(id="a1", content="answer"), + "not-a-message", + ] + ) + + assert [event["content"]["id"] for event in events] == ["h1", "a1"] + + +def test_seed_persists_hidden_ai_and_tool_rows_like_runjournal() -> None: + """RunJournal's on_llm_end / _persist_tool_result_message write hide_from_ui + AI and tool rows unconditionally (the frontend hides them client-side), so + the seed must too — otherwise seeded rows diverge from journaled ones and a + hidden turn disappears from a forked feed.""" + events = _seed( + [ + HumanMessage(id="h1", content="question"), + AIMessage(id="a-hidden", content="internal", additional_kwargs={"hide_from_ui": True}), + AIMessage(id="a1", content="answer"), + ToolMessage(id="t-hidden", content="internal", tool_call_id="call-h", additional_kwargs={"hide_from_ui": True}), + ] + ) + + assert [event["content"]["id"] for event in events] == ["h1", "a-hidden", "a1", "t-hidden"] + assert [event["event_type"] for event in events] == [ + "llm.human.input", + "llm.ai.response", + "llm.ai.response", + "llm.tool.result", + ] + # The hidden marker is preserved in content so the frontend still hides them. + assert events[1]["content"]["additional_kwargs"]["hide_from_ui"] is True + assert events[3]["content"]["additional_kwargs"]["hide_from_ui"] is True + + +def test_seed_deserializes_dict_shaped_checkpoint_messages() -> None: + """Checkpoint messages can arrive as model_dump()-shaped dicts (the + branch-matching helpers in threads.py already handle both); the seed must + deserialize them instead of silently producing an empty batch.""" + dict_messages = [ + HumanMessage(id="h1", content="question").model_dump(), + AIMessage( + id="a1", + content="answer", + tool_calls=[{"name": "search", "args": {"q": "x"}, "id": "call-1", "type": "tool_call"}], + ).model_dump(), + ToolMessage(id="t1", content="tool output", tool_call_id="call-1").model_dump(), + ] + + events = _seed(dict_messages) + + assert [event["event_type"] for event in events] == ["llm.human.input", "llm.ai.response", "llm.tool.result"] + assert [event["content"]["id"] for event in events] == ["h1", "a1", "t1"] + # Faithful reconstruction: AI tool_calls and the tool's tool_call_id survive. + assert events[1]["content"]["tool_calls"][0]["id"] == "call-1" + assert events[2]["content"]["tool_call_id"] == "call-1" + + +def test_seed_drops_unparseable_dict_message() -> None: + """A dict with no usable ``type`` can't be reconstructed; it is dropped + rather than crashing the seed (best-effort — the branch stays usable).""" + events = _seed([{"content": "orphan", "no_type": True}, HumanMessage(id="h1", content="q")]) + + assert [event["content"]["id"] for event in events] == ["h1"] + + +def test_seed_persists_allowlisted_hidden_human_input_response() -> None: + """Mirrors RunJournal: answered clarification replies stay recoverable.""" + response = { + "version": 1, + "kind": "human_input_response", + "source": "ask_clarification", + "request_id": "req-1", + "value": "yes", + "response_kind": "text", + } + events = _seed( + [ + HumanMessage( + id="h-response", + content="yes", + additional_kwargs={"hide_from_ui": True, "human_input_response": response}, + ) + ] + ) + + assert [event["content"]["id"] for event in events] == ["h-response"] + assert events[0]["event_type"] == "llm.human.input" + + +def test_seed_restores_original_user_content() -> None: + """The model-facing wrapper must not leak into the seeded feed.""" + message = HumanMessage( + id="h1", + content="question with injected context", + additional_kwargs={"original_user_content": "question"}, + ) + + events = _seed([message]) + + assert events[0]["content"]["content"] == "question" + assert "original_user_content" not in events[0]["content"]["additional_kwargs"] + + +def test_seed_roundtrips_through_memory_store_feed() -> None: + """put_batch preserves order and list_messages returns the seeded feed.""" + store = MemoryRunEventStore() + events = _seed( + [ + HumanMessage(id="h1", content="question"), + AIMessage(id="a1", content="answer"), + ] + ) + + async def roundtrip(): + await store.put_batch(events) + return await store.list_messages("branch-thread", user_id=None) + + rows = asyncio.run(roundtrip()) + + assert [row["content"]["id"] for row in rows] == ["h1", "a1"] + seqs = [row["seq"] for row in rows] + assert seqs == sorted(seqs) + assert all(row["category"] == "message" for row in rows) diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index e41395edd..e83501424 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -1465,6 +1465,111 @@ def test_state_endpoints_preserve_extension_reducer_channels(monkeypatch, mode) assert [message.id for message in branch_values["messages"]] == ["h1", "a1"] +async def _seed_branch_history_source(checkpointer, custom_factory, mode, source_thread_id): + """Seed a completed turn whose history includes hidden and tool messages.""" + 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"), + HumanMessage(id="h-hidden", content="internal", additional_kwargs={"hide_from_ui": True}), + ] + }, + as_node="model", + ) + await accessor.aupdate( + config, + { + "messages": [ + ToolMessage(id="t1", content="tool output", tool_call_id="call-1"), + AIMessage(id="a1", content="answer"), + ] + }, + as_node="model", + ) + + +@pytest.mark.parametrize("mode", ["full", "delta"]) +def test_branch_seeds_run_events_with_parent_history(monkeypatch, mode) -> None: + """Branching must seed the branch's run-event feed with the parent history. + + The thread feed (``GET /messages`` / ``/messages/page``) reads the + run-event store, not checkpoints; without seeding, a fresh branch has no + message rows, so the inherited history vanishes from the UI as soon as + the branch's first run refreshes the feed (#4380 problem 2). + """ + from deerflow.runtime.events.store.memory import MemoryRunEventStore + + app, _store, checkpointer = _build_thread_app() + custom_factory = _wire_extension_agent(monkeypatch, app, checkpointer, mode) + event_store = MemoryRunEventStore() + app.state.run_event_store = event_store + source_thread_id = f"branch-history-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_branch_history_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"] + + rows = asyncio.run(event_store.list_messages(branch_thread_id, user_id=None)) + + # The visible parent history is seeded in order; hidden messages are not. + assert [row["content"]["id"] for row in rows] == ["h1", "t1", "a1"] + assert [row["event_type"] for row in rows] == ["llm.human.input", "llm.tool.result", "llm.ai.response"] + assert all(row["category"] == "message" for row in rows) + assert all(row["run_id"] == f"branch-seed-{branch_thread_id}" for row in rows) + assert all((row.get("metadata") or {}).get("branch_seed") is True for row in rows) + seqs = [row["seq"] for row in rows] + assert seqs == sorted(seqs) + assert branch_response.json()["history_seed_mode"] == "seeded" + + # The parent thread's feed stays untouched. + assert asyncio.run(event_store.list_messages(source_thread_id, user_id=None)) == [] + + +def test_branch_history_seed_failure_keeps_branch_usable(monkeypatch) -> None: + """A seeding failure must degrade, not fail the branch (best-effort).""" + + class _ExplodingStore: + async def put_batch(self, events): + raise RuntimeError("event store down") + + app, _store, checkpointer = _build_thread_app() + custom_factory = _wire_extension_agent(monkeypatch, app, checkpointer, "full") + app.state.run_event_store = _ExplodingStore() + source_thread_id = "branch-history-source-failure" + + 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_branch_history_source(checkpointer, custom_factory, "full", 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 + assert branch_response.json()["history_seed_mode"] == "failed" + + 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()