github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-08-13 16:28:38 +00:00
Code Issues Packages Projects Releases Wiki Activity
deer-flow/backend/tests/test_thread_regenerate_prepare.py
Vanzeren 42baed8c8c
feat(checkpoint): dual-mode checkpoint storage with LangGraph DeltaChannel (#4292) 
* feat(checkpoint): dual-mode checkpoint storage with LangGraph DeltaChannel

Add a restart-required database.checkpoint_channel_mode ("full" default,
"delta") that stores the messages channel via LangGraph 1.2 DeltaChannel,
cutting checkpoint storage from O(n^2) to O(n) for append-only history.
Existing full checkpoints seed delta state transparently; no data migration.

- config: mode schema + freeze-on-first-use with
  CheckpointModeReconfigurationError; mode marker persisted in checkpoint
  metadata; unsafe delta->full downgrade rejected fail-closed with
  CheckpointModeMismatchError (run-level error, failed state read)
- state: delta message state schema; CheckpointStateAccessor centralizes
  materialized reads for all consumers (threads API, branches,
  regeneration, compaction, state updates, memory, goal workers)
- runtime: raw writers (run durations, interrupted title, thread goal)
  parent their checkpoints to the checkpoint they derive from, preserving
  delta ancestry; rollback forks the pre-run lineage through a state
  mutation graph with Overwrite restores; InMemorySaver delta-history
  override delegates to the base walk (fixes dropped first write after
  migration, also present upstream)
- tests: conformance suite over {memory, sqlite, postgres} covering
  migration replay, stable message IDs, storage shape and writer
  preservation; conftest fixture isolates the frozen mode between tests;
  stale config fakes refreshed
- ci: backend unit tests gain a postgres service

* fix(checkpoint): close materialization gaps in goal flow, guard public factory

- Route goal-continuation message reads through CheckpointStateAccessor:
  raw channel_values reads see the delta sentinel in delta mode, which
  disabled goal continuation (stand_down=no_durable_end_of_turn) after
  durable assistant turns. Raw tuples remain for tuple-only metadata
  (checkpoint id, pending_writes).
- Reject checkpoint_channel_mode='delta' + checkpointer in
  create_deerflow_agent at construction: factory-built persisted graphs
  bypass mode-marker injection and the fail-closed gate, reproducing
  silent mixed-mode state loss. Delta without persistence stays allowed.
- Import the postgres saver lazily (pytest.importorskip in the fixture)
  so the documented default install collects the suite; add a CI job
  running pytest --collect-only on uv sync --group dev without extras.
- Fix test_checkpointer fallback test to patch get_app_config at its
  use site (provider module), making it deterministic when a local
  config.yaml selects a persistent backend.

* fix(gateway): preserve extension-owned channels in state mutations, bump config version

- build_state_mutation_graph / build_checkpoint_state_mutation_accessor
  accept an explicit state_schema; branch and POST /state now compile the
  mutation graph from the thread's effective schema
  (graph_state_schema on the assistant graph). The base-ThreadState
  fallback silently discarded channels contributed by custom
  AgentMiddleware.state_schema on branch (data loss) and returned a
  false-success 200 on POST /state.
- POST /state validates values keys against the mutation graph's
  channels and rejects unknown fields with 422 instead of ignoring
  them; reducer detection covers extension channels
  (BinaryOperatorAggregate or DeltaChannel) so Overwrite replace
  semantics work for middleware reducers in both modes.
- Endpoint regression: custom AgentMiddleware.state_schema value
  survives branch, updates through POST /state, and an unknown field
  receives 422.
- config_version 26 -> 27 for the new database.checkpoint_channel_mode
  (example, Helm chart values + README, support-bundle fixture), so
  existing installs get the outdated-config warning and
  make config-upgrade merges the field; covered by a test driving the
  real example file and the real config-upgrade script.

* fix(gateway): resolve assistant schema via one boundary, copy branch reducer values with Overwrite

GET /threads/{id}/state now resolves the thread's assistant_id through a
single reusable boundary (thread metadata -> assistant_id -> effective
graph), so channels contributed by a custom AgentMiddleware.state_schema
are materialized instead of dropped by the default lead schema. POST
/state uses the same boundary instead of resolving the schema ad hoc.

Branch writes wrap every copied reducer channel in Overwrite (derived
from the effective mutation graph: BinaryOperatorAggregate + DeltaChannel),
not just messages, so already-aggregated values are never re-merged.

Regression tests use a real AgentMiddleware.state_schema with a
non-identity reducer in both full and delta modes: GET /state returns the
extension value, POST /state replaces it, branch preserves it
byte-for-byte; the unknown-field 422 is a separate assertion.

* refactor(checkpoint): collapse read-path round-trips and ship dual-mode parity tests

Address review round 4 on PR #4292:

- Push ahistory/history limit through Pregel into checkpointer.alist
  (SQL LIMIT) instead of materializing all rows and breaking in Python
- Fold the read-side mode-compat gate onto the returned snapshot's
  metadata; only writes keep the pre-write tuple fetch (fail-closed)
- Cache factory-built accessor graphs per (assistant_id, mode) with
  factory-identity revalidation; state reads no longer build a lead
  agent per request
- get_thread: one snapshot fetch + one raw pending_writes fetch on the
  resolved checkpoint (post-checkpoint __error__ writes never surface
  in snapshot.tasks; verified empirically)
- DeerFlowClient.get_thread: single checkpointer.list walk collects
  pending_writes per checkpoint instead of N get_tuple calls
- InMemorySaver delta-history patch: stand-down when the upstream
  override disappears, try/except guard, validated-version warning,
  guard tests
- make_lead_agent mode precedence: first freeze is owned by app_config
  (client-supplied configurable key ignored); once frozen, injected
  key/app_config must match or fail closed
- Rollback: lock in non-message channel restoration via fork
  inheritance with a dedicated reducer-channel test
- Add tests/test_threads_checkpoint_mode.py and
  tests/test_gateway_checkpoint_mode.py referenced by AGENTS.md and
  the PR validation section: lifecycle parity (memory + sqlite),
  per-step blob-count storage guard, gateway endpoint parity

Counted-saver tests pin checkpoint round-trips for aget/ahistory so
these regressions cannot silently return.

* fix(checkpoint): precise mode-mismatch HTTP mapping, gate E2E, and accessor resilience

- threads router: map CheckpointModeMismatchError to 409 (with cause and
  thread id) and CheckpointModeReconfigurationError to 503 across all state
  endpoints instead of swallowing both into a generic 500
- gate coverage: seed a real delta checkpoint into AsyncSqliteSaver and
  assert aget/aupdate/ahistory fail closed; assert 409 at the HTTP boundary
  through the real route stack
- rollback: compile the restore mutation graph with the thread's effective
  state schema per the build_state_mutation_graph contract
- inheritance contract locks: rollback and manual compaction preserve
  middleware-contributed channels via checkpoint fork cloning
- services: revalidate the accessor-graph cache against app_config identity
  so config.yaml hot-reloads never serve a stale compiled graph
- services: degrade full-mode state reads to raw checkpointer reads when the
  agent factory is unavailable (delta gate still applies; delta mode has no
  fallback)
- deps: override websockets==16.0 (langgraph-sdk 0.4.2's <16 pin silently
  downgraded 16.0 -> 15.0.1; pin is not grounded in any API incompatibility)
  and bump the langchain lower bound to what the lockfile actually resolves

* fix(checkpoint): include anchor checkpoint in degraded history walk + cover get_thread

- _RawCheckpointReadAccessor.ahistory: alist(before=...) is exclusive while
  pregel's get_state_history treats config.checkpoint_id as the inclusive
  start; fetch the anchor explicitly so both read paths paginate identically
- extend the degraded-path gateway test: GET /thread returns raw values, and
  POST /history with before starts at the anchor checkpoint

* fix(gateway): preserve degraded checkpoint timestamps

* fix(gateway): harden degraded checkpoint access

* fix(gateway): resolve assistants for checkpoint reads

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
2026-07-22 08:33:29 +08:00

565 lines
21 KiB
Python
Raw Blame History

from __future__ import annotations
import asyncio
from types import SimpleNamespace
from unittest.mock import AsyncMock, patch
import pytest
from fastapi import HTTPException
from langchain_core.messages import AIMessage, HumanMessage
from deerflow.runtime import RunStatus
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
def _checkpoint(checkpoint_id: str, messages: list[object], *, metadata: dict | None = None):
return SimpleNamespace(
config={
"configurable": {
"thread_id": "thread-1",
"checkpoint_ns": "",
"checkpoint_id": checkpoint_id,
"checkpoint_map": None,
}
},
checkpoint={"channel_values": {"messages": messages}},
metadata=metadata or {},
)
class FakeCheckpointer:
def __init__(self, history, *, latest=None, materialized_history=None, materialized_latest=None):
self.history = history
self.latest = latest
self.materialized_history = materialized_history
self.materialized_latest = materialized_latest
self.alist_limits = []
async def aget_tuple(self, config):
checkpoint_id = config.get("configurable", {}).get("checkpoint_id")
if checkpoint_id:
return next((item for item in self.history if item.config["configurable"]["checkpoint_id"] == checkpoint_id), None)
return self.latest or (self.history[0] if self.history else None)
async def alist(self, config, limit=200):
self.alist_limits.append(limit)
for item in self.history[:limit]:
yield item
def _snapshot(checkpoint_id: str, messages: list[object], *, metadata: dict | None = None):
return SimpleNamespace(
values={"messages": messages},
config={
"configurable": {
"thread_id": "thread-1",
"checkpoint_ns": "",
"checkpoint_id": checkpoint_id,
"checkpoint_map": None,
}
},
metadata=metadata or {},
)
class FakeAccessor:
def __init__(self, checkpointer: FakeCheckpointer):
self.checkpointer = checkpointer
@staticmethod
def _from_raw(checkpoint):
return SimpleNamespace(
values=dict(checkpoint.checkpoint.get("channel_values", {})),
config=checkpoint.config,
metadata=checkpoint.metadata,
)
async def aget(self, _config):
if self.checkpointer.materialized_latest is not None:
return self.checkpointer.materialized_latest
raw = self.checkpointer.latest or (self.checkpointer.history[0] if self.checkpointer.history else None)
return self._from_raw(raw) if raw is not None else SimpleNamespace(values={}, config={}, metadata={})
async def ahistory(self, _config, *, limit=None):
self.checkpointer.alist_limits.append(limit)
history = self.checkpointer.materialized_history
if history is None:
history = [self._from_raw(item) for item in self.checkpointer.history]
return history[:limit]
@pytest.fixture(autouse=True)
def _patch_checkpoint_accessor(monkeypatch):
from app.gateway.routers import thread_runs
def build_accessor(request, *, thread_id, assistant_id=None, checkpoint_id=None):
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
if checkpoint_id is not None:
config["configurable"]["checkpoint_id"] = checkpoint_id
return FakeAccessor(request.app.state.checkpointer), config
async def build_thread_accessor(request, *, thread_id, checkpoint_id=None):
return build_accessor(request, thread_id=thread_id, checkpoint_id=checkpoint_id)
monkeypatch.setattr(thread_runs, "build_checkpoint_state_accessor", build_accessor)
monkeypatch.setattr(thread_runs, "build_thread_checkpoint_state_accessor", build_thread_accessor)
class FakeEventStore:
def __init__(self, rows):
self.rows = rows
async def list_messages(self, thread_id, *, limit=50, before_seq=None, after_seq=None):
return self.rows[-limit:]
class FakeRunManager:
def __init__(self, records):
self.records = records
async def list_by_thread(self, thread_id, *, user_id=None, limit=100):
return self.records[:limit]
def _request(checkpointer, event_store, *, run_manager=None, user_id="user-1"):
from app.gateway.auth_disabled import AUTH_SOURCE_SESSION
return SimpleNamespace(
app=SimpleNamespace(
state=SimpleNamespace(
checkpointer=checkpointer,
run_event_store=event_store,
run_manager=run_manager or FakeRunManager([]),
)
),
state=SimpleNamespace(user=SimpleNamespace(id=user_id), auth_source=AUTH_SOURCE_SESSION),
)
def test_run_wait_readers_return_materialized_final_values() -> None:
from app.gateway.routers import runs, thread_runs
snapshot = SimpleNamespace(
values={
"messages": [
HumanMessage(id="h1", content="question"),
AIMessage(id="a1", content="answer"),
]
},
config={
"configurable": {
"thread_id": "thread-1",
"checkpoint_ns": "",
"checkpoint_id": "ckpt-2",
}
},
parent_config={"configurable": {"checkpoint_id": "ckpt-1"}},
metadata={"step": 2},
next=(),
tasks=(),
created_at=None,
)
accessor = SimpleNamespace(aget=AsyncMock(return_value=snapshot))
record = SimpleNamespace(
run_id="run-1",
thread_id="thread-1",
task=None,
status=RunStatus.success,
error=None,
)
request = SimpleNamespace()
body = thread_runs.RunCreateRequest(
assistant_id="lead-agent",
config={"configurable": {"thread_id": "thread-1"}},
)
async def _scenario() -> tuple[dict, dict]:
with (
patch.object(thread_runs, "get_stream_bridge", return_value=object()),
patch.object(thread_runs, "get_run_manager", return_value=object()),
patch.object(thread_runs, "start_run", AsyncMock(return_value=record)),
patch.object(
thread_runs,
"build_checkpoint_state_accessor",
create=True,
return_value=(accessor, snapshot.config),
),
patch.object(runs, "get_stream_bridge", return_value=object()),
patch.object(runs, "get_run_manager", return_value=object()),
patch.object(runs, "start_run", AsyncMock(return_value=record)),
patch.object(
runs,
"build_checkpoint_state_accessor",
create=True,
return_value=(accessor, snapshot.config),
),
):
thread_result = await thread_runs.wait_run.__wrapped__("thread-1", body, request)
stateless_result = await runs.stateless_wait(body, request)
return thread_result, stateless_result
thread_result, stateless_result = asyncio.run(_scenario())
assert [message["id"] for message in thread_result["messages"]] == ["h1", "a1"]
assert [message["id"] for message in stateless_result["messages"]] == ["h1", "a1"]
def test_run_wait_readers_preserve_terminal_error_without_checkpoint() -> None:
from app.gateway.routers import runs, thread_runs
snapshot = SimpleNamespace(
values={},
config={"configurable": {"thread_id": "thread-1", "checkpoint_ns": ""}},
parent_config=None,
metadata={},
next=(),
tasks=(),
created_at=None,
)
accessor = SimpleNamespace(aget=AsyncMock(return_value=snapshot))
record = SimpleNamespace(
run_id="run-1",
thread_id="thread-1",
task=None,
status=RunStatus.error,
error="run failed before checkpoint",
)
request = SimpleNamespace()
body = thread_runs.RunCreateRequest(config={"configurable": {"thread_id": "thread-1"}})
async def _scenario() -> tuple[dict, dict]:
with (
patch.object(thread_runs, "get_stream_bridge", return_value=object()),
patch.object(thread_runs, "get_run_manager", return_value=object()),
patch.object(thread_runs, "start_run", AsyncMock(return_value=record)),
patch.object(
thread_runs,
"build_checkpoint_state_accessor",
return_value=(accessor, snapshot.config),
),
patch.object(runs, "get_stream_bridge", return_value=object()),
patch.object(runs, "get_run_manager", return_value=object()),
patch.object(runs, "start_run", AsyncMock(return_value=record)),
patch.object(
runs,
"build_checkpoint_state_accessor",
return_value=(accessor, snapshot.config),
),
):
thread_result = await thread_runs.wait_run.__wrapped__("thread-1", body, request)
stateless_result = await runs.stateless_wait(body, request)
return thread_result, stateless_result
thread_result, stateless_result = asyncio.run(_scenario())
expected = {"status": "error", "error": "run failed before checkpoint"}
assert thread_result == expected
assert stateless_result == expected
@pytest.mark.parametrize("route_name", ["thread", "stateless"])
def test_run_wait_readers_preserve_terminal_error_when_accessor_builder_fails(route_name: str) -> None:
from app.gateway.routers import runs, thread_runs
record = SimpleNamespace(
run_id="run-1",
thread_id="thread-1",
task=None,
status=RunStatus.error,
error="run failed before checkpoint",
)
request = SimpleNamespace()
body = thread_runs.RunCreateRequest(config={"configurable": {"thread_id": "thread-1"}})
async def _scenario() -> dict:
if route_name == "thread":
with (
patch.object(thread_runs, "get_stream_bridge", return_value=object()),
patch.object(thread_runs, "get_run_manager", return_value=object()),
patch.object(thread_runs, "start_run", AsyncMock(return_value=record)),
patch.object(
thread_runs,
"build_checkpoint_state_accessor",
side_effect=RuntimeError("graph construction failed"),
),
):
return await thread_runs.wait_run.__wrapped__("thread-1", body, request)
with (
patch.object(runs, "get_stream_bridge", return_value=object()),
patch.object(runs, "get_run_manager", return_value=object()),
patch.object(runs, "start_run", AsyncMock(return_value=record)),
patch.object(
runs,
"build_checkpoint_state_accessor",
side_effect=RuntimeError("graph construction failed"),
),
):
return await runs.stateless_wait(body, request)
result = asyncio.run(_scenario())
assert result == {"status": "error", "error": "run failed before checkpoint"}
def test_prepare_regenerate_payload_returns_clean_input_and_base_checkpoint():
from app.gateway.routers import thread_runs
human = HumanMessage(
id="human-1",
content="<uploaded_files>injected</uploaded_files>\n\n/data-analysis analyze data.csv",
additional_kwargs={
ORIGINAL_USER_CONTENT_KEY: "/data-analysis analyze data.csv",
"files": [{"filename": "data.csv", "path": "/mnt/user-data/uploads/data.csv"}],
},
)
ai = AIMessage(id="ai-1", content="answer v1")
base = _checkpoint("ckpt-base", [])
after_human = _checkpoint("ckpt-human", [human])
latest = _checkpoint("ckpt-ai", [human, ai])
checkpointer = FakeCheckpointer([latest, after_human, base])
event_store = FakeEventStore(
[
{
"run_id": "run-old",
"event_type": "llm.ai.response",
"category": "message",
"content": {"id": "ai-1", "type": "ai", "content": "answer v1"},
"metadata": {"caller": "lead_agent"},
}
]
)
original_builder = thread_runs.build_thread_checkpoint_state_accessor
thread_builder = AsyncMock(side_effect=original_builder)
with patch.object(thread_runs, "build_thread_checkpoint_state_accessor", thread_builder):
response = asyncio.run(thread_runs._prepare_regenerate_payload("thread-1", "ai-1", _request(checkpointer, event_store)))
assert [call.kwargs["thread_id"] for call in thread_builder.await_args_list] == ["thread-1", "thread-1"]
assert response.checkpoint == {
"checkpoint_ns": "",
"checkpoint_id": "ckpt-base",
"checkpoint_map": None,
}
assert response.target_run_id == "run-old"
assert response.metadata == {
"regenerate_from_message_id": "ai-1",
"regenerate_from_run_id": "run-old",
"regenerate_checkpoint_id": "ckpt-base",
}
regenerated_human = response.input["messages"][0]
assert regenerated_human["id"] == "human-1"
assert regenerated_human["content"] == [{"type": "text", "text": "/data-analysis analyze data.csv"}]
assert regenerated_human["additional_kwargs"] == {"files": [{"filename": "data.csv", "path": "/mnt/user-data/uploads/data.csv"}]}
def test_prepare_regenerate_uses_materialized_history_when_raw_messages_are_omitted():
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
earlier_human = HumanMessage(id="human-0", content="earlier question")
earlier_ai = AIMessage(id="ai-0", content="earlier answer")
target_human = HumanMessage(id="human-1", content="question")
target_ai = AIMessage(id="ai-1", content="answer")
raw_latest = _checkpoint("ckpt-ai", [])
raw_after_human = _checkpoint("ckpt-human", [])
raw_base = _checkpoint("ckpt-base", [])
materialized_history = [
_snapshot("ckpt-ai", [earlier_human, earlier_ai, target_human, target_ai]),
_snapshot("ckpt-human", [earlier_human, earlier_ai, target_human]),
_snapshot("ckpt-base", [earlier_human, earlier_ai]),
]
checkpointer = FakeCheckpointer(
[raw_latest, raw_after_human, raw_base],
latest=raw_latest,
materialized_history=materialized_history,
materialized_latest=materialized_history[0],
)
event_store = FakeEventStore(
[
{
"run_id": "run-target",
"event_type": "ai_message",
"category": "message",
"content": {"id": "ai-1", "type": "ai", "content": "answer"},
"metadata": {"caller": "lead_agent"},
}
]
)
response = asyncio.run(
_prepare_regenerate_payload(
"thread-1",
"ai-1",
_request(checkpointer, event_store),
)
)
assert response.checkpoint["checkpoint_id"] == "ckpt-base"
assert response.metadata["regenerate_checkpoint_id"] == "ckpt-base"
assert response.input["messages"][0]["id"] == "human-1"
assert checkpointer.alist_limits == [400]
def test_prepare_regenerate_payload_rejects_non_latest_assistant():
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
human = HumanMessage(id="human-1", content="question")
old_ai = AIMessage(id="ai-old", content="old")
latest_ai = AIMessage(id="ai-latest", content="latest")
base = _checkpoint("ckpt-base", [])
after_human = _checkpoint("ckpt-human", [human])
latest = _checkpoint("ckpt-latest", [human, old_ai, latest_ai])
checkpointer = FakeCheckpointer([latest, after_human, base])
event_store = FakeEventStore(
[
{
"run_id": "run-old",
"event_type": "ai_message",
"category": "message",
"content": {"id": "ai-old", "type": "ai", "content": "old"},
"metadata": {"caller": "lead_agent"},
}
]
)
with pytest.raises(HTTPException) as exc:
asyncio.run(_prepare_regenerate_payload("thread-1", "ai-old", _request(checkpointer, event_store)))
assert exc.value.status_code == 409
assert exc.value.detail == "Only the latest assistant message can be regenerated"
def test_prepare_regenerate_payload_falls_back_to_matching_run_when_events_are_missing():
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
human = HumanMessage(id="human-1", content="question")
ai = AIMessage(id="ai-1", content="answer")
base = _checkpoint("ckpt-base", [])
after_human = _checkpoint("ckpt-human", [human])
latest = _checkpoint("ckpt-ai", [human, ai])
checkpointer = FakeCheckpointer([latest, after_human, base])
run_manager = FakeRunManager(
[
SimpleNamespace(run_id="run-latest", status=RunStatus.success, last_ai_message="answer"),
SimpleNamespace(run_id="run-older", status=RunStatus.error, last_ai_message="answer"),
]
)
response = asyncio.run(
_prepare_regenerate_payload(
"thread-1",
"ai-1",
_request(checkpointer, FakeEventStore([]), run_manager=run_manager),
)
)
assert response.target_run_id == "run-latest"
assert response.metadata["regenerate_from_run_id"] == "run-latest"
def test_prepare_regenerate_payload_rejects_unverified_run_fallback_when_events_are_missing():
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
human = HumanMessage(id="human-1", content="question")
ai = AIMessage(id="ai-1", content="answer")
base = _checkpoint("ckpt-base", [])
after_human = _checkpoint("ckpt-human", [human])
latest = _checkpoint("ckpt-ai", [human, ai])
checkpointer = FakeCheckpointer([latest, after_human, base])
run_manager = FakeRunManager(
[
SimpleNamespace(run_id="run-latest", status=RunStatus.success, last_ai_message="different"),
]
)
with pytest.raises(HTTPException) as exc:
asyncio.run(
_prepare_regenerate_payload(
"thread-1",
"ai-1",
_request(checkpointer, FakeEventStore([]), run_manager=run_manager),
)
)
assert exc.value.status_code == 409
assert exc.value.detail == "Could not find source run for assistant message"
def test_prepare_regenerate_payload_requires_addressable_checkpoint_before_human():
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
human = HumanMessage(id="human-1", content="question")
ai = AIMessage(id="ai-1", content="answer")
latest = _checkpoint("ckpt-ai", [human, ai])
checkpointer = FakeCheckpointer([latest])
event_store = FakeEventStore(
[
{
"run_id": "run-old",
"event_type": "llm.ai.response",
"category": "message",
"content": {"id": "ai-1", "type": "ai", "content": "answer"},
"metadata": {"caller": "lead_agent"},
}
]
)
with pytest.raises(HTTPException) as exc:
asyncio.run(_prepare_regenerate_payload("thread-1", "ai-1", _request(checkpointer, event_store)))
assert exc.value.status_code == 409
assert exc.value.detail == "Could not find an addressable checkpoint before the target user message"
assert checkpointer.alist_limits == [400]
def test_prepare_regenerate_payload_reports_recent_checkpoint_scan_limit():
from app.gateway.routers.thread_runs import _prepare_regenerate_payload
human = HumanMessage(id="human-1", content="question")
ai = AIMessage(id="ai-1", content="answer")
latest = _checkpoint("ckpt-latest", [human, ai])
history_without_human = [_checkpoint(f"ckpt-{index}", []) for index in range(201)]
checkpointer = FakeCheckpointer(history_without_human, latest=latest)
event_store = FakeEventStore(
[
{
"run_id": "run-old",
"event_type": "llm.ai.response",
"category": "message",
"content": {"id": "ai-1", "type": "ai", "content": "answer"},
"metadata": {"caller": "lead_agent"},
}
]
)
with pytest.raises(HTTPException) as exc:
asyncio.run(_prepare_regenerate_payload("thread-1", "ai-1", _request(checkpointer, event_store)))
assert exc.value.status_code == 409
assert exc.value.detail == "Could not locate target user message in recent checkpoint history (limit=200)"
assert checkpointer.alist_limits == [400]
def test_find_base_checkpoint_ignores_duration_only_checkpoints() -> None:
from app.gateway.routers.thread_runs import _find_base_checkpoint_before_human
human = HumanMessage(id="human-1", content="question")
duration_checkpoints = [
_checkpoint(
f"duration-{index}",
[],
metadata={"writes": {"runtime_run_duration": {"run_ids": [f"run-{index}"]}}},
)
for index in range(200)
]
base = _checkpoint("ckpt-base", [])
after_human = _checkpoint("ckpt-human", [human])
checkpointer = FakeCheckpointer([*duration_checkpoints, after_human, base])
result = asyncio.run(_find_base_checkpoint_before_human("thread-1", "human-1", _request(checkpointer, FakeEventStore([]))))
assert result.config == base.config
assert checkpointer.alist_limits == [400]
Reference in New Issue View Git Blame Copy Permalink