"""Runtime boundaries for project-scoped cross-conversation memory.""" import json from datetime import UTC, datetime, timedelta from types import SimpleNamespace from unittest.mock import AsyncMock import pytest from fastapi import HTTPException from langchain_core.messages import AIMessage, HumanMessage from app.gateway import analysis_project_memory as runtime from app.gateway.routers import threads as thread_routes from deerflow.config import system_settings from deerflow.persistence.analysis_project_memory import AnalysisProjectMemoryStore @pytest.mark.asyncio async def test_model_extraction_projects_existing_memory_to_json_safe_fields(monkeypatch): store = SimpleNamespace( recall=AsyncMock( return_value={ "snapshot": {"summary": "已有摘要"}, "items": [ { "logical_key": "fact:region", "kind": "fact", "content": "目标区域是华东", "confidence": 0.9, "importance": 0.8, "pinned": False, "source_thread_id": "thread-a", "created_at": datetime.now(UTC), "embedding": [0.1] * 1_024, "embedding_json": "[0.1]", } ], } ) ) captured = {} class Model: async def ainvoke(self, messages, *, config): captured["system"] = messages[0].content captured["source"] = json.loads(messages[1].content) captured["config"] = config return SimpleNamespace( content=json.dumps( { "items": [ { "logical_key": "fact:region", "kind": "fact", "content": "目标区域是华东", "confidence": 0.9, "importance": 0.8, "evidence_ids": ["m2"], } ], "snapshot": {"summary": "目标区域是华东"}, }, ensure_ascii=False, ) ) monkeypatch.setattr(runtime, "create_chat_model", lambda **_kwargs: Model()) app = SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, config=SimpleNamespace(memory=SimpleNamespace(v1=SimpleNamespace(model_name=None))), ) ) result = await runtime.AnalysisProjectMemoryExecutor(app)._model_extraction( { "project_id": "project-a", "owner_user_id": "alice", "payload": { "project_name": "区域规划", "project_background": "年度规划", "messages": [ {"role": "user", "content": "我们的目标区域是什么？"}, {"role": "assistant", "content": "华东。"}, ], }, } ) existing = captured["source"]["existing_memory"]["items"][0] assert existing == { "logical_key": "fact:region", "kind": "fact", "content": "目标区域是华东", "confidence": 0.9, "importance": 0.8, "pinned": False, "source_thread_id": "thread-a", "evidence_authority": "unknown", } assert captured["source"]["conversation"] == [ { "evidence_id": "m1", "id": "", "role": "user", "content": "我们的目标区域是什么？", }, { "evidence_id": "m2", "id": "", "role": "assistant", "content": "华东。", }, ] assert captured["config"] == {"run_name": "analysis_project_memory_extract"} assert "仅过滤助手自行报告的真实运行时当前日期/时间" in captured["system"] assert "本项目假定今天为某日" in captured["system"] assert result["items"][0]["content"] == "目标区域是华东" assert result["items"][0]["confidence"] == 0.55 assert result["items"][0]["metadata"]["evidence_authority"] == "assistant" def test_extraction_redacts_credentials_and_privacy_data(): cleaned = runtime._clean_extraction( { "items": [ { "logical_key": "fact:contact", "kind": "fact", "content": "token=super-secret；手机号 13800138000", "confidence": 0.8, "importance": 0.7, } ], "snapshot": {"summary": "password=hunter2；联系 13800138000"}, } ) serialized = json.dumps(cleaned, ensure_ascii=False) assert "super-secret" not in serialized assert "hunter2" not in serialized assert "13800138000" not in serialized assert "REDACTED" in serialized or "脱敏" in serialized def test_fallback_requires_complete_exchange_and_is_bounded(): user_only = runtime._fallback_extraction( {"messages": [{"role": "user", "content": "只有问题"}]} ) assert user_only["items"][0]["content"] == "历史用户消息：只有问题" result = runtime._fallback_extraction( { "messages": [ {"role": "user", "content": "项目代号是什么？"}, {"role": "assistant", "content": "项目代号是星河。"}, ] } ) assert len(result["items"]) == 1 assert result["items"][0]["kind"] == "episode" assert "项目代号是什么" in result["items"][0]["content"] assert "星河" not in result["items"][0]["content"] assert result["items"][0]["metadata"]["evidence_authority"] == "user" def test_fallback_preserves_earlier_explicit_user_setting_not_only_last_question(): result = runtime._fallback_extraction( { "messages": [ { "role": "user", "content": "修改一下设定，记忆日期为2026年12月31日。", }, {"role": "assistant", "content": "已记录。"}, {"role": "user", "content": "今天日期多少号？"}, {"role": "assistant", "content": "今天是2026年9月24日。"}, ] } ) assert [item["content"] for item in result["items"]] == [ "历史用户消息：修改一下设定，记忆日期为2026年12月31日。", "历史用户消息：今天日期多少号？", ] assert all("2026年9月24日" not in item["content"] for item in result["items"]) assert all( item["metadata"]["evidence_authority"] == "user" for item in result["items"] ) def test_ground_extraction_rejects_assistant_capability_and_volatile_date_claims(): conversation = runtime._conversation_with_evidence_ids( [ {"id": "u1", "role": "user", "content": "今天日期多少号？"}, { "id": "a1", "role": "assistant", "content": "当前系统日期为2026年9月24日，星期四。", }, {"id": "u2", "role": "user", "content": "你能查看其他对话吗？"}, {"id": "a2", "role": "assistant", "content": "我无法访问其他对话。"}, ] ) extraction = runtime._clean_extraction( { "items": [ { "logical_key": "memory-access", "kind": "constraint", "content": "助手无法访问其他对话", "confidence": 0.95, "importance": 0.8, # The model incorrectly cites the user's lookup request as # if it confirmed the assistant's capability statement. "evidence_ids": ["m3", "m4"], }, { "logical_key": "memory-date", "kind": "fact", "content": "当前系统日期为2026年9月24日，星期四", "confidence": 0.95, "importance": 0.8, # The user only asked the question and never supplied this # date, so m1 must not elevate the assistant answer. "evidence_ids": ["m1", "m2"], }, { "logical_key": "memory-date-en", "kind": "fact", "content": "System date is 2026-09-24 (Thursday)", "confidence": 0.95, "importance": 0.8, # The same unsupported runtime claim may be translated by # the extractor; it must be rejected in English too. "evidence_ids": ["m1", "m2"], }, ] } ) grounded = runtime._ground_extraction(extraction, conversation) assert grounded["items"] == [] def test_ground_extraction_keeps_explicit_user_setting_with_user_authority(): conversation = runtime._conversation_with_evidence_ids( [ { "id": "u1", "role": "user", "content": "本项目假定今天为2026年12月31日。", }, {"id": "a1", "role": "assistant", "content": "已经修改。"}, ] ) extraction = runtime._clean_extraction( { "items": [ { "logical_key": "memory-date", "kind": "fact", "content": "本项目模拟日期为2026年12月31日", "confidence": 0.98, "importance": 0.9, "evidence_ids": ["m1", "m2"], } ] } ) [item] = runtime._ground_extraction(extraction, conversation)["items"] assert item["content"] == "本项目模拟日期为2026年12月31日" assert item["source_message_ids"] == ["u1"] assert item["metadata"] == { "evidence_authority": "user", "evidence_roles": ["user"], } @pytest.mark.parametrize( ("user_text", "nickname"), [ ("你叫小爱同学", "小爱同学"), ("你现在叫铁蛋，我给你起的新名字", "铁蛋"), ("对，你的昵称叫铁蛋", "铁蛋"), ("以后就叫你 Tiedan", "Tiedan"), ("我给你起了个新昵称：小鹿", "小鹿"), ], ) def test_ground_extraction_deterministically_keeps_user_assigned_nickname( user_text, nickname ): conversation = runtime._conversation_with_evidence_ids( [ {"id": "u1", "role": "user", "content": user_text}, { "id": "a1", "role": "assistant", "content": "系统身份不能修改。", }, ] ) [item] = runtime._ground_extraction( {"items": [], "snapshot": None}, conversation )["items"] assert item["logical_key"] == "assistant.nickname" assert item["kind"] == "preference" assert item["content"] == f"本项目中用户给助手设定的昵称是“{nickname}”" assert item["source_message_ids"] == ["u1"] assert item["metadata"] == { "evidence_authority": "user", "evidence_roles": ["user"], "deterministic": "assistant_nickname", } def test_ground_extraction_latest_nickname_wins_and_rejects_model_semantic_reversal(): conversation = runtime._conversation_with_evidence_ids( [ {"id": "u1", "role": "user", "content": "你叫小爱同学"}, { "id": "a1", "role": "assistant", "content": "我不是小爱同学，我叫CM助手。", }, {"id": "u2", "role": "user", "content": "对，你的昵称叫铁蛋"}, {"id": "a2", "role": "assistant", "content": "你可以叫我铁蛋。"}, ] ) extraction = runtime._clean_extraction( { "items": [ { "logical_key": "cm-assistant-identity", "kind": "fact", "content": "CM助手是项目内部智能体，不是小爱同学", "confidence": 1, "importance": 0.8, "evidence_ids": ["m1", "m2"], }, { "logical_key": "assistant-name", "kind": "fact", "content": "助手的名字是CM助手", "confidence": 1, "importance": 0.8, "evidence_ids": ["m3", "m4"], }, ] } ) [item] = runtime._ground_extraction(extraction, conversation)["items"] assert item["logical_key"] == "assistant.nickname" assert item["content"] == "本项目中用户给助手设定的昵称是“铁蛋”" assert item["source_message_ids"] == ["u2"] @pytest.mark.parametrize( "user_text", [ "你的昵称叫什么？", "你叫铁蛋吗？", "你叫铁蛋？", "你的昵称叫铁蛋吗？", "你叫 Tiedan 吗？", "你的昵称叫 Tiedan 吗", "我不是给你起了一个新的昵称", "不要叫铁蛋", "取消昵称铁蛋", ], ) def test_ground_extraction_does_not_invent_nickname_from_question_or_negation( user_text, ): conversation = runtime._conversation_with_evidence_ids( [{"id": "u1", "role": "user", "content": user_text}] ) assert runtime._ground_extraction( {"items": [], "snapshot": None}, conversation )["items"] == [] def test_conversation_episode_keeps_bounded_raw_exchange_for_other_project_threads(): conversation = runtime._conversation_with_evidence_ids( [ {"id": "u1", "role": "user", "content": "项目代号是什么？"}, {"id": "a1", "role": "assistant", "content": "项目代号是星河。"}, ] ) [item] = runtime._conversation_episode_items(conversation) assert item["kind"] == "episode" assert "用户：项目代号是什么？" in item["content"] assert "助手：项目代号是星河。" in item["content"] assert item["source_message_ids"] == ["u1", "a1"] assert item["metadata"]["raw_project_conversation"] is True @pytest.mark.asyncio async def test_project_pause_skips_extractor_model(monkeypatch): store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": False}), complete_job=AsyncMock(return_value=None), ) app = SimpleNamespace( state=SimpleNamespace(analysis_project_memory_store=store) ) executor = runtime.AnalysisProjectMemoryExecutor(app) extraction = AsyncMock() monkeypatch.setattr(executor, "_model_extraction", extraction) await executor.execute( {"id": "job-a", "project_id": "project-a", "owner_user_id": "alice"}, lease_owner="worker-a", ) extraction.assert_not_awaited() store.complete_job.assert_awaited_once_with( "job-a", lease_owner="worker-a", result={"skipped": "project_memory_disabled"}, ) @pytest.mark.asyncio async def test_executor_redacts_persisted_failure_details(monkeypatch): store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": True}), apply_extraction=AsyncMock( side_effect=RuntimeError( "upstream failed with token=super-secret and phone 13800138000" ) ), fail_job=AsyncMock(return_value=None), ) app = SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, llmwiki_embedding=None, ) ) executor = runtime.AnalysisProjectMemoryExecutor(app) monkeypatch.setattr( executor, "_model_extraction", AsyncMock(return_value={"items": [], "snapshot": None}), ) await executor.execute( { "id": "job-a", "project_id": "project-a", "owner_user_id": "alice", "payload": {"messages": []}, "attempt": 1, }, lease_owner="worker-a", ) error = store.fail_job.await_args.kwargs["error"] assert "super-secret" not in error assert "13800138000" not in error assert "REDACTED" in error or "脱敏" in error @pytest.mark.asyncio async def test_queued_payload_is_redacted_before_durable_storage(monkeypatch): store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": True}), enqueue_job=AsyncMock( return_value=({"id": "job-a", "status": "queued"}, True) ), ) class Checkpointer: async def aget_tuple(self, _config): return SimpleNamespace( checkpoint={ "channel_values": { "messages": [ HumanMessage( id="message-user", content="token=super-secret；手机号 13800138000", ), AIMessage( id="message-assistant", content="password=hunter2；已经记录", ), ] } } ) app = SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, analysis_project_memory_dispatcher=None, checkpointer=Checkpointer(), ) ) result = await runtime.enqueue_completed_project_run( app=app, project={ "id": "project-a", "user_id": "alice", "mode": "qa", "name": "password=project-secret", "background": "联系人 13800138000", }, thread_id="thread-a", run_id="run-a", ) assert result == {"id": "job-a", "status": "queued"} payload = store.enqueue_job.await_args.kwargs["payload"] serialized = json.dumps(payload, ensure_ascii=False) for secret in ( "super-secret", "hunter2", "project-secret", "13800138000", ): assert secret not in serialized assert "REDACTED" in serialized or "脱敏" in serialized @pytest.mark.asyncio async def test_dispatcher_does_not_claim_beyond_concurrency(): store = SimpleNamespace(claim_jobs=AsyncMock(return_value=[])) dispatcher = runtime.AnalysisProjectMemoryDispatcher( store, SimpleNamespace(), concurrency=2, ) dispatcher._running = {"job-a": object(), "job-b": object()} assert await dispatcher.dispatch_once() == 0 store.claim_jobs.assert_not_awaited() @pytest.mark.asyncio async def test_generic_memory_switch_off_does_not_block_project_recall(monkeypatch): monkeypatch.setattr( system_settings, "load_system_settings", lambda: SimpleNamespace(memory=SimpleNamespace(enabled=False)), ) store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": True}), recall=AsyncMock( return_value={ "state": {"status": "ready", "revision": 1}, "snapshot": {}, "items": [{"content": "项目日期为2026年12月31日"}], } ), ) request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, llmwiki_embedding=None, ) ) ) scope, recalled = await runtime.prepare_project_memory( request=request, project={ "id": "project-a", "user_id": "alice", "mode": "qa", }, thread_id="thread-b", query="上次结论是什么？", ) assert scope is not None and scope["enabled"] is True assert recalled is not None assert recalled["items"][0]["content"] == "项目日期为2026年12月31日" store.get_settings.assert_awaited_once() store.recall.assert_awaited_once() @pytest.mark.asyncio async def test_project_recall_does_not_wait_for_query_embedding_before_stream(): embedding = SimpleNamespace( embed_query=AsyncMock(return_value=[0.1, 0.2]), fingerprint="remote-encoder-v1", ) store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": True}), recall=AsyncMock( return_value={ "state": {"status": "ready", "revision": 1}, "snapshot": {}, "items": [{"content": "本项目中助手昵称是铁蛋"}], } ), ) request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, llmwiki_embedding=embedding, ) ) ) _scope, recalled = await runtime.prepare_project_memory( request=request, project={"id": "project-a", "user_id": "alice", "mode": "qa"}, thread_id="thread-b", query="我给你起的昵称是什么？", ) assert recalled is not None embedding.embed_query.assert_not_awaited() assert store.recall.await_args.kwargs["query_embedding"] is None @pytest.mark.asyncio async def test_incremental_building_state_keeps_indexed_memory_recallable(): store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": True}), recall=AsyncMock( return_value={ "state": {"status": "building", "revision": 1}, "snapshot": {}, "items": [ { "logical_key": "assistant.nickname", "kind": "preference", "content": "本项目中用户给助手设定的昵称是“铁蛋”", } ], } ), ) request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, llmwiki_embedding=None, ) ) ) scope, recalled = await runtime.prepare_project_memory( request=request, project={"id": "project-a", "user_id": "alice", "mode": "qa"}, thread_id="thread-b", query="我给你起的昵称是什么？", ) assert scope is not None and scope["enabled"] is True assert recalled is not None assert recalled["revision"] == 1 assert recalled["items"][0]["content"] == "本项目中用户给助手设定的昵称是“铁蛋”" @pytest.mark.parametrize("status", ["rebuilding", "recovering"]) @pytest.mark.asyncio async def test_destructive_rebuild_state_is_fail_closed_for_prompt_recall( status, monkeypatch ): store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": True}), recall=AsyncMock( return_value={ "state": {"status": status, "revision": 9}, "snapshot": {"summary": "must not leak during recovery"}, "items": [{"content": "stale fact"}], } ), ) request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, llmwiki_embedding=None, ) ) ) scope, recalled = await runtime.prepare_project_memory( request=request, project={"id": "project-a", "user_id": "alice", "mode": "qa"}, thread_id="thread-b", query="之前的结论是什么？", ) assert scope is not None and scope["enabled"] is True assert recalled is None @pytest.mark.asyncio async def test_source_invalidation_failure_propagates_before_mutation(): project = {"id": "project-a", "user_id": "alice", "mode": "qa"} projects = SimpleNamespace(project_for_thread=AsyncMock(return_value=project)) store = SimpleNamespace( invalidate_source=AsyncMock(side_effect=RuntimeError("database unavailable")) ) app = SimpleNamespace( state=SimpleNamespace( analysis_project_store=projects, analysis_project_memory_store=store, ) ) with pytest.raises(RuntimeError, match="database unavailable"): await runtime.invalidate_project_memory_before_thread_mutation( app=app, thread_id="thread-a", source_message_ids=["message-a"], ) @pytest.mark.asyncio async def test_generic_memory_switch_off_still_rebuilds_project_after_source_mutation( monkeypatch, ): project = {"id": "project-a", "user_id": "alice", "mode": "qa"} projects = SimpleNamespace( thread_ids=AsyncMock(return_value=["thread-a", "thread-b"]) ) store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": True}), ) app = SimpleNamespace( state=SimpleNamespace( analysis_project_store=projects, analysis_project_memory_store=store, ) ) monkeypatch.setattr( system_settings, "load_system_settings", lambda: SimpleNamespace(memory=SimpleNamespace(enabled=False)), ) enqueue = AsyncMock(return_value=2) monkeypatch.setattr(runtime, "enqueue_project_rebuild", enqueue) await runtime.finalize_project_memory_after_thread_mutation( app=app, project=project, thread_id="thread-a", ) projects.thread_ids.assert_awaited_once_with("project-a", "alice") enqueue.assert_awaited_once_with( app=app, project=project, thread_ids=["thread-a", "thread-b"], ) @pytest.mark.asyncio async def test_rebuild_enumeration_crash_remains_durably_recoverable(monkeypatch): store = AnalysisProjectMemoryStore() class BrokenCheckpointer: async def aget_tuple(self, _config): raise RuntimeError("checkpoint unavailable") app = SimpleNamespace( state=SimpleNamespace( analysis_project_memory_store=store, analysis_project_memory_dispatcher=None, checkpointer=BrokenCheckpointer(), ) ) project = { "id": "project-a", "user_id": "alice", "mode": "qa", "name": "Project A", "background": "", } with pytest.raises(RuntimeError, match="checkpoint unavailable"): await runtime.enqueue_project_rebuild( app=app, project=project, thread_ids=["thread-a"], ) state = await store.get_state("project-a", "alice") assert state is not None and state["status"] == "rebuilding" future = datetime.now(UTC) + timedelta(days=365) [claimed] = await store.claim_stalled_rebuilds( stale_before=future, claimed_at=future, ) assert claimed["status"] == "recovering" @pytest.mark.asyncio async def test_rebuild_enqueue_honors_authoritative_project_switch(): project = { "id": "project-a", "user_id": "alice", "mode": "qa", "name": "Project A", "background": "", } paused_store = SimpleNamespace( get_settings=AsyncMock(return_value={"enabled": False}), start_rebuild=AsyncMock(side_effect=AssertionError("must not rebuild")), ) app = SimpleNamespace( state=SimpleNamespace(analysis_project_memory_store=paused_store) ) assert await runtime.enqueue_project_rebuild( app=app, project=project, thread_ids=["thread-a"] ) == 0 paused_store.start_rebuild.assert_not_awaited() @pytest.mark.asyncio async def test_dispatcher_replays_claimed_stalled_rebuild(monkeypatch): project = { "id": "project-a", "user_id": "alice", "mode": "qa", "name": "Project A", "background": "", } store = SimpleNamespace( claim_stalled_rebuilds=AsyncMock( return_value=[ { "project_id": "project-a", "owner_user_id": "alice", "status": "recovering", } ] ), get_settings=AsyncMock(return_value={"enabled": True}), ) projects = SimpleNamespace( get=AsyncMock(return_value=project), thread_ids=AsyncMock(return_value=["thread-a", "thread-b"]), ) app = SimpleNamespace( state=SimpleNamespace( analysis_project_store=projects, analysis_project_memory_store=store, ) ) enqueue = AsyncMock(return_value=2) monkeypatch.setattr(runtime, "enqueue_project_rebuild", enqueue) dispatcher = runtime.AnalysisProjectMemoryDispatcher( store, SimpleNamespace(app=app), ) assert await dispatcher.recover_stalled_rebuilds_once() == 1 enqueue.assert_awaited_once_with( app=app, project=project, thread_ids=["thread-a", "thread-b"], ) @pytest.mark.asyncio async def test_edit_route_does_not_mutate_checkpoint_when_memory_fence_fails(): class Checkpointer: def __init__(self): self.aput = AsyncMock() async def aget_tuple(self, _config): return SimpleNamespace( checkpoint={ "channel_values": { "messages": [HumanMessage(id="message-a", content="旧内容")] }, "channel_versions": {"messages": 1}, }, metadata={}, ) def get_next_version(self, value, _channel): return int(value or 0) + 1 checkpointer = Checkpointer() project = {"id": "project-a", "user_id": "alice", "mode": "qa"} request = SimpleNamespace( app=SimpleNamespace( state=SimpleNamespace( checkpointer=checkpointer, analysis_project_store=SimpleNamespace( project_for_thread=AsyncMock(return_value=project) ), analysis_project_memory_store=SimpleNamespace( invalidate_source=AsyncMock( side_effect=RuntimeError("memory database unavailable") ) ), ) ) ) with pytest.raises(HTTPException) as exc_info: await thread_routes.edit_thread_message.__wrapped__( "thread-a", "message-a", thread_routes.EditMessageRequest(content="新内容"), request, ) assert exc_info.value.status_code == 503 checkpointer.aput.assert_not_awaited() @pytest.mark.asyncio async def test_edit_validation_after_invalidation_rebuilds_unchanged_checkpoint( monkeypatch, ): class Checkpointer: def __init__(self): self.aput = AsyncMock() async def aget_tuple(self, _config): return SimpleNamespace( checkpoint={ "channel_values": { "messages": [HumanMessage(id="message-a", content="旧内容")] }, "channel_versions": {"messages": 1}, }, metadata={}, ) def get_next_version(self, value, _channel): return int(value or 0) + 1 project = {"id": "project-a", "user_id": "alice", "mode": "qa"} invalidate = AsyncMock(return_value=project) finalize = AsyncMock() monkeypatch.setattr( thread_routes, "_invalidate_project_memory_source_before_mutation", invalidate, ) monkeypatch.setattr( thread_routes, "_finalize_project_memory_source_mutation", finalize, ) def fail_edit(_message, _content): raise TypeError("provider message is immutable") monkeypatch.setattr(thread_routes, "_set_message_content", fail_edit) checkpointer = Checkpointer() request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(checkpointer=checkpointer)) ) with pytest.raises(HTTPException) as exc_info: await thread_routes.edit_thread_message.__wrapped__( "thread-a", "message-a", thread_routes.EditMessageRequest(content="新内容"), request, ) assert exc_info.value.status_code == 422 checkpointer.aput.assert_not_awaited() finalize.assert_awaited_once_with( request, project=project, thread_id="thread-a", )