"""Persistence coverage for the standalone position-collaboration workflow.""" from __future__ import annotations import asyncio import importlib.util import json import zipfile from pathlib import Path from types import SimpleNamespace from typing import Any import pytest import pytest_asyncio from fastapi import HTTPException from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine from app.gateway import deps as gateway_deps from app.gateway import services as gateway_services from app.gateway.app import create_app from app.gateway.routers import _roundtable_seed as roundtable_seed from app.gateway.routers import intent as intent_router from app.gateway.routers import multi_agent from app.gateway.routers import position_roundtable as position_roundtable_router from app.gateway.routers._roundtable_seed import ( POSITION_INTENT_AGENT_ID, required_agent_ids, ) from deerflow.persistence.base import Base from deerflow.persistence.position_roundtable import PositionRoundtableRepository from deerflow.persistence.thread_meta.sql import ThreadMetaRepository @pytest_asyncio.fixture async def store(tmp_path): engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path / 'position-roundtable.db'}") async with engine.begin() as conn: # Register both tables with Base.metadata before creating the temporary DB. import deerflow.persistence.position_roundtable.model # noqa: F401 await conn.run_sync(Base.metadata.create_all) try: yield PositionRoundtableRepository(async_sessionmaker(engine, expire_on_commit=False)) finally: await engine.dispose() def test_reject_node_request_allows_an_optional_reason(): """A rejection must work even when the reviewer leaves the reason blank.""" request = position_roundtable_router.RejectNodeRequest() assert request.title == "产物驳回" assert request.reason == "" def test_position_roundtable_router_has_intranet_compatibility_prefix(): """The workspace must survive proxies that only expose /api/multi-agent.""" app = create_app() registrations = { (route.include_context.prefix, route.include_context.include_in_schema) for route in app.routes if getattr(route, "original_router", None) is position_roundtable_router.router } assert ("/api", True) in registrations assert ("/api/multi-agent", False) in registrations assert position_roundtable_router.router.prefix == "/position-roundtable" def test_position_intent_agent_is_a_seeded_functional_agent(): """Fresh deployments must materialize the dedicated position intent agent.""" assert POSITION_INTENT_AGENT_ID in set(required_agent_ids()) def test_action_plan_seed_uses_a_model_derived_action_type_and_nine_field_contract(): """Action types are semantic names, while the remaining task fields stay compatible.""" action_assets = roundtable_seed._ASSETS_DIR / roundtable_seed.ACTION_PLAN_AGENT_ID soul = (action_assets / "SOUL.md").read_text(encoding="utf-8") config = (action_assets / "config.yaml").read_text(encoding="utf-8") assert "position-roundtable-action-create" in config for field in ( "actionName", "actionDescription", "actionType", "publishChannel", "infoFormat", "responsiblePosition", "startTime", "endTime", "taskId", ): assert field in soul assert '"actionType": "行动类型名称"' in soul assert "自行分析归纳" in soul assert "/task/action/add" in soul assert "批量新增" in soul @pytest.mark.asyncio async def test_action_plan_prompt_derives_action_type_instead_of_using_frontend_values(): """Only non-type fields keep their task-system dictionary values.""" dictionaries = position_roundtable_router.ActionPlanDictionaries( publish_channels=[{"label": "微信公众号", "value": "11"}], # Legacy callers may still include this field. It must be ignored rather # than become an actionType candidate in the planner prompt. action_types=[{"label": "旧类型", "value": "21"}], info_formats=[{"label": "图文", "value": "31"}], responsible_positions=[{"label": "规划岗", "value": "41"}], ) prompt = await position_roundtable_router._build_builtin_prompt( kind="action_plan", session={ "task_snapshot": {"title": "任务 A", "description": "生成行动"}, "summary_snapshot": {"status": "done", "content": "方案总结正文"}, }, nodes=[], action_task_id=1001, action_dictionaries=dictionaries, ) assert "当前 taskId：1001" in prompt assert "微信公众号 → 11" in prompt assert "图文 → 31" in prompt assert "规划岗 → 41" in prompt assert "actionType 必须由你根据方案总结和岗位交付自行分析归纳" in prompt assert "行动类型（actionType）" not in prompt assert "旧类型" not in prompt assert "id、title、goal" in prompt @pytest.mark.asyncio async def test_action_plan_prompt_keeps_analyzed_type_when_task_dictionaries_are_unavailable(): """Unavailable task enums must not turn the independently inferred type into zero.""" prompt = await position_roundtable_router._build_builtin_prompt( kind="action_plan", session={ "task_snapshot": {"title": "本地演示任务"}, "summary_snapshot": {"status": "done", "content": "方案总结正文"}, }, nodes=[], action_task_id="abc123", action_dictionaries=position_roundtable_router.ActionPlanDictionaries(), ) assert "当前三类任务系统字典未获取完整" in prompt assert "actionType 仍须由你分析并填写中文类型名称" in prompt assert "不得调用 Skill" in prompt assert "0 占位" in prompt @pytest.mark.asyncio async def test_action_plan_prompt_allows_direct_planning_without_a_summary(): """Removing the summary step must leave the planner with a usable direct path.""" prompt = await position_roundtable_router._build_builtin_prompt( kind="action_plan", session={ "task_snapshot": {"title": "直接规划任务", "description": "基于岗位交付生成行动"}, }, nodes=[], ) assert "已生效方案总结（如有）" in prompt assert "本轮未生成方案总结" in prompt assert "直接依据下方有效岗位交付、任务信息和已确认意图进行归纳" in prompt @pytest.mark.asyncio async def test_action_plan_prepare_allows_a_missing_summary_snapshot(monkeypatch): """The action planner may start as soon as all position deliveries are done.""" session = { "task_snapshot": {"title": "直接规划任务"}, "chain_snapshot": {"intentPositionId": "intent"}, } nodes = [ { "node_key": "stage-1-seat-1", "status": "done", "position_id": "planning", "revision": 1, "artifact_manifest": [ { "name": "岗位交付.md", "path": "mnt/user-data/outputs/岗位交付.md", "content": "岗位交付内容", } ], } ] class Store: async def list_nodes(self, session_id, user_id): assert session_id == "session-direct" assert user_id == "user-a" return nodes async def fake_require_writable_session(request, session_id): assert session_id == "session-direct" return Store(), "user-a", session monkeypatch.setattr(position_roundtable_router, "_require_writable_session", fake_require_writable_session) prepared = await position_roundtable_router._prepare_builtin_run( object(), "session-direct", kind="action_plan", ) assert prepared.agent_id == position_roundtable_router.ACTION_PLAN_AGENT_ID assert prepared.thread_id is None assert "本轮未生成方案总结" in prepared.prompt def test_action_plan_prepare_keeps_a_local_mock_task_id_without_numeric_coercion(): """A local demo id must not stop the action-plan conversation at prepare time.""" body = position_roundtable_router.PrepareBuiltinRunRequest(task_id="abc123") assert body.task_id == "abc123" def test_action_batch_skill_accepts_a_text_action_type_in_the_nine_field_array(tmp_path): """The batch Skill preserves the model-derived action type name unchanged.""" archive_path = roundtable_seed._ASSETS_DIR.parents[3] / "skill-packages" / "position-roundtable-action-create.skill" with zipfile.ZipFile(archive_path) as archive: archive.extractall(tmp_path / "skill-package") script_path = tmp_path / "skill-package" / "position-roundtable-action-create" / "scripts" / "batch_add_actions.py" spec = importlib.util.spec_from_file_location("position_action_batch_skill", script_path) assert spec and spec.loader module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) valid = [{ "actionName": "发布内容", "actionDescription": "向目标渠道发布审核通过的信息。", "actionType": "内容发布", "publishChannel": "2", "infoFormat": "3", "responsiblePosition": "4", "startTime": "2026-08-12T09:00:00+08:00", "endTime": "2026-08-12T18:00:00+08:00", "taskId": 1001, }] source = tmp_path / "action-plan-subtasks.json" source.write_text(json.dumps(valid, ensure_ascii=False), encoding="utf-8") assert module.read_actions(source, 1001) == [{ **valid[0], "publishChannel": 2, "infoFormat": 3, "responsiblePosition": 4, }] source.write_text(json.dumps([{**valid[0], "id": "legacy"}], ensure_ascii=False), encoding="utf-8") with pytest.raises(module.ActionInputError, match="多余字段"): module.read_actions(source, 1001) def test_position_intent_prompt_keeps_the_original_step_one_guardrails(): """Dedicated schema must not regress into solutioning or gatekeeping.""" prompt = ( roundtable_seed._ASSETS_DIR / POSITION_INTENT_AGENT_ID / "SOUL.md" ).read_text(encoding="utf-8") for phrase in ( "最多 5 轮用户协助", "除 ask_clarification 外，禁止调用任何工具、技能或操作", "你都不拒绝、不评价、不设限", "不得说“不能帮你”“不在职责范围内”“请换一个问题”", "coreGoals / riskWarnings / keyPoints / strategicSignificance", "不能把“主题已识别”误判为“意图已明确”", "任何任务若缺少范围、对象、时间、决策用途、重点维度或成功标准等关键信息时，必须请求用户协助", "必须分析任务级风险", "同一轮可发起多张用户协助卡", "不得把多个问题塞进一张卡", ): assert phrase in prompt def test_fresh_intent_clarification_waits_for_the_user_before_ready(): """A same-run READY block must not consume a fresh assistance pause.""" assert intent_router._clarification_blocks_ready( intent_router.POSITION_INTENT_AGENT_ID, {"name": "ask_clarification"}, is_fresh=True, ) assert intent_router._clarification_blocks_ready( intent_router.INTENT_AGENT_ID, {"name": "ask_clarification"}, is_fresh=True, ) assert not intent_router._clarification_blocks_ready( intent_router.POSITION_INTENT_AGENT_ID, None, is_fresh=True, ) assert not intent_router._clarification_blocks_ready( intent_router.POSITION_INTENT_AGENT_ID, {"name": "ask_clarification"}, is_fresh=False, ) def test_position_intent_counts_multi_card_assistance_as_one_round(): """Five user-assistance rounds must not be consumed by five cards at once.""" messages: list[dict[str, Any]] = [ { "type": "ai", "tool_calls": [ {"id": "scope", "name": "ask_clarification", "args": {}}, {"id": "purpose", "name": "ask_clarification", "args": {}}, {"id": "time", "name": "ask_clarification", "args": {}}, ], }, {"type": "human", "content": "用户统一答复"}, { "type": "ai", "tool_calls": [ {"id": "risk", "name": "ask_clarification", "args": {}}, ], }, ] assert intent_router._count_past_clarifications(messages) == 4 assert intent_router._count_past_clarification_rounds(messages) == 2 assert intent_router._max_clarification_rounds(intent_router.INTENT_AGENT_ID) == 3 assert ( intent_router._max_clarification_rounds(intent_router.POSITION_INTENT_AGENT_ID) == 5 ) def test_position_intent_seed_repairs_its_contract_without_touching_other_agents(tmp_path, monkeypatch): """The dedicated schema prompt must replace an accidentally seeded old version.""" agents_dir = tmp_path / "agents" monkeypatch.setattr( roundtable_seed, "get_paths", lambda: SimpleNamespace(agent_dir=lambda agent_id: agents_dir / agent_id), ) position_dir = agents_dir / POSITION_INTENT_AGENT_ID position_dir.mkdir(parents=True) (position_dir / "SOUL.md").write_text("outdated prompt", encoding="utf-8") assert roundtable_seed._ensure_one(POSITION_INTENT_AGENT_ID) is True expected_soul = ( roundtable_seed._ASSETS_DIR / POSITION_INTENT_AGENT_ID / "SOUL.md" ).read_text(encoding="utf-8") assert (position_dir / "SOUL.md").read_text(encoding="utf-8") == expected_soul generic_dir = agents_dir / roundtable_seed.INTENT_AGENT_ID generic_dir.mkdir(parents=True) generic_soul = generic_dir / "SOUL.md" generic_soul.write_text("deployed generic tune", encoding="utf-8") assert roundtable_seed._ensure_one(roundtable_seed.INTENT_AGENT_ID) is False assert generic_soul.read_text(encoding="utf-8") == "deployed generic tune" def test_existing_action_planner_refreshes_the_action_type_contract(tmp_path, monkeypatch): """A deployed planner must stop requiring the retired action-type dictionary.""" agents_dir = tmp_path / "agents" monkeypatch.setattr( roundtable_seed, "get_paths", lambda: SimpleNamespace(agent_dir=lambda agent_id: agents_dir / agent_id), ) action_dir = agents_dir / roundtable_seed.ACTION_PLAN_AGENT_ID action_dir.mkdir(parents=True) (action_dir / "SOUL.md").write_text("deployed action-plan tune", encoding="utf-8") (action_dir / "config.yaml").write_text( "id: position-action-planner\nskills:\n - existing-skill\n", encoding="utf-8", ) assert roundtable_seed._ensure_one(roundtable_seed.ACTION_PLAN_AGENT_ID) is True config = (action_dir / "config.yaml").read_text(encoding="utf-8") assert "existing-skill" in config assert "position-roundtable-action-create" in config expected_soul = ( roundtable_seed._ASSETS_DIR / roundtable_seed.ACTION_PLAN_AGENT_ID / "SOUL.md" ).read_text(encoding="utf-8") assert (action_dir / "SOUL.md").read_text(encoding="utf-8") == expected_soul @pytest.mark.asyncio async def test_intent_init_creates_thread_directly_without_loopback(monkeypatch): """Step 1 initialization must not depend on the deployed Gateway port.""" captured: dict[str, object] = {} monkeypatch.setattr(intent_router, "ensure_roundtable_functional_agents", lambda: None) async def fake_create_thread_direct(request, *, metadata): captured["request"] = request captured["metadata"] = metadata return "intent-thread-direct" monkeypatch.setattr(intent_router, "_create_thread_direct", fake_create_thread_direct) request = SimpleNamespace() result = await intent_router.init_intent( request, {"task_context": {"task": {"id": "task-001"}}}, ) assert result == { "status": "success", "agent_id": intent_router.INTENT_AGENT_ID, "thread_id": "intent-thread-direct", } assert captured["request"] is request assert captured["metadata"] == { **intent_router._ROUNDTABLE_THREAD_METADATA, "taskId": "task-001", "position_roundtable_task_id": "task-001", "position_task_id": "task-001:intelligence", "position_id": "intelligence", "position_roundtable_phase": "intent", } @pytest.mark.asyncio async def test_position_intent_init_selects_its_dedicated_agent(monkeypatch): """Position roundtable must never mutate the generic Step-1 agent contract.""" captured: dict[str, object] = {} monkeypatch.setattr(intent_router, "ensure_roundtable_functional_agents", lambda: None) async def fake_create_thread_direct(request, *, metadata): captured["metadata"] = metadata return "position-intent-thread" monkeypatch.setattr(intent_router, "_create_thread_direct", fake_create_thread_direct) result = await intent_router.init_intent( SimpleNamespace(), { "intent_agent_id": intent_router.POSITION_INTENT_AGENT_ID, "task_context": {"task": {"id": "task-position-001"}}, }, ) assert result == { "status": "success", "agent_id": intent_router.POSITION_INTENT_AGENT_ID, "thread_id": "position-intent-thread", } assert captured["metadata"] == { **intent_router._ROUNDTABLE_THREAD_METADATA, "taskId": "task-position-001", "position_roundtable_task_id": "task-position-001", "position_task_id": "task-position-001:intelligence", "position_id": "intelligence", "position_roundtable_phase": "intent", "position_roundtable_intent_agent_id": intent_router.POSITION_INTENT_AGENT_ID, } def test_position_intent_schema_is_required_for_new_snapshots_and_legacy_is_kept(): assert position_roundtable_router._has_confirmed_intent( { "coreGoals": ["形成统一处置方案"], "riskWarnings": [], "keyPoints": ["需要在本周完成"], "strategicSignificance": ["为后续岗位协同提供统一边界"], } ) assert not position_roundtable_router._has_confirmed_intent( {"coreGoals": ["形成统一处置方案"], "riskWarnings": []} ) assert position_roundtable_router._has_confirmed_intent( {"objective": "历史任务", "constraints": []} ) answer = position_roundtable_router._intent_node_answer( { "coreGoals": ["形成统一处置方案"], "riskWarnings": ["关键数据尚待确认"], "keyPoints": ["聚焦当前任务范围"], "strategicSignificance": ["支撑业务链协同推进"], } ) assert "核心目标" in answer assert "风险提示" in answer assert "关键要点" in answer assert "战略意义" in answer @pytest.mark.asyncio async def test_session_activation_freezes_chain_and_creates_nodes_once(store): created = await store.create_session( "user-a", external_task_id="task-1", task_snapshot={"title": "测试任务"}, ) await store.update_session( created["id"], "user-a", intent_thread_id="intent-thread", intent_snapshot={"objective": "识别意图"}, ) chain = { "id": "chain-a", "seats": [ {"agent_id": "agent-a", "position_id": "planning"}, {"agent_id": "agent-b", "position_id": "solution"}, ], } nodes = [ { "node_key": "stage-1-seat-1-agent-a", "stage_index": 0, "seat_index": 0, "agent_id": "agent-a", "position_id": "planning", "status": "ready", }, { "node_key": "stage-2-seat-1-agent-b", "stage_index": 1, "seat_index": 0, "agent_id": "agent-b", "position_id": "solution", "status": "locked", }, ] activated, created_nodes = await store.activate_session( created["id"], "user-a", chain_id="chain-a", chain_snapshot=chain, nodes=nodes ) assert activated["status"] == "active" assert activated["chain_snapshot"] == chain assert [node["status"] for node in created_nodes] == ["ready", "locked"] # Repeated activate calls must return the first frozen snapshot/nodes unchanged. repeated, repeated_nodes = await store.activate_session( created["id"], "user-a", chain_id="other", chain_snapshot={"id": "other"}, nodes=[] ) assert repeated["chain_id"] == "chain-a" assert [node["node_key"] for node in repeated_nodes] == [node["node_key"] for node in created_nodes] @pytest.mark.asyncio async def test_conversation_snapshots_round_trip_and_history_delete_removes_nodes(store): """数据库恢复副本必须覆盖三个对话阶段、岗位节点和未发送草稿。""" created = await store.create_session( "user-a", external_task_id="task-persisted", task_snapshot={"title": "持久化测试"}, ) intent_conversation = { "messages": [{"id": "intent-user-1", "type": "human", "content": "请分析任务"}], "timeline": [{"kind": "message", "message_id": "intent-user-1"}], "draft": "尚未发送的意图补充", } summary_conversation = { "messages": [{"id": "summary-ai-1", "type": "ai", "content": "方案总结"}], "draft": "总结修改意见", } action_conversation = { "messages": [{"id": "action-ai-1", "type": "ai", "content": "行动规划"}], "draft": "", } updated = await store.update_session( created["id"], "user-a", intent_thread_id="intent-thread", intent_conversation=intent_conversation, summary_conversation=summary_conversation, action_plan_conversation=action_conversation, ) assert updated is not None assert updated["intent_conversation"] == intent_conversation assert updated["summary_conversation"] == summary_conversation assert updated["action_plan_conversation"] == action_conversation _, nodes = await store.activate_session( created["id"], "user-a", chain_id="chain-a", chain_snapshot={"id": "chain-a"}, nodes=[ { "node_key": "stage-1-seat-1-agent-a", "stage_index": 0, "seat_index": 0, "agent_id": "agent-a", "position_id": "planning", "status": "ready", } ], ) node_conversation = { "messages": [{"id": "node-ai-1", "type": "ai", "content": "岗位答复"}], "draft": "岗位草稿", } saved_node = await store.update_node( created["id"], nodes[0]["node_key"], "user-a", conversation_snapshot=node_conversation, ) assert saved_node is not None assert saved_node["conversation_snapshot"] == node_conversation # History is partitioned by creator: another user working the same taskId # can neither read nor remove this conversation. assert await store.get_session(created["id"], "user-b") is None assert not await store.delete_session(created["id"], "user-b") assert await store.get_session(created["id"], "user-a") is not None assert await store.delete_session(created["id"], "user-a") assert await store.get_session(created["id"], "user-a") is None assert await store.list_nodes(created["id"], "user-a") is None @pytest.mark.asyncio async def test_task_scoped_history_is_partitioned_by_user_but_never_mixes_tasks(store): creator_session = await store.create_session( "user-a", external_task_id="external-task-100", task_snapshot={"id": "external-task-100", "title": "共享任务"}, ) visible_to_creator = await store.list_sessions( "user-a", external_task_id="external-task-100" ) assert [row["id"] for row in visible_to_creator] == [creator_session["id"]] # The same taskId no longer exposes another user's rows: a collaborator # starts from an empty list and only ever sees rows they created. assert await store.list_sessions("user-b", external_task_id="external-task-100") == [] collaborator_session = await store.create_session( "user-b", external_task_id="external-task-100", task_snapshot={"id": "external-task-100", "title": "协作者自己的会话"}, ) visible_to_collaborator = await store.list_sessions( "user-b", external_task_id="external-task-100" ) assert [row["id"] for row in visible_to_collaborator] == [collaborator_session["id"]] assert await store.update_session( creator_session["id"], "user-b", task_snapshot={"id": "external-task-100", "title": "协作者已更新"}, ) is None assert (await store.get_session(creator_session["id"], "user-a"))["task_snapshot"]["title"] == "共享任务" @pytest.mark.asyncio async def test_unscoped_history_keeps_original_user_isolation(store): personal = await store.create_session( "user-a", external_task_id=None, task_snapshot={"id": "legacy-card-id", "title": "个人会话"}, ) assert await store.get_session(personal["id"], "user-b") is None assert await store.list_sessions("user-b") == [] assert await store.update_session(personal["id"], "user-b", status="active") is None assert not await store.delete_session(personal["id"], "user-b") # Old client records could already carry a mock task id. The new database # marker keeps them personal after the migration; only a new explicit # route taskId creates a task-bound row. legacy_with_mock_task_id = await store.create_session( "user-a", external_task_id="legacy-mock-task-id", task_scoped=False, task_snapshot={"id": "legacy-card-id", "title": "鏃х増涓汉浼氳瘽"}, ) assert await store.get_session(legacy_with_mock_task_id["id"], "user-b") is None assert await store.list_sessions("user-b", external_task_id="legacy-mock-task-id") == [] def test_session_create_allows_missing_task_id_and_scoped_snapshots_keep_scope(): body = position_roundtable_router.SessionCreateRequest( external_task_id=" task-route-001 ", task_snapshot={"id": "wrong-id", "title": "任务"}, ) assert body.external_task_id == "task-route-001" assert position_roundtable_router._task_snapshot_for_scope( body.external_task_id, body.task_snapshot, ) == { "id": "task-route-001", "externalTaskId": "task-route-001", "title": "任务", } assert position_roundtable_router.SessionCreateRequest(task_snapshot={}).external_task_id is None assert position_roundtable_router.SessionCreateRequest(external_task_id=" ").external_task_id is None assert position_roundtable_router._task_snapshot_for_scope( "legacy-mock-task-id", {"id": "legacy-card-id", "title": "旧记录"}, task_scoped=False, ) == {"id": "legacy-card-id", "title": "旧记录"} @pytest.mark.asyncio async def test_demo_seed_is_complete_idempotent_and_user_scoped(store): payload = { "external_task_id": "position-demo-full-v1", "task_snapshot": {"title": "暑期活动"}, "chain_id": "position-demo-chain", "chain_snapshot": { "id": "position-demo-chain", "seats": [ { "agent_id": "demo-agent", "name": "演示分析智能体", "position_id": "planning", } ], }, "intent_thread_id": "frontend-placeholder", "intent_snapshot": { "objective": "形成完整宣传方案", "constraints": ["预算不少于 15 万元"], }, "intent_conversation": { "messages": [{"id": 1, "sender": "agent", "content": "意图已确认"}], "inputDraft": "", }, "summary_thread_id": "frontend-placeholder", "summary_snapshot": {"status": "done", "content": "方案总结"}, "summary_conversation": { "messages": [{"id": "summary-ai", "type": "ai", "content": "方案总结"}] }, "action_plan_thread_id": "frontend-placeholder", "action_plan_snapshot": {"status": "done", "content": "行动规划"}, "action_plan_conversation": { "messages": [{"id": "action-ai", "type": "ai", "content": "行动规划"}] }, "status": "completed", } nodes = [ { "node_key": "stage-1-seat-1-demo-agent", "stage_index": 0, "seat_index": 0, "agent_id": "demo-agent", "position_id": "planning", "thread_id": "frontend-placeholder", "status": "done", "latest_answer": "报告已完成", "artifact_manifest": [ { "name": "演示报告.md", "path": "mnt/user-data/outputs/演示报告.md", "mime_type": "text/markdown", "content": "# 演示报告\n\n完整正文", } ], "conversation_snapshot": { "messages": [{"id": "node-ai", "type": "ai", "content": "报告已完成"}], "inputDraft": "", }, "revision": 1, } ] first, first_nodes = await store.ensure_demo_session( "user-a", demo_key="full-v1", session_payload=payload, nodes=nodes, ) repeated, repeated_nodes = await store.ensure_demo_session( "user-a", demo_key="full-v1", session_payload={**payload, "task_snapshot": {"title": "不应覆盖"}}, nodes=[], ) other_user, other_nodes = await store.ensure_demo_session( "user-b", demo_key="full-v1", session_payload=payload, nodes=nodes, ) assert first["id"] == repeated["id"] assert first["task_snapshot"]["title"] == "暑期活动" assert repeated["task_snapshot"]["title"] == "暑期活动" assert first["intent_conversation"] == payload["intent_conversation"] assert first["summary_snapshot"]["thread_id"] == first["summary_thread_id"] assert first["action_plan_snapshot"]["thread_id"] == first["action_plan_thread_id"] assert first_nodes[0]["artifact_manifest"][0]["content"] == "# 演示报告\n\n完整正文" assert repeated_nodes[0]["id"] == first_nodes[0]["id"] assert other_user["id"] != first["id"] assert other_user["intent_thread_id"] != first["intent_thread_id"] assert other_nodes[0]["thread_id"] != first_nodes[0]["thread_id"] @pytest.mark.asyncio async def test_embedded_artifact_content_survives_missing_sandbox(monkeypatch): """后续岗位优先读取 SQL 正文，不能因沙箱卷丢失而只拿到文件名。""" async def unexpected_resolve(*_args, **_kwargs): raise AssertionError("embedded database content should avoid sandbox resolution") monkeypatch.setattr( position_roundtable_router, "aresolve_thread_virtual_path", unexpected_resolve, ) artifacts, consumed = await position_roundtable_router._artifact_refs_with_content( "removed-thread", [ { "name": "report.md", "path": "mnt/user-data/outputs/report.md", "mime_type": "text/markdown", "content": "# 数据库恢复报告\n\n完整正文", } ], 10_000, ) assert artifacts[0]["content"] == "# 数据库恢复报告\n\n完整正文" assert consumed == len("# 数据库恢复报告\n\n完整正文") def test_task_external_context_reaches_agent_runtime_context(): """岗位会商的左侧任务快照不能在网关白名单中丢失。""" task_context = { "task": { "title": "重点区域态势研判与协同处置方案", "description": "梳理风险信号并形成协同处置建议", } } config: dict = {"configurable": {}, "context": {}} gateway_services.merge_run_context_overrides( config, {"task_external_context": task_context}, ) assert config["configurable"]["task_external_context"] == task_context assert config["context"]["task_external_context"] == task_context def test_position_intent_thread_metadata_keeps_parent_and_child_task_ids(): metadata = intent_router._intent_thread_metadata( { "task": {"id": "task-100"}, "current_position": {"id": "intelligence"}, } ) assert metadata["taskId"] == "task-100" assert metadata["position_roundtable_task_id"] == "task-100" assert metadata["position_task_id"] == "task-100:intelligence" assert metadata["position_id"] == "intelligence" def test_chain_node_activation_rejects_missing_agent_ids(): chain = { "id": "chain-a", "seats": [ {"agent_id": "agent-a", "name": "有效智能体", "position_id": "planning"}, {"agent_id": "missing-agent", "name": "缺失智能体", "position_id": "solution"}, ], } with pytest.raises(HTTPException) as error: position_roundtable_router._chain_nodes( chain, available_agent_ids={"agent-a"}, ) assert error.value.status_code == 400 assert "缺失智能体" in str(error.value.detail) assert "缺少可用智能体" in str(error.value.detail) def test_intent_position_nodes_complete_from_confirmed_intent_and_unlock_next_stage(): nodes = [ { "node_key": "stage-1-seat-1-intent-agent", "stage_index": 0, "seat_index": 0, "agent_id": "intent-agent", "position_id": "intelligence", "status": "ready", }, { "node_key": "stage-2-seat-1-plan-agent", "stage_index": 1, "seat_index": 0, "agent_id": "plan-agent", "position_id": "planning", "status": "locked", }, ] primed = position_roundtable_router._prime_intent_nodes( nodes, {"objective": "识别用户想要形成协同方案", "constraints": ["两天内完成"]}, ) assert [node["status"] for node in primed] == ["done", "ready"] assert primed[0]["revision"] == 1 assert "识别用户想要形成协同方案" in primed[0]["latest_answer"] def test_all_nodes_delivered_requires_markdown_for_non_intent_nodes(): assert not position_roundtable_router._all_nodes_delivered( [ { "position_id": "planning", "status": "done", "artifact_manifest": [{"name": "plain.txt", "path": "mnt/user-data/outputs/plain.txt"}], } ] ) assert position_roundtable_router._all_nodes_delivered( [ {"position_id": "intelligence", "status": "done", "artifact_manifest": []}, { "position_id": "planning", "status": "done", "artifact_manifest": [{"name": "report.md", "path": "mnt/user-data/outputs/report.md"}], }, ] ) @pytest.mark.asyncio async def test_builtin_markdown_fallback_writes_visible_summary_to_thread_outputs(tmp_path, monkeypatch): target = tmp_path / "outputs" / "岗位会商方案总结报告.md" async def resolve(thread_id: str, path: str): assert thread_id == "summary-thread" assert path == "mnt/user-data/outputs/岗位会商方案总结报告.md" return target monkeypatch.setattr(position_roundtable_router, "aresolve_thread_virtual_path", resolve) artifact = await position_roundtable_router._write_builtin_markdown_fallback( thread_id="summary-thread", kind="summary", content="# 岗位会商方案总结报告\n\n这是对话中生成的完整报告正文。", ) assert target.read_text(encoding="utf-8") == "# 岗位会商方案总结报告\n\n这是对话中生成的完整报告正文。\n" assert artifact["name"] == "岗位会商方案总结报告.md" assert artifact["path"] == "mnt/user-data/outputs/岗位会商方案总结报告.md" assert artifact["mime_type"] == "text/markdown" assert artifact["size_bytes"] == target.stat().st_size @pytest.mark.asyncio async def test_builtin_summary_completion_persists_fallback_markdown_when_agent_skips_write_file(monkeypatch): node = { "id": "node-1", "session_id": "session-a", "node_key": "intent-node", "stage_index": 0, "seat_index": 0, "agent_id": "intent-agent", "position_id": "intelligence", "thread_id": "intent-thread", "status": "done", "latest_answer": "已确认任务意图", "artifact_manifest": [], "revision": 1, "upstream_revision": None, "rejection_count": 0, "last_rejection": None, "invalidated_by": None, } session = { "id": "session-a", "external_task_id": "task-fallback", "status": "active", "task_snapshot": {}, "chain_id": "chain-a", "chain_snapshot": {}, "intent_thread_id": "intent-thread", "intent_snapshot": {"objective": "形成方案", "constraints": []}, "summary_thread_id": None, "summary_snapshot": None, "action_plan_thread_id": None, "action_plan_snapshot": None, } class StubStore: updates: dict | None = None async def get_session(self, session_id, user_id): assert (session_id, user_id) == ("session-a", "user-a") return session async def list_nodes(self, session_id, user_id): assert (session_id, user_id) == ("session-a", "user-a") return [node] async def update_session(self, session_id, user_id, **updates): assert (session_id, user_id) == ("session-a", "user-a") self.updates = updates return {**session, **updates} store = StubStore() async def write_fallback(**kwargs): assert kwargs["thread_id"] == "summary-thread" assert kwargs["kind"] == "summary" assert kwargs["content"] == "# 方案总结\n\n完整报告正文" return { "name": "岗位会商方案总结报告.md", "path": "mnt/user-data/outputs/岗位会商方案总结报告.md", "mime_type": "text/markdown", "modified_at": "2026-07-23T00:00:00+00:00", "size_bytes": 42, } monkeypatch.setattr(position_roundtable_router, "_write_builtin_markdown_fallback", write_fallback) request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=store)), state=SimpleNamespace(user=SimpleNamespace(id="user-a")), ) body = position_roundtable_router.CompleteBuiltinRunRequest( thread_id="summary-thread", content="# 方案总结\n\n完整报告正文", artifact_manifest=[], source_revisions={"intent-node": 1}, ) result = await position_roundtable_router._complete_builtin_run( request, "session-a", kind="summary", body=body, ) assert store.updates is not None snapshot = store.updates["summary_snapshot"] assert snapshot["content"] == "# 方案总结\n\n完整报告正文" assert snapshot["primary_artifact"]["name"] == "岗位会商方案总结报告.md" assert result.summary_snapshot["primary_artifact"]["path"] == ( "mnt/user-data/outputs/岗位会商方案总结报告.md" ) @pytest.mark.asyncio async def test_node_updates_are_user_scoped_without_task_id(store): created = await store.create_session("user-a", external_task_id=None, task_snapshot={}) _, nodes = await store.activate_session( created["id"], "user-a", chain_id="chain-a", chain_snapshot={"id": "chain-a"}, nodes=[ { "node_key": "stage-1-seat-1-agent-a", "stage_index": 0, "seat_index": 0, "agent_id": "agent-a", "position_id": "planning", "status": "ready", } ], ) assert len(nodes) == 1 updated = await store.update_node( created["id"], nodes[0]["node_key"], "user-a", thread_id="langgraph-thread-a", status="done", latest_answer="节点回答", artifact_manifest=[{"name": "产物.md"}], upstream_revision={"intent": 1}, increment_revision=True, ) assert updated is not None assert updated["thread_id"] == "langgraph-thread-a" assert updated["latest_answer"] == "节点回答" assert updated["artifact_manifest"] == [{"name": "产物.md"}] assert updated["revision"] == 1 assert await store.get_node(created["id"], nodes[0]["node_key"], "user-b") is None @pytest.mark.asyncio async def test_reject_node_marks_current_and_invalidates_later_stages(store): created = await store.create_session( "user-a", external_task_id="task-reject", task_snapshot={} ) _, nodes = await store.activate_session( created["id"], "user-a", chain_id="chain-a", chain_snapshot={"id": "chain-a"}, nodes=[ { "node_key": "stage-1-seat-1-agent-a", "stage_index": 0, "seat_index": 0, "agent_id": "agent-a", "position_id": "planning", "status": "ready", }, { "node_key": "stage-2-seat-1-agent-b", "stage_index": 1, "seat_index": 0, "agent_id": "agent-b", "position_id": "solution", "status": "ready", }, { "node_key": "stage-3-seat-1-agent-c", "stage_index": 2, "seat_index": 0, "agent_id": "agent-c", "position_id": "command", "status": "ready", }, ], ) await store.update_node( created["id"], nodes[0]["node_key"], "user-a", status="done", latest_answer="第一阶段产物", increment_revision=True, ) await store.update_node( created["id"], nodes[1]["node_key"], "user-a", status="done", latest_answer="第二阶段产物", increment_revision=True, ) rejected = await store.reject_node( created["id"], nodes[0]["node_key"], "user-a", title="依据不足", reason="第一阶段判断依据需要补充。", rejected_by="user-a", ) assert rejected is not None _, rejected_nodes = rejected by_key = {node["node_key"]: node for node in rejected_nodes} first = by_key[nodes[0]["node_key"]] second = by_key[nodes[1]["node_key"]] third = by_key[nodes[2]["node_key"]] assert first["status"] == "rejected" assert first["rejection_count"] == 1 assert first["last_rejection"]["title"] == "依据不足" assert first["invalidated_by"] is None assert second["status"] == "stale" assert second["latest_answer"] == "第二阶段产物" assert second["invalidated_by"] == nodes[0]["node_key"] assert third["status"] == "locked" assert third["invalidated_by"] == nodes[0]["node_key"] @pytest.mark.asyncio async def test_builtin_summary_and_action_plan_snapshots_are_invalidated_by_reject(store): created = await store.create_session( "user-a", external_task_id="task-builtin-invalidation", task_snapshot={ "title": "测试任务", "description": "形成协同方案", "direction": "态势研判", }, ) await store.update_session( created["id"], "user-a", intent_snapshot={"objective": "明确处置意图", "constraints": ["时间紧"]}, ) chain = { "id": "chain-a", "seats": [ {"agent_id": "agent-a", "name": "研判智能体", "position_id": "planning"}, {"agent_id": "agent-b", "name": "处置智能体", "position_id": "solution"}, ], } _, nodes = await store.activate_session( created["id"], "user-a", chain_id="chain-a", chain_snapshot=chain, nodes=[ { "node_key": "stage-1-seat-1-agent-a", "stage_index": 0, "seat_index": 0, "agent_id": "agent-a", "position_id": "planning", "status": "ready", }, { "node_key": "stage-2-seat-1-agent-b", "stage_index": 1, "seat_index": 0, "agent_id": "agent-b", "position_id": "solution", "status": "ready", }, ], ) for node in nodes: await store.update_node( created["id"], node["node_key"], "user-a", status="done", latest_answer=f"{node['agent_id']} 产物", increment_revision=True, ) session = await store.get_session(created["id"], "user-a") assert session is not None nodes = await store.list_nodes(created["id"], "user-a") or [] revisions = position_roundtable_router._source_revisions(nodes) summary = position_roundtable_router._build_builtin_snapshot( kind="summary", thread_id="summary-thread", content="# 岗位会商方案总结报告\n\n研判智能体的关键结论。", artifacts=[ { "name": "岗位会商方案总结报告.md", "path": "mnt/user-data/outputs/岗位会商方案总结报告.md", "mime_type": "text/markdown", "modified_at": "2026-07-21T00:00:00+00:00", "size_bytes": 100, } ], source_revisions=revisions, model="test-model", ) assert "岗位会商方案总结报告" in summary["content"] session = await store.update_session( created["id"], "user-a", summary_thread_id="summary-thread", summary_snapshot=summary ) assert session is not None action_plan = position_roundtable_router._build_builtin_snapshot( kind="action_plan", thread_id="action-plan-thread", content="# 岗位会商行动规划\n\n## 子任务拆解", artifacts=[ { "name": "行动规划报告.md", "path": "mnt/user-data/outputs/行动规划报告.md", "mime_type": "text/markdown", "modified_at": "2026-07-21T00:01:00+00:00", "size_bytes": 100, }, { "name": "action-plan-subtasks.json", "path": "mnt/user-data/outputs/action-plan-subtasks.json", "mime_type": "application/json", "modified_at": "2026-07-21T00:01:00+00:00", "size_bytes": 200, }, ], source_revisions=revisions, model="test-model", ) session = await store.update_session( created["id"], "user-a", action_plan_thread_id="action-plan-thread", action_plan_snapshot=action_plan, status="completed", ) assert session is not None assert session["summary_snapshot"]["status"] == "done" assert session["action_plan_snapshot"]["status"] == "done" rejected = await store.reject_node( created["id"], nodes[0]["node_key"], "user-a", title="重做第一阶段", reason="前序依据变化", rejected_by="user-a", ) assert rejected is not None rejected_session, _ = rejected assert rejected_session["status"] == "active" assert rejected_session["summary_snapshot"]["status"] == "stale" assert rejected_session["summary_snapshot"]["invalidated_by"] == nodes[0]["node_key"] assert rejected_session["action_plan_snapshot"]["status"] == "stale" assert rejected_session["action_plan_snapshot"]["invalidated_by"] == nodes[0]["node_key"] def test_rejected_or_stale_node_requires_a_new_markdown_delivery() -> None: from deerflow.persistence.position_roundtable.delivery import markdown_delivery_changed old = [ { "name": "report.md", "path": "mnt/user-data/outputs/report.md", "modified_at": "2026-07-21T00:00:00+00:00", "size_bytes": 100, } ] assert not markdown_delivery_changed(old, old) assert markdown_delivery_changed( old, [{**old[0], "modified_at": "2026-07-21T00:05:00+00:00", "size_bytes": 120}], ) def test_activation_requires_a_structured_intent_snapshot() -> None: assert not position_roundtable_router._has_confirmed_intent({"objective": "只有目标"}) assert not position_roundtable_router._has_confirmed_intent({"constraints": []}) assert position_roundtable_router._has_confirmed_intent( {"objective": "明确任务目标", "constraints": []} ) @pytest.mark.asyncio async def test_handoff_context_inlines_only_upstream_thread_text_artifacts(tmp_path, monkeypatch): upstream_dir = tmp_path / "up" / "outputs" upstream_dir.mkdir(parents=True) artifact = upstream_dir / "handoff.md" artifact.write_text("上游产物正文：可供下一岗位直接参考。", encoding="utf-8") mirror_root = tmp_path / "current" / "workspace" / "upstream" calls: list[tuple[str, str]] = [] class StubStore: async def get_session(self, session_id, user_id): assert (session_id, user_id) == ("session-a", "user-a") return {"task_snapshot": {"title": "测试任务"}, "intent_snapshot": {"goal": "方案分析"}} async def list_nodes(self, session_id, user_id): assert (session_id, user_id) == ("session-a", "user-a") return [ { "node_key": "stage-1-seat-1-agent-a", "stage_index": 0, "seat_index": 0, "agent_id": "agent-a", "position_id": "planning", "thread_id": "upstream-thread", "status": "done", "revision": 1, "latest_answer": "上游问答结果", "artifact_manifest": [ { "name": "handoff.md", "path": "mnt/user-data/outputs/handoff.md", "mime_type": "text/markdown", }, { "name": "outside.md", "path": "mnt/user-data/uploads/outside.md", "mime_type": "text/markdown", }, ], }, { "node_key": "stage-2-seat-1-agent-b", "stage_index": 1, "seat_index": 0, "agent_id": "agent-b", "position_id": "solution", "thread_id": "current-thread", "status": "ready", "revision": 0, "latest_answer": None, "artifact_manifest": [], }, { "node_key": "stage-1-seat-2-agent-stale", "stage_index": 0, "seat_index": 1, "agent_id": "agent-stale", "position_id": "planning", "thread_id": "stale-thread", "status": "stale", "revision": 1, "latest_answer": "失效上游产物", "artifact_manifest": [], }, ] async def resolve(thread_id: str, path: str): calls.append((thread_id, path)) if thread_id == "upstream-thread": assert path == "mnt/user-data/outputs/handoff.md" return artifact assert thread_id == "current-thread" assert path == "mnt/user-data/workspace/upstream/handoff.md" return mirror_root / "handoff.md" monkeypatch.setattr(position_roundtable_router, "aresolve_thread_virtual_path", resolve) request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=StubStore())), state=SimpleNamespace(user=SimpleNamespace(id="user-a")), ) context = await position_roundtable_router._handoff_context( request, "session-a", "stage-2-seat-1-agent-b" ) upstream = context["upstream_nodes"][0] assert len(context["upstream_nodes"]) == 1 assert upstream["latest_answer"] == "上游问答结果" assert upstream["artifacts"][0]["content"] == "上游产物正文：可供下一岗位直接参考。" assert upstream["artifacts"][0]["upstream_path"] == "mnt/user-data/outputs/handoff.md" assert upstream["artifacts"][0]["path"] == "mnt/user-data/workspace/upstream/handoff.md" assert upstream["artifacts"][0]["mirrored_to_sandbox"] is True assert "content" not in upstream["artifacts"][1] assert "mirrored_to_sandbox" not in upstream["artifacts"][1] assert upstream["artifacts"][1]["path"] == "mnt/user-data/uploads/outside.md" assert (mirror_root / "handoff.md").read_text(encoding="utf-8") == "上游产物正文：可供下一岗位直接参考。" assert "/mnt/user-data/workspace/upstream/handoff.md" in context["instructions"] assert set(calls) == { ("upstream-thread", "mnt/user-data/outputs/handoff.md"), ("current-thread", "mnt/user-data/workspace/upstream/handoff.md"), } @pytest.mark.asyncio async def test_handoff_mirror_falls_back_to_persisted_content_and_skips_unsafe_paths( tmp_path, monkeypatch ): mirror_root = tmp_path / "mirror" resolved: list[tuple[str, str]] = [] class StubStore: async def get_session(self, session_id, user_id): assert (session_id, user_id) == ("session-a", "user-a") return {"task_snapshot": {"title": "测试任务"}, "intent_snapshot": {}} async def list_nodes(self, session_id, user_id): return [ { "node_key": "stage-1-seat-1-agent-a", "stage_index": 0, "seat_index": 0, "agent_id": "agent-a", "position_id": "planning", "thread_id": "upstream-thread", "status": "done", "revision": 1, "latest_answer": "上游问答结果", "artifact_manifest": [ { "name": "gone.md", "path": "mnt/user-data/outputs/gone.md", "mime_type": "text/markdown", "content": "数据库恢复正文", }, { "name": "escape.md", "path": "mnt/user-data/outputs/../escape.md", "mime_type": "text/markdown", "content": "不应被镜像", }, ], }, { "node_key": "stage-2-seat-1-agent-b", "stage_index": 1, "seat_index": 0, "agent_id": "agent-b", "position_id": "solution", "thread_id": "current-thread", "status": "ready", "revision": 0, "latest_answer": None, "artifact_manifest": [], }, ] missing_source = tmp_path / "missing-source" async def resolve(thread_id: str, path: str): resolved.append((thread_id, path)) if thread_id == "upstream-thread": return missing_source / "not-materialized.md" assert thread_id == "current-thread" return mirror_root / Path(path).name monkeypatch.setattr(position_roundtable_router, "aresolve_thread_virtual_path", resolve) request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=StubStore())), state=SimpleNamespace(user=SimpleNamespace(id="user-a")), ) context = await position_roundtable_router._handoff_context( request, "session-a", "stage-2-seat-1-agent-b" ) artifacts = context["upstream_nodes"][0]["artifacts"] assert artifacts[0]["path"] == "mnt/user-data/workspace/upstream/gone.md" assert artifacts[0]["mirrored_to_sandbox"] is True assert (mirror_root / "gone.md").read_text(encoding="utf-8") == "数据库恢复正文" # 持久化正文同时内嵌，沙箱磁盘丢失也不影响下一岗位读到内容。 assert artifacts[0]["content"] == "数据库恢复正文" assert artifacts[1]["path"] == "mnt/user-data/outputs/../escape.md" assert "mirrored_to_sandbox" not in artifacts[1] assert not (mirror_root / "escape.md").exists() assert all(".." not in path for _, path in resolved) @pytest.mark.asyncio async def test_archived_session_rejects_node_writes_and_keeps_history_readable(): class ArchivedStore: async def get_session(self, session_id, user_id): assert (session_id, user_id) == ("session-a", "user-a") return {"id": session_id, "status": "archived"} async def get_node(self, session_id, node_key, user_id): return { "node_key": node_key, "status": "ready", "thread_id": None, } request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=ArchivedStore())), state=SimpleNamespace(user=SimpleNamespace(id="user-a")), ) with pytest.raises(HTTPException) as error: await position_roundtable_router.bind_node_thread( request, "session-a", "stage-1-seat-1-agent-a", position_roundtable_router.NodeThreadRequest(thread_id="thread-a"), ) assert error.value.status_code == 409 assert "已归档" in str(error.value.detail) @pytest.mark.asyncio async def test_locked_node_cannot_bind_a_thread_before_its_stage_is_ready(): class LockedStore: async def get_session(self, session_id, user_id): return {"id": session_id, "status": "active"} async def get_node(self, session_id, node_key, user_id): return { "node_key": node_key, "status": "locked", "thread_id": None, } async def apply_node_command( self, session_id, node_key, user_id, *, command_id, command_type, payload=None, expected_session_version=None, ): # Mirrors the in-lock rejection a locked node produces in the real store. return { "ok": False, "http_status": 409, "detail": "上游节点尚未完成，当前节点未解锁", } request = SimpleNamespace( app=SimpleNamespace(state=SimpleNamespace(position_roundtable_store=LockedStore())), state=SimpleNamespace(user=SimpleNamespace(id="user-a")), ) with pytest.raises(HTTPException) as error: await position_roundtable_router.bind_node_thread( request, "session-a", "stage-2-seat-1-agent-b", position_roundtable_router.NodeThreadRequest(thread_id="thread-b"), ) assert error.value.status_code == 409 assert "未解锁" in str(error.value.detail) class _ThreadMetaSessionProbe: def __init__(self, events: list[str]) -> None: self._events = events async def __aenter__(self): return self async def __aexit__(self, exc_type, exc, tb) -> None: return None def add(self, row) -> None: self._events.append(f"add:{row.thread_id}") async def commit(self) -> None: self._events.append("commit") async def refresh(self, row) -> None: self._events.append(f"refresh:{row.thread_id}") @pytest.mark.asyncio async def test_thread_meta_create_never_reads_back_after_commit(): """No caller may depend on a post-commit re-SELECT of the new row. On MySQL behind a read/write-splitting endpoint that read can land on a lagging replica, come back empty, and raise "Could not refresh instance". """ events: list[str] = [] repo = ThreadMetaRepository(lambda: _ThreadMetaSessionProbe(events)) # type: ignore[arg-type] direct = await repo.create( "roundtable-thread", user_id=None, metadata={"thread_type": "roundtable"}, ) ordinary = await repo.create("ordinary-thread", user_id=None, display_name="对话") assert events == [ "add:roundtable-thread", "commit", "add:ordinary-thread", "commit", ] # The returned dict is assembled from the in-memory row, so it must still be # complete for every column -- including ones no caller passed in. assert direct["status"] == "idle" assert direct["metadata"] == {"thread_type": "roundtable"} assert direct["created_at"] and direct["updated_at"] assert ordinary["status"] == "idle" assert ordinary["display_name"] == "对话" class _ThreadStoreProbe: def __init__(self) -> None: self.created: list[tuple[str, dict[str, Any]]] = [] self.deleted: list[str] = [] async def create(self, thread_id: str, **kwargs) -> dict[str, Any]: self.created.append((thread_id, kwargs)) return {"thread_id": thread_id} async def delete(self, thread_id: str) -> None: self.deleted.append(thread_id) class _FailingCheckpointer: def __init__(self, exc: BaseException) -> None: self._exc = exc self.deleted: list[str] = [] async def aput(self, *args, **kwargs): raise self._exc async def adelete_thread(self, thread_id: str) -> None: self.deleted.append(thread_id) @pytest.mark.asyncio async def test_direct_thread_creation_defers_unavailable_checkpoint(monkeypatch): store = _ThreadStoreProbe() checkpointer = _FailingCheckpointer(RuntimeError("checkpoint unavailable")) monkeypatch.setattr(gateway_deps, "get_thread_store", lambda request: store) monkeypatch.setattr(gateway_deps, "get_checkpointer", lambda request: checkpointer) thread_id = await multi_agent._create_thread_direct( SimpleNamespace(), metadata={"thread_type": "roundtable"}, ) assert len(store.created) == 1 created_id, create_kwargs = store.created[0] assert created_id == thread_id assert create_kwargs["metadata"] == {"thread_type": "roundtable"} assert store.deleted == [] assert checkpointer.deleted == [] @pytest.mark.asyncio async def test_direct_thread_creation_propagates_metadata_cancellation(monkeypatch): store = _ThreadStoreProbe() async def cancelled_create(*args, **kwargs): raise asyncio.CancelledError() monkeypatch.setattr(store, "create", cancelled_create) monkeypatch.setattr(gateway_deps, "get_thread_store", lambda request: store) with pytest.raises(asyncio.CancelledError): await multi_agent._create_thread_direct( SimpleNamespace(), metadata={"thread_type": "roundtable"}, ) assert store.created == []