董宏凯andCursor
a9d66536c3
Skip post-commit ORM refresh on threads_meta creates so concurrent MySQL read/write-split deployments no longer fail with Could not refresh instance; clean up partial roundtable threads on checkpoint failure, and ship related position-roundtable/frontend build fixes. Co-authored-by: Cursor <cursoragent@cursor.com>
1096 lines
39 KiB
Python
1096 lines
39 KiB
Python
"""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]
|