"""Persistence coverage for the standalone position-collaboration workflow.""" from __future__ import annotations import asyncio 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 intent as intent_router from app.gateway.routers import multi_agent from app.gateway.routers import position_roundtable as position_roundtable_router 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) return PositionRoundtableRepository(async_sessionmaker(engine, expire_on_commit=False)) 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" @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_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 # Ownership checks apply to delete as well. 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_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", "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_and_keep_thread_out_of_messages(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=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", }, { "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=None, 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: 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 position_roundtable_router._markdown_delivery_changed(old, old) assert position_roundtable_router._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): artifact = tmp_path / "handoff.md" artifact.write_text("上游产物正文：可供下一岗位直接参考。", encoding="utf-8") 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)) assert thread_id == "upstream-thread" assert path == "mnt/user-data/outputs/handoff.md" return artifact 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 "content" not in upstream["artifacts"][1] assert calls == [("upstream-thread", "mnt/user-data/outputs/handoff.md")] @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, } 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_cleans_failed_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) with pytest.raises(HTTPException) as exc_info: await multi_agent._create_thread_direct( SimpleNamespace(), metadata={"thread_type": "roundtable"}, ) assert exc_info.value.status_code == 500 assert len(store.created) == 1 thread_id, create_kwargs = store.created[0] assert create_kwargs["metadata"] == {"thread_type": "roundtable"} assert store.deleted == [thread_id] assert checkpointer.deleted == [thread_id] @pytest.mark.asyncio async def test_direct_thread_creation_cleans_metadata_when_cancelled(monkeypatch): store = _ThreadStoreProbe() checkpointer = _FailingCheckpointer(asyncio.CancelledError()) monkeypatch.setattr(gateway_deps, "get_thread_store", lambda request: store) monkeypatch.setattr(gateway_deps, "get_checkpointer", lambda request: checkpointer) with pytest.raises(asyncio.CancelledError): await multi_agent._create_thread_direct( SimpleNamespace(), metadata={"thread_type": "roundtable"}, ) thread_id, _ = store.created[0] assert store.deleted == [thread_id] assert checkpointer.deleted == [thread_id]