github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-25 07:28:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(gateway): seed branch run-events so inherited history survives forking (#4385)

Browse Source 
* 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.
This commit is contained in:
Aari 2026-07-23 13:57:32 +08:00 committed by GitHub
parent a38b1daec3
commit 70fb91654d
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
5 changed files with 414 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
backend/AGENTS.md
View File

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

30
backend/app/gateway/routers/threads.py
View File

@ -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,
)

100
backend/packages/harness/deerflow/runtime/journal.py
View File

@ -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."""

181
backend/tests/test_branch_history_seed.py Normal file
View File

@ -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="<wrapped>question with injected context</wrapped>",
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)

105
backend/tests/test_threads_router.py
View File

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