mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix: honor checkpoint object in thread state updates (#7115)
* fix: honor checkpoint object in thread state updates Prefer the checkpoint object over the deprecated checkpoint_id, validate its thread_id, and preserve checkpoint namespace when updating state. Add regression coverage for checkpoint-object forks and thread mismatch. * fix: preserve client context handling in checkpoint PR * fix: support state updates in checkpoint namespaces * fix: reject unknown checkpoint namespaces * fix(gateway): isolate subgraph titles from thread metadata Only synchronize state titles into conversation metadata for the root checkpoint namespace. Cover persisted child updates with and without checkpoint IDs in full and delta modes, and retain root title synchronization and branch sequence cleanup. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Gusanity
GitHub
parent
75df1be6b7
commit
d318a89b80
@ -2255,6 +2255,8 @@ information.
|
||||
|
||||
Interrupted first-turn runs still persist a fallback conversation title, so stopping a streaming response does not leave the thread as "Untitled" after refresh.
|
||||
|
||||
Updating a subgraph checkpoint's title through the state API keeps the root conversation's title unchanged.
|
||||
|
||||
Streaming Markdown responses animate only newly arrived words; text that is already visible is not faded out and replayed when the next chunk extends the same block.
|
||||
|
||||
In the Web UI, completed assistant turns can be branched into a new main conversation. Automatically inherited branch titles use the next free language-neutral numeric suffix (`Title (2)`, then `Title (3)` for another sibling or a branch of the numbered conversation) so generated sibling titles remain distinct without persisting a locale-specific label; matching explicit or renamed sibling titles also reserve their displayed suffix even though they carry no generated sequence metadata. An explicit API title is preserved. Renaming a branch clears its generated sequence, so its next automatic branch starts from the renamed title at `(2)`. Recent chats also groups loaded branches directly beneath a loaded parent with subtle tree connectors. Missing parents, malformed or cyclic lineage, and branches in a different pinned state stay visible at the top level instead of being hidden or moved across the pinned boundary. The new thread starts from that turn's checkpoint and keeps the preceding replay checkpoint, so the branched response can be regenerated immediately. The latest response can also be regenerated after an interruption, even when its streamed partial text never reached a checkpoint. Stopping mid-answer retains the assistant text that was already streamed in the thread's message feed, so refreshing no longer discards visible progress; that partial text remains outside authoritative completed checkpoint state. Regenerating the latest response preserves the thread's current title, including a title you renamed manually after the original response. Legacy or imported histories without checkpoint parent links use a bounded chronological fallback; if no earlier replay checkpoint exists, branching still succeeds with the legacy single-checkpoint shape, while regeneration remains unavailable for that inherited response. Existing single-checkpoint branches are left unchanged rather than attempting an unsafe checkpoint copy. Because workspace files are not checkpointed, the branch only receives a best-effort copy of the current workspace when you branch from the latest turn; branching from an older turn keeps just the restored message history so the branch never inherits files that were created in a later part of the conversation.
|
||||
|
||||
@ -66,3 +66,8 @@ external filesystem writers do not participate in this history guarantee.
|
||||
Plugin actions stamp `ActionContext.interaction=session` only from authenticated
|
||||
session provenance. Never accept it from payloads or propagate it to model tools;
|
||||
extensions use it to distinguish owner review from model/tool output.
|
||||
|
||||
State updates synchronize titles into thread metadata only for the root checkpoint
|
||||
namespace. Nonroot writes preserve the conversation title, branch-title sequence
|
||||
and activity timestamp. `tests/test_threads_router.py` checks metadata/search
|
||||
isolation and root title synchronization in full and delta modes.
|
||||
|
||||
@ -1635,33 +1635,49 @@ async def update_thread_state(thread_id: ThreadId, body: ThreadStateUpdateReques
|
||||
thread_store = get_thread_store(request)
|
||||
# Validate external roles before materializing a graph or reserving a write.
|
||||
values = strip_server_owned_state_metadata(dict(body.values or {}))
|
||||
if body.checkpoint_id is not None:
|
||||
if not body.checkpoint_id:
|
||||
checkpoint_id = body.checkpoint_id
|
||||
checkpoint_ns = ""
|
||||
if body.checkpoint is not None:
|
||||
checkpoint_thread_id = body.checkpoint.get("thread_id")
|
||||
if checkpoint_thread_id is not None and str(checkpoint_thread_id) != thread_id:
|
||||
raise HTTPException(status_code=400, detail="checkpoint thread_id does not match request thread_id")
|
||||
raw_checkpoint_id = body.checkpoint.get("checkpoint_id")
|
||||
checkpoint_id = str(raw_checkpoint_id) if raw_checkpoint_id else raw_checkpoint_id
|
||||
raw_checkpoint_ns = body.checkpoint.get("checkpoint_ns")
|
||||
if raw_checkpoint_ns is not None:
|
||||
checkpoint_ns = str(raw_checkpoint_ns)
|
||||
|
||||
if checkpoint_id is not None or checkpoint_ns:
|
||||
if checkpoint_id is not None and not checkpoint_id:
|
||||
raise HTTPException(status_code=404, detail="Checkpoint not found")
|
||||
selected_config = {
|
||||
"configurable": {
|
||||
"thread_id": thread_id,
|
||||
"checkpoint_ns": "",
|
||||
"checkpoint_id": body.checkpoint_id,
|
||||
"checkpoint_ns": checkpoint_ns,
|
||||
}
|
||||
}
|
||||
if checkpoint_id is not None:
|
||||
selected_config["configurable"]["checkpoint_id"] = checkpoint_id
|
||||
try:
|
||||
checkpoint_tuple = await get_checkpointer(request).aget_tuple(selected_config)
|
||||
except Exception:
|
||||
logger.exception("Failed to get state for thread %s", sanitize_log_param(thread_id))
|
||||
raise HTTPException(status_code=500, detail="Failed to get thread state")
|
||||
if checkpoint_tuple is None:
|
||||
raise HTTPException(status_code=404, detail=f"Checkpoint {body.checkpoint_id} not found")
|
||||
detail = f"Checkpoint {checkpoint_id} not found" if checkpoint_id is not None else "Checkpoint not found"
|
||||
raise HTTPException(status_code=404, detail=detail)
|
||||
|
||||
mutation_node = body.as_node or "manual_state_update"
|
||||
# Resolve through the shared boundary (thread metadata -> assistant_id ->
|
||||
# effective schema) so extension middleware channels stay writable.
|
||||
accessor, read_config = await build_thread_checkpoint_state_mutation_accessor(
|
||||
request,
|
||||
thread_id=thread_id,
|
||||
as_node=mutation_node,
|
||||
checkpoint_id=body.checkpoint_id,
|
||||
)
|
||||
accessor_kwargs: dict[str, Any] = {
|
||||
"thread_id": thread_id,
|
||||
"as_node": mutation_node,
|
||||
"checkpoint_id": checkpoint_id,
|
||||
}
|
||||
if checkpoint_ns:
|
||||
accessor_kwargs["checkpoint_ns"] = checkpoint_ns
|
||||
accessor, read_config = await build_thread_checkpoint_state_mutation_accessor(request, **accessor_kwargs)
|
||||
writable_channels = graph_writable_channels(getattr(accessor, "graph", None))
|
||||
if writable_channels is not None:
|
||||
unknown_fields = sorted(set(values) - writable_channels)
|
||||
@ -1699,7 +1715,7 @@ async def update_thread_state(thread_id: ThreadId, body: ThreadStateUpdateReques
|
||||
logger.exception("Failed to update state for thread %s", sanitize_log_param(thread_id))
|
||||
raise HTTPException(status_code=500, detail="Failed to update thread state")
|
||||
|
||||
if thread_store and body.values and "title" in body.values:
|
||||
if not checkpoint_ns and thread_store and body.values and "title" in body.values:
|
||||
new_title = body.values["title"]
|
||||
if new_title:
|
||||
try:
|
||||
|
||||
@ -1393,6 +1393,7 @@ def build_checkpoint_state_mutation_accessor(
|
||||
thread_id: str,
|
||||
as_node: str,
|
||||
checkpoint_id: str | None = None,
|
||||
checkpoint_ns: str = "",
|
||||
state_schema: Any | None = None,
|
||||
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
||||
"""Build a state-only graph whose writer node finishes immediately.
|
||||
@ -1406,7 +1407,7 @@ def build_checkpoint_state_mutation_accessor(
|
||||
config: dict[str, Any] = {
|
||||
"configurable": {
|
||||
"thread_id": thread_id,
|
||||
"checkpoint_ns": "",
|
||||
"checkpoint_ns": checkpoint_ns,
|
||||
}
|
||||
}
|
||||
if checkpoint_id is not None:
|
||||
@ -1419,6 +1420,7 @@ def build_checkpoint_state_mutation_accessor(
|
||||
get_checkpointer(request),
|
||||
store=getattr(request.app.state, "store", None),
|
||||
mode=mode,
|
||||
direct_namespace=True,
|
||||
)
|
||||
return accessor, config
|
||||
|
||||
@ -1583,12 +1585,13 @@ def build_checkpoint_state_accessor(
|
||||
thread_id: str,
|
||||
assistant_id: str | None = None,
|
||||
checkpoint_id: str | None = None,
|
||||
checkpoint_ns: str = "",
|
||||
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
||||
"""Build the mode-selected lead graph used for materialized checkpoint state."""
|
||||
ctx = get_run_context(request)
|
||||
config = build_run_config(thread_id, None, None, assistant_id=assistant_id)
|
||||
configurable = config.setdefault("configurable", {})
|
||||
configurable["checkpoint_ns"] = ""
|
||||
configurable["checkpoint_ns"] = checkpoint_ns
|
||||
if checkpoint_id is not None:
|
||||
configurable["checkpoint_id"] = checkpoint_id
|
||||
|
||||
@ -1634,6 +1637,7 @@ async def abuild_checkpoint_state_accessor(
|
||||
thread_id: str,
|
||||
assistant_id: str | None = None,
|
||||
checkpoint_id: str | None = None,
|
||||
checkpoint_ns: str = "",
|
||||
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
||||
"""Async variant of :func:`build_checkpoint_state_accessor`.
|
||||
|
||||
@ -1647,13 +1651,14 @@ async def abuild_checkpoint_state_accessor(
|
||||
identity changed while it waited rebuilds instead of reusing the
|
||||
winner's graph.
|
||||
"""
|
||||
return await run_assembly(
|
||||
build_checkpoint_state_accessor,
|
||||
request,
|
||||
thread_id=thread_id,
|
||||
assistant_id=assistant_id,
|
||||
checkpoint_id=checkpoint_id,
|
||||
)
|
||||
kwargs: dict[str, Any] = {
|
||||
"thread_id": thread_id,
|
||||
"assistant_id": assistant_id,
|
||||
"checkpoint_id": checkpoint_id,
|
||||
}
|
||||
if checkpoint_ns:
|
||||
kwargs["checkpoint_ns"] = checkpoint_ns
|
||||
return await run_assembly(build_checkpoint_state_accessor, request, **kwargs)
|
||||
|
||||
|
||||
async def resolve_thread_assistant_id(
|
||||
@ -1686,6 +1691,7 @@ async def build_thread_checkpoint_state_accessor(
|
||||
*,
|
||||
thread_id: str,
|
||||
checkpoint_id: str | None = None,
|
||||
checkpoint_ns: str = "",
|
||||
fail_closed: bool = False,
|
||||
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
||||
"""Single resolution boundary for state endpoints.
|
||||
@ -1695,12 +1701,14 @@ async def build_thread_checkpoint_state_accessor(
|
||||
``AgentMiddleware.state_schema`` from the response.
|
||||
"""
|
||||
assistant_id = await resolve_thread_assistant_id(request, thread_id, fail_closed=fail_closed)
|
||||
return await abuild_checkpoint_state_accessor(
|
||||
request,
|
||||
thread_id=thread_id,
|
||||
assistant_id=assistant_id,
|
||||
checkpoint_id=checkpoint_id,
|
||||
)
|
||||
kwargs: dict[str, Any] = {
|
||||
"thread_id": thread_id,
|
||||
"assistant_id": assistant_id,
|
||||
"checkpoint_id": checkpoint_id,
|
||||
}
|
||||
if checkpoint_ns:
|
||||
kwargs["checkpoint_ns"] = checkpoint_ns
|
||||
return await abuild_checkpoint_state_accessor(request, **kwargs)
|
||||
|
||||
|
||||
async def build_thread_checkpoint_state_mutation_accessor(
|
||||
@ -1709,6 +1717,7 @@ async def build_thread_checkpoint_state_mutation_accessor(
|
||||
thread_id: str,
|
||||
as_node: str,
|
||||
checkpoint_id: str | None = None,
|
||||
checkpoint_ns: str = "",
|
||||
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
||||
"""Mutation accessor compiled with the thread's effective state schema.
|
||||
|
||||
@ -1716,20 +1725,24 @@ async def build_thread_checkpoint_state_mutation_accessor(
|
||||
so writes carrying materialized state do not silently discard
|
||||
extension-owned channels.
|
||||
"""
|
||||
read_accessor, _read_config = await build_thread_checkpoint_state_accessor(
|
||||
request,
|
||||
thread_id=thread_id,
|
||||
checkpoint_id=checkpoint_id,
|
||||
fail_closed=True,
|
||||
)
|
||||
read_kwargs: dict[str, Any] = {
|
||||
"thread_id": thread_id,
|
||||
"checkpoint_id": checkpoint_id,
|
||||
"fail_closed": True,
|
||||
}
|
||||
if checkpoint_ns:
|
||||
read_kwargs["checkpoint_ns"] = checkpoint_ns
|
||||
read_accessor, _read_config = await build_thread_checkpoint_state_accessor(request, **read_kwargs)
|
||||
state_schema = graph_state_schema(getattr(read_accessor, "graph", None))
|
||||
return build_checkpoint_state_mutation_accessor(
|
||||
request,
|
||||
thread_id=thread_id,
|
||||
as_node=as_node,
|
||||
checkpoint_id=checkpoint_id,
|
||||
state_schema=state_schema,
|
||||
)
|
||||
mutation_kwargs: dict[str, Any] = {
|
||||
"thread_id": thread_id,
|
||||
"as_node": as_node,
|
||||
"checkpoint_id": checkpoint_id,
|
||||
"state_schema": state_schema,
|
||||
}
|
||||
if checkpoint_ns:
|
||||
mutation_kwargs["checkpoint_ns"] = checkpoint_ns
|
||||
return build_checkpoint_state_mutation_accessor(request, **mutation_kwargs)
|
||||
|
||||
|
||||
async def apply_checkpoint_to_run_config(
|
||||
|
||||
@ -114,6 +114,7 @@ class CheckpointStateAccessor:
|
||||
graph: Any
|
||||
checkpointer: Any
|
||||
mode: CheckpointChannelMode
|
||||
direct_namespace: bool = False
|
||||
|
||||
@classmethod
|
||||
def bind(
|
||||
@ -123,11 +124,12 @@ class CheckpointStateAccessor:
|
||||
*,
|
||||
store: Any | None = None,
|
||||
mode: CheckpointChannelMode = "full",
|
||||
direct_namespace: bool = False,
|
||||
) -> CheckpointStateAccessor:
|
||||
graph.checkpointer = checkpointer
|
||||
if store is not None:
|
||||
graph.store = store
|
||||
return cls(graph=graph, checkpointer=checkpointer, mode=mode)
|
||||
return cls(graph=graph, checkpointer=checkpointer, mode=mode, direct_namespace=direct_namespace)
|
||||
|
||||
def _prepare_config(self, config: dict[str, Any]) -> dict[str, Any]:
|
||||
prepared = {
|
||||
@ -135,6 +137,15 @@ class CheckpointStateAccessor:
|
||||
"configurable": dict(config.get("configurable", {})),
|
||||
"metadata": dict(config.get("metadata", {})),
|
||||
}
|
||||
if self.direct_namespace and prepared["configurable"].get("checkpoint_ns"):
|
||||
# A state-only mutation graph has no subgraphs to delegate to. Bind
|
||||
# the saver explicitly so LangGraph reads/writes this namespace on
|
||||
# the graph itself, including when reading back an updated config.
|
||||
# This private key is verified against LangGraph 1.2.9; recheck it
|
||||
# when upgrading LangGraph because its internal location may move.
|
||||
from langgraph._internal._constants import CONFIG_KEY_CHECKPOINTER
|
||||
|
||||
prepared["configurable"][CONFIG_KEY_CHECKPOINTER] = self.checkpointer
|
||||
inject_checkpoint_mode(prepared, self.mode)
|
||||
return prepared
|
||||
|
||||
|
||||
@ -4396,6 +4396,233 @@ def test_update_thread_state_real_mutation_graph_finishes_without_scheduling(mon
|
||||
assert body["next"] == []
|
||||
|
||||
|
||||
def test_update_thread_state_forks_from_checkpoint_object(monkeypatch) -> None:
|
||||
app, _store, _checkpointer = _build_thread_app()
|
||||
thread_id = "state-checkpoint-object"
|
||||
real_mutation_builder = gateway_services.build_checkpoint_state_mutation_accessor
|
||||
|
||||
async def mutation_boundary(
|
||||
request,
|
||||
*,
|
||||
thread_id,
|
||||
as_node,
|
||||
checkpoint_id=None,
|
||||
checkpoint_ns="",
|
||||
):
|
||||
return real_mutation_builder(
|
||||
request,
|
||||
thread_id=thread_id,
|
||||
as_node=as_node,
|
||||
checkpoint_id=checkpoint_id,
|
||||
checkpoint_ns=checkpoint_ns,
|
||||
)
|
||||
|
||||
monkeypatch.setattr(
|
||||
threads,
|
||||
"build_thread_checkpoint_state_mutation_accessor",
|
||||
mutation_boundary,
|
||||
)
|
||||
|
||||
with TestClient(app) as client:
|
||||
created = client.post(
|
||||
"/api/threads",
|
||||
json={"thread_id": thread_id, "metadata": {}},
|
||||
)
|
||||
assert created.status_code == 200, created.text
|
||||
|
||||
first = client.post(
|
||||
f"/api/threads/{thread_id}/state",
|
||||
json={"values": {"title": "First"}},
|
||||
)
|
||||
assert first.status_code == 200, first.text
|
||||
first_checkpoint_id = first.json()["checkpoint_id"]
|
||||
|
||||
second = client.post(
|
||||
f"/api/threads/{thread_id}/state",
|
||||
json={"values": {"title": "Second"}},
|
||||
)
|
||||
assert second.status_code == 200, second.text
|
||||
second_checkpoint_id = second.json()["checkpoint_id"]
|
||||
|
||||
forked = client.post(
|
||||
f"/api/threads/{thread_id}/state",
|
||||
json={
|
||||
"checkpoint_id": second_checkpoint_id,
|
||||
"checkpoint": {
|
||||
"thread_id": thread_id,
|
||||
"checkpoint_ns": "",
|
||||
"checkpoint_id": first_checkpoint_id,
|
||||
},
|
||||
"values": {"title": "Forked from first"},
|
||||
},
|
||||
)
|
||||
|
||||
assert forked.status_code == 200, forked.text
|
||||
assert forked.json()["parent_checkpoint_id"] == first_checkpoint_id
|
||||
assert forked.json()["values"]["title"] == "Forked from first"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
@pytest.mark.parametrize("include_checkpoint_id", [True, False])
|
||||
def test_update_thread_state_in_persisted_subgraph_namespace(monkeypatch, mode, include_checkpoint_id) -> None:
|
||||
from langgraph.graph import StateGraph
|
||||
|
||||
from deerflow.agents.thread_state import get_thread_state_schema
|
||||
|
||||
app, _store, checkpointer = _build_thread_app()
|
||||
app.state.checkpoint_channel_mode = mode
|
||||
thread_id = f"state-subgraph-{mode}"
|
||||
with TestClient(app) as client:
|
||||
assert client.post("/api/threads", json={"thread_id": thread_id, "metadata": {"branch_title_sequence": 2}}).status_code == 200
|
||||
schema = get_thread_state_schema(mode)
|
||||
child = StateGraph(schema)
|
||||
child.add_node("write_child", lambda state: {"title": "Child original"})
|
||||
child.set_entry_point("write_child")
|
||||
child.set_finish_point("write_child")
|
||||
parent = StateGraph(schema)
|
||||
parent.add_node("child", child.compile())
|
||||
parent.set_entry_point("child")
|
||||
parent.add_node("restore_root_title", lambda state: {"title": "Root original"})
|
||||
parent.add_edge("child", "restore_root_title")
|
||||
parent.set_finish_point("restore_root_title")
|
||||
graph = parent.compile(checkpointer=checkpointer)
|
||||
asyncio.run(graph.ainvoke({"title": "Root original"}, {"configurable": {"thread_id": thread_id}}))
|
||||
|
||||
namespaces = [ns for ns in checkpointer.storage[thread_id] if ns]
|
||||
assert len(namespaces) == 1
|
||||
namespace = namespaces[0]
|
||||
selected = asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": namespace}}))
|
||||
assert selected is not None
|
||||
selected_id = selected.config["configurable"]["checkpoint_id"]
|
||||
root_before = asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}))
|
||||
asyncio.run(app.state.thread_store.update_display_name(thread_id, "Root original"))
|
||||
metadata_before = asyncio.run(app.state.thread_store.get(thread_id))
|
||||
|
||||
async def mutation_boundary(request, *, thread_id, as_node, checkpoint_id=None, checkpoint_ns=""):
|
||||
return gateway_services.build_checkpoint_state_mutation_accessor(request, thread_id=thread_id, as_node=as_node, checkpoint_id=checkpoint_id, checkpoint_ns=checkpoint_ns, state_schema=schema)
|
||||
|
||||
monkeypatch.setattr(threads, "build_thread_checkpoint_state_mutation_accessor", mutation_boundary)
|
||||
with TestClient(app) as client:
|
||||
checkpoint = {"thread_id": thread_id, "checkpoint_ns": namespace}
|
||||
if include_checkpoint_id:
|
||||
checkpoint["checkpoint_id"] = selected_id
|
||||
response = client.post(
|
||||
f"/api/threads/{thread_id}/state",
|
||||
json={"checkpoint": checkpoint, "values": {"title": "Child updated"}},
|
||||
)
|
||||
root_response = client.get(f"/api/threads/{thread_id}/state")
|
||||
search_response = client.post("/api/threads/search", json={})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json()["values"]["title"] == "Child updated"
|
||||
assert response.json()["parent_checkpoint_id"] == selected_id
|
||||
written = asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": namespace}}))
|
||||
assert written.config["configurable"]["checkpoint_id"] == response.json()["checkpoint_id"]
|
||||
assert written.checkpoint["channel_values"]["title"] == "Child updated"
|
||||
root_after = asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}))
|
||||
assert root_after.config == root_before.config
|
||||
assert root_after.checkpoint == root_before.checkpoint
|
||||
assert root_response.status_code == 200, root_response.text
|
||||
assert root_response.json()["values"]["title"] == "Root original"
|
||||
assert asyncio.run(app.state.thread_store.get(thread_id)) == metadata_before
|
||||
assert search_response.status_code == 200, search_response.text
|
||||
entry = next(row for row in search_response.json() if row["thread_id"] == thread_id)
|
||||
assert entry["values"]["title"] == "Root original"
|
||||
assert entry["metadata"]["branch_title_sequence"] == 2
|
||||
|
||||
|
||||
@pytest.mark.parametrize("mode", ["full", "delta"])
|
||||
@pytest.mark.parametrize("selector", ["omitted", "checkpoint", "checkpoint_id"])
|
||||
def test_update_thread_state_root_title_syncs_metadata(monkeypatch, mode, selector) -> None:
|
||||
from deerflow.agents.thread_state import get_thread_state_schema
|
||||
|
||||
app, _store, checkpointer = _build_thread_app()
|
||||
app.state.checkpoint_channel_mode = mode
|
||||
thread_id = f"state-root-title-{mode}-{selector}"
|
||||
schema = get_thread_state_schema(mode)
|
||||
|
||||
async def mutation_boundary(request, *, thread_id, as_node, checkpoint_id=None, checkpoint_ns=""):
|
||||
return gateway_services.build_checkpoint_state_mutation_accessor(request, thread_id=thread_id, as_node=as_node, checkpoint_id=checkpoint_id, checkpoint_ns=checkpoint_ns, state_schema=schema)
|
||||
|
||||
monkeypatch.setattr(threads, "build_thread_checkpoint_state_mutation_accessor", mutation_boundary)
|
||||
with TestClient(app) as client:
|
||||
created = client.post("/api/threads", json={"thread_id": thread_id, "metadata": {"branch_title_sequence": 2}})
|
||||
assert created.status_code == 200, created.text
|
||||
checkpoint = asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}))
|
||||
checkpoint_id = checkpoint.config["configurable"]["checkpoint_id"]
|
||||
body = {"values": {"title": "Renamed root"}}
|
||||
if selector == "checkpoint":
|
||||
body["checkpoint"] = {"thread_id": thread_id, "checkpoint_ns": "", "checkpoint_id": checkpoint_id}
|
||||
elif selector == "checkpoint_id":
|
||||
body["checkpoint_id"] = checkpoint_id
|
||||
response = client.post(f"/api/threads/{thread_id}/state", json=body)
|
||||
root_response = client.get(f"/api/threads/{thread_id}/state")
|
||||
search_response = client.post("/api/threads/search", json={})
|
||||
|
||||
assert response.status_code == 200, response.text
|
||||
assert response.json()["values"]["title"] == "Renamed root"
|
||||
assert response.json()["parent_checkpoint_id"] == checkpoint_id
|
||||
assert root_response.status_code == 200, root_response.text
|
||||
assert root_response.json()["values"]["title"] == "Renamed root"
|
||||
record = asyncio.run(app.state.thread_store.get(thread_id))
|
||||
assert record["display_name"] == "Renamed root"
|
||||
assert "branch_title_sequence" not in record["metadata"]
|
||||
assert search_response.status_code == 200, search_response.text
|
||||
entry = next(row for row in search_response.json() if row["thread_id"] == thread_id)
|
||||
assert entry["values"]["title"] == "Renamed root"
|
||||
assert "branch_title_sequence" not in entry["metadata"]
|
||||
|
||||
|
||||
def test_update_thread_state_rejects_nonexistent_namespace_without_checkpoint_id(monkeypatch) -> None:
|
||||
app, _store, checkpointer = _build_thread_app()
|
||||
thread_id = "state-unknown-namespace"
|
||||
namespace = "child:ghost"
|
||||
|
||||
async def mutation_boundary(request, *, thread_id, as_node, checkpoint_id=None, checkpoint_ns=""):
|
||||
return gateway_services.build_checkpoint_state_mutation_accessor(request, thread_id=thread_id, as_node=as_node, checkpoint_id=checkpoint_id, checkpoint_ns=checkpoint_ns)
|
||||
|
||||
monkeypatch.setattr(threads, "build_thread_checkpoint_state_mutation_accessor", mutation_boundary)
|
||||
with TestClient(app) as client:
|
||||
assert client.post("/api/threads", json={"thread_id": thread_id, "metadata": {}}).status_code == 200
|
||||
root_before = asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}))
|
||||
response = client.post(
|
||||
f"/api/threads/{thread_id}/state",
|
||||
json={"checkpoint": {"thread_id": thread_id, "checkpoint_ns": namespace}, "values": {"title": "Must not be written"}},
|
||||
)
|
||||
|
||||
assert response.status_code == 404, response.text
|
||||
assert asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": namespace}})) is None
|
||||
root_after = asyncio.run(checkpointer.aget_tuple({"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}))
|
||||
assert root_after.config == root_before.config
|
||||
assert root_after.checkpoint == root_before.checkpoint
|
||||
|
||||
|
||||
def test_update_thread_state_rejects_checkpoint_object_for_another_thread() -> None:
|
||||
app, _store, _checkpointer = _build_thread_app()
|
||||
thread_id = "state-checkpoint-thread-mismatch"
|
||||
|
||||
with TestClient(app) as client:
|
||||
created = client.post(
|
||||
"/api/threads",
|
||||
json={"thread_id": thread_id, "metadata": {}},
|
||||
)
|
||||
assert created.status_code == 200, created.text
|
||||
response = client.post(
|
||||
f"/api/threads/{thread_id}/state",
|
||||
json={
|
||||
"checkpoint": {
|
||||
"thread_id": "another-thread",
|
||||
"checkpoint_ns": "",
|
||||
"checkpoint_id": "not-used",
|
||||
},
|
||||
"values": {"title": "Must not be written"},
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 400, response.text
|
||||
assert response.json()["detail"] == "checkpoint thread_id does not match request thread_id"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("missing_checkpoint_id", ["does-not-exist", ""])
|
||||
def test_update_thread_state_rejects_missing_explicit_checkpoint_without_writing(
|
||||
missing_checkpoint_id: str,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user