768 lines
28 KiB
Python
768 lines
28 KiB
Python
from __future__ import annotations
|
||
|
||
import os
|
||
|
||
import pytest
|
||
import pytest_asyncio
|
||
from sqlalchemy import JSON, select
|
||
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||
|
||
os.environ["DATABASE_URL"] = "postgresql+asyncpg://tester:tester@localhost:5432/pmflow_test"
|
||
|
||
from app.models.base import Base
|
||
from app.models.feishu_reply_outbox import FeishuReplyOutbox
|
||
from app.models.mounted_ai_runtime import (
|
||
MountedAgentTrace,
|
||
MountedConversationRun,
|
||
MountedSourceActivation,
|
||
MountedTopicWorkspace,
|
||
)
|
||
from app.services.mounted_ai_runtime_write_service import MountedAIRuntimeWriteService
|
||
|
||
|
||
def _use_sqlite_json_types() -> None:
|
||
for table in (
|
||
MountedTopicWorkspace.__table__,
|
||
MountedConversationRun.__table__,
|
||
MountedAgentTrace.__table__,
|
||
MountedSourceActivation.__table__,
|
||
):
|
||
for column in table.columns:
|
||
if column.name in {
|
||
"facts_json",
|
||
"goals_json",
|
||
"constraints_json",
|
||
"options_json",
|
||
"open_questions",
|
||
"pending_actions",
|
||
"bound_entities",
|
||
"metadata_json",
|
||
"context_layers_hit",
|
||
"tool_calls",
|
||
"execution_outcome",
|
||
"input_summary",
|
||
"output_summary",
|
||
"tool_trace",
|
||
}:
|
||
column.type = JSON()
|
||
|
||
|
||
@pytest_asyncio.fixture
|
||
async def db_session():
|
||
engine = create_async_engine("sqlite+aiosqlite:///:memory:")
|
||
_use_sqlite_json_types()
|
||
async with engine.begin() as conn:
|
||
await conn.run_sync(
|
||
lambda sync_conn: Base.metadata.create_all(
|
||
sync_conn,
|
||
tables=[
|
||
MountedTopicWorkspace.__table__,
|
||
MountedConversationRun.__table__,
|
||
MountedAgentTrace.__table__,
|
||
MountedSourceActivation.__table__,
|
||
FeishuReplyOutbox.__table__,
|
||
],
|
||
)
|
||
)
|
||
maker = async_sessionmaker(engine, expire_on_commit=False)
|
||
async with maker() as session:
|
||
yield session
|
||
await engine.dispose()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_begin_ingress_run_uses_thread_anchor_and_sets_runtime_context(db_session) -> None:
|
||
db_session.add(
|
||
MountedSourceActivation(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
scope_type="chat",
|
||
scope_ref="chat_1",
|
||
status="active",
|
||
cutover_mode="mounted_only",
|
||
auto_reply_enabled=True,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_1",
|
||
message_text="帮我复盘一下这个直播需求。",
|
||
sender="林宇航",
|
||
channel_ref="chat_1",
|
||
thread_context={
|
||
"thread_id": "thread_1",
|
||
"root_message_id": "om_root_1",
|
||
"reply_to_message_id": "om_reply_1",
|
||
},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (
|
||
await db_session.execute(select(MountedTopicWorkspace))
|
||
).scalar_one()
|
||
saved_run = (
|
||
await db_session.execute(select(MountedConversationRun))
|
||
).scalar_one()
|
||
|
||
assert tracking.topic_workspace.id == saved_topic.id
|
||
assert saved_topic.topic_key == "feishu:chat_1:root:om_root_1"
|
||
assert saved_topic.thread_key == "thread_1"
|
||
assert saved_topic.last_message_id == "om_message_1"
|
||
assert saved_run.status == "running"
|
||
assert saved_run.mode_before is None
|
||
assert saved_run.context_layers_hit == {
|
||
"current_message": True,
|
||
"thread": True,
|
||
"topic": False,
|
||
"source_activation": True,
|
||
}
|
||
assert saved_topic.metadata_json["source_activation"]["cutover_mode"] == "mounted_only"
|
||
assert saved_run.trigger_reason == "ingress:feishu:mounted_only"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_begin_ingress_run_reuses_canonical_source_topic_for_feishu_bot_reply_anchor(db_session) -> None:
|
||
existing_topic = MountedTopicWorkspace(
|
||
tenant_id=1,
|
||
topic_key="feishu:chat_reply_canonical:root:om_source_1",
|
||
source_system="feishu",
|
||
channel_ref="chat_reply_canonical",
|
||
root_message_id="om_source_1",
|
||
mode="plan",
|
||
status="active",
|
||
plan_status="collecting",
|
||
metadata_json={"plan_topic_action": "collecting"},
|
||
)
|
||
db_session.add(existing_topic)
|
||
db_session.add(
|
||
FeishuReplyOutbox(
|
||
tenant_id=1,
|
||
trace_id="trace_reply_canonical",
|
||
source_message_id="om_source_1",
|
||
sent_message_id="om_bot_reply_1",
|
||
chat_id="chat_reply_canonical",
|
||
message_text="上一轮 requirement-plan 选项",
|
||
status="sent",
|
||
attempts=1,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_user_reply_1",
|
||
message_text="C",
|
||
sender="林宇航",
|
||
channel_ref="chat_reply_canonical",
|
||
thread_context={
|
||
"root_message_id": "om_bot_reply_1",
|
||
"reply_to_message_id": "om_bot_reply_1",
|
||
},
|
||
source_sender_open_id="ou_reply_canonical",
|
||
)
|
||
await db_session.commit()
|
||
|
||
topic_rows = (
|
||
await db_session.execute(
|
||
select(MountedTopicWorkspace).order_by(MountedTopicWorkspace.id.asc())
|
||
)
|
||
).scalars().all()
|
||
saved_run = (
|
||
await db_session.execute(select(MountedConversationRun))
|
||
).scalar_one()
|
||
|
||
assert tracking is not None
|
||
assert len(topic_rows) == 1
|
||
assert tracking.topic_workspace.id == int(existing_topic.id)
|
||
assert tracking.topic_workspace.topic_key == "feishu:chat_reply_canonical:root:om_source_1"
|
||
assert tracking.topic_workspace.root_message_id == "om_source_1"
|
||
assert saved_run.topic_workspace_id == int(existing_topic.id)
|
||
assert saved_run.context_layers_hit == {
|
||
"current_message": True,
|
||
"thread": True,
|
||
"topic": True,
|
||
"source_activation": False,
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_begin_ingress_run_reuses_existing_source_topic_for_direct_reply_anchor(db_session) -> None:
|
||
existing_topic = MountedTopicWorkspace(
|
||
tenant_id=1,
|
||
topic_key="feishu:chat_reply_source:message:om_source_2",
|
||
source_system="feishu",
|
||
channel_ref="chat_reply_source",
|
||
root_message_id="om_source_2",
|
||
mode="plan",
|
||
status="active",
|
||
plan_status="collecting",
|
||
metadata_json={"plan_mode": True},
|
||
)
|
||
db_session.add(existing_topic)
|
||
await db_session.commit()
|
||
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_user_reply_2",
|
||
message_text="工单数据来自禅道部署工单。",
|
||
sender="林宇航",
|
||
channel_ref="chat_reply_source",
|
||
thread_context={
|
||
"root_message_id": "om_source_2",
|
||
"reply_to_message_id": "om_source_2",
|
||
},
|
||
source_sender_open_id="ou_reply_source",
|
||
)
|
||
await db_session.commit()
|
||
|
||
topic_rows = (
|
||
await db_session.execute(
|
||
select(MountedTopicWorkspace).order_by(MountedTopicWorkspace.id.asc())
|
||
)
|
||
).scalars().all()
|
||
saved_run = (
|
||
await db_session.execute(select(MountedConversationRun))
|
||
).scalar_one()
|
||
|
||
assert tracking is not None
|
||
assert len(topic_rows) == 1
|
||
assert tracking.topic_workspace.id == int(existing_topic.id)
|
||
assert tracking.topic_workspace.topic_key == "feishu:chat_reply_source:message:om_source_2"
|
||
assert tracking.topic_workspace.root_message_id == "om_source_2"
|
||
assert saved_run.topic_workspace_id == int(existing_topic.id)
|
||
assert saved_run.context_layers_hit == {
|
||
"current_message": True,
|
||
"thread": True,
|
||
"topic": True,
|
||
"source_activation": False,
|
||
}
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_record_decision_and_finalize_run_persist_topic_mode_and_traces(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_2",
|
||
message_text="分配给谁做比较合适？",
|
||
sender="林宇航",
|
||
channel_ref="chat_2",
|
||
thread_context={"root_message_id": "om_root_2"},
|
||
)
|
||
|
||
await service.record_orchestrator_decision(
|
||
tracking=tracking,
|
||
workflow_route="project_signal.create",
|
||
intent_name="signal_create",
|
||
intent_confidence=0.82,
|
||
trigger_reason="manual_review",
|
||
auto_execute=False,
|
||
work_item_id=321,
|
||
route_metadata={"workflow_family": "project_signal", "workflow_action": "create"},
|
||
)
|
||
await service.finalize_run(
|
||
tracking=tracking,
|
||
result={
|
||
"handled": True,
|
||
"status": "captured",
|
||
"classification": "command",
|
||
"work_item_id": 321,
|
||
"workflow_route": "project_signal.create",
|
||
"direct_reply_text": "我先帮你收敛需求，再给负责人建议。",
|
||
},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (
|
||
await db_session.execute(select(MountedTopicWorkspace))
|
||
).scalar_one()
|
||
saved_run = (
|
||
await db_session.execute(select(MountedConversationRun))
|
||
).scalar_one()
|
||
traces = (
|
||
await db_session.execute(
|
||
select(MountedAgentTrace).order_by(MountedAgentTrace.id.asc())
|
||
)
|
||
).scalars().all()
|
||
|
||
assert saved_topic.mode == "plan"
|
||
assert saved_topic.metadata_json["last_workflow_route"] == "project_signal.create"
|
||
assert saved_run.mode_after == "plan"
|
||
assert saved_run.status == "completed"
|
||
assert saved_run.final_reply == "我先帮你收敛需求，再给负责人建议。"
|
||
assert len(traces) == 2
|
||
assert traces[0].agent_name == "ConversationOrchestrator"
|
||
assert traces[1].agent_name == "ReplyComposer"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_record_orchestrator_decision_promotes_requirement_plan_route_for_requirement_plan_turns(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_plan_preserve",
|
||
message_text="C",
|
||
sender="林宇航",
|
||
channel_ref="chat_plan_preserve",
|
||
thread_context={"root_message_id": "om_root_plan_preserve"},
|
||
source_sender_open_id="ou_plan_preserve",
|
||
)
|
||
assert tracking is not None
|
||
tracking.topic_workspace.mode = "plan"
|
||
tracking.topic_workspace.plan_status = "options_ready"
|
||
tracking.topic_workspace.metadata_json = {
|
||
**(tracking.topic_workspace.metadata_json if isinstance(tracking.topic_workspace.metadata_json, dict) else {}),
|
||
"plan_mode": True,
|
||
}
|
||
await db_session.flush()
|
||
|
||
await service.record_orchestrator_decision(
|
||
tracking=tracking,
|
||
workflow_route="query.answer",
|
||
intent_name="option_selection",
|
||
intent_confidence=0.93,
|
||
trigger_reason="reply_anchor_followup",
|
||
auto_execute=False,
|
||
work_item_id=654,
|
||
route_metadata={"message_intent": "requirement_plan"},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (await db_session.execute(select(MountedTopicWorkspace))).scalar_one()
|
||
saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one()
|
||
|
||
assert saved_topic.mode == "plan"
|
||
assert saved_run.mode_after == "plan"
|
||
assert saved_topic.metadata_json["last_workflow_route"] == "requirement_plan.collect"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_record_orchestrator_decision_ignores_chat_mode_override_when_requirement_plan_context_exists(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_plan_override_preserve",
|
||
message_text="补充一个需求事实。",
|
||
sender="林宇航",
|
||
channel_ref="chat_plan_override_preserve",
|
||
thread_context={"root_message_id": "om_root_plan_override_preserve"},
|
||
source_sender_open_id="ou_plan_override_preserve",
|
||
)
|
||
assert tracking is not None
|
||
tracking.topic_workspace.mode = "plan"
|
||
tracking.topic_workspace.plan_status = "collecting"
|
||
tracking.topic_workspace.pending_actions = ["continue_requirement_plan"]
|
||
tracking.topic_workspace.metadata_json = {
|
||
**(tracking.topic_workspace.metadata_json if isinstance(tracking.topic_workspace.metadata_json, dict) else {}),
|
||
"plan_mode": True,
|
||
}
|
||
await db_session.flush()
|
||
|
||
await service.record_orchestrator_decision(
|
||
tracking=tracking,
|
||
workflow_route="query.answer",
|
||
intent_name="contextual_fact_addition",
|
||
intent_confidence=0.95,
|
||
trigger_reason="reply_anchor_followup",
|
||
auto_execute=False,
|
||
work_item_id=655,
|
||
route_metadata={"message_intent": "contextual_fact_addition"},
|
||
mode_override="chat",
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (await db_session.execute(select(MountedTopicWorkspace))).scalar_one()
|
||
saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one()
|
||
|
||
assert saved_topic.mode == "plan"
|
||
assert saved_run.mode_after == "plan"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_begin_ingress_run_skips_unmatched_chat_when_source_activation_whitelist_exists(db_session) -> None:
|
||
db_session.add(
|
||
MountedSourceActivation(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
scope_type="chat",
|
||
scope_ref="whitelisted_chat",
|
||
status="active",
|
||
cutover_mode="mounted_only",
|
||
auto_reply_enabled=True,
|
||
)
|
||
)
|
||
await db_session.commit()
|
||
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_3",
|
||
message_text="这个 chat 不在白名单里。",
|
||
sender="林宇航",
|
||
channel_ref="non_whitelisted_chat",
|
||
thread_context={"root_message_id": "om_root_3"},
|
||
)
|
||
|
||
assert tracking is None
|
||
topic_rows = (await db_session.execute(select(MountedTopicWorkspace))).scalars().all()
|
||
run_rows = (await db_session.execute(select(MountedConversationRun))).scalars().all()
|
||
assert topic_rows == []
|
||
assert run_rows == []
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_begin_ingress_run_persists_conversation_scope_key_for_feishu(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_4",
|
||
message_text="想跟你聊一个新需求。",
|
||
sender="林宇航",
|
||
channel_ref="chat_scope_1",
|
||
thread_context=None,
|
||
source_sender_open_id="ou_scope_1",
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (
|
||
await db_session.execute(select(MountedTopicWorkspace))
|
||
).scalar_one()
|
||
|
||
assert tracking is not None
|
||
assert saved_topic.conversation_scope_key == "feishu:chat_scope_1:ou_scope_1"
|
||
assert saved_topic.plan_status == "idle"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_update_topic_plan_state_persists_explicit_plan_status_and_pending_actions(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_5",
|
||
message_text="先聊需求，不直接创建。",
|
||
sender="林宇航",
|
||
channel_ref="chat_plan_state",
|
||
thread_context=None,
|
||
source_sender_open_id="ou_plan_state",
|
||
)
|
||
|
||
await service.update_topic_plan_state(
|
||
tracking=tracking,
|
||
plan_status="awaiting_execute_confirm",
|
||
topic_status="active",
|
||
pending_actions=["confirm_requirement_draft"],
|
||
metadata={"plan_topic_action": "draft_ready"},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (
|
||
await db_session.execute(select(MountedTopicWorkspace))
|
||
).scalar_one()
|
||
|
||
assert saved_topic.plan_status == "awaiting_execute_confirm"
|
||
assert saved_topic.pending_actions == ["confirm_requirement_draft"]
|
||
assert saved_topic.metadata_json["plan_topic_action"] == "draft_ready"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_record_semantic_shadow_merges_with_finalize_run_execution_outcome(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_6",
|
||
message_text="先继续整理语义上下文，不要直接执行。",
|
||
sender="林宇航",
|
||
channel_ref="chat_semantic_shadow",
|
||
thread_context={"root_message_id": "om_root_6"},
|
||
source_sender_open_id="ou_semantic_shadow",
|
||
)
|
||
|
||
await service.record_semantic_shadow(
|
||
tracking=tracking,
|
||
source_message_preview="先继续整理语义上下文，不要直接执行。",
|
||
semantic_packet_summary={"packet_version": "v1", "recent_turn_count": 2},
|
||
semantic_decision={"topic_relation": "continue_current_topic", "mode_decision": "stay_plan"},
|
||
semantic_gate_result={"accepted": True, "next_plan_status": "collecting"},
|
||
semantic_diff={"changed": True, "legacy_route": "query.answer"},
|
||
)
|
||
await service.finalize_run(
|
||
tracking=tracking,
|
||
result={
|
||
"handled": True,
|
||
"status": "planned",
|
||
"classification": "query",
|
||
"work_item_id": 654,
|
||
"workflow_route": "query.answer",
|
||
"direct_reply_text": "我先把语义上下文整理成 shadow 观测结果。",
|
||
},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (await db_session.execute(select(MountedTopicWorkspace))).scalar_one()
|
||
saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one()
|
||
|
||
assert saved_topic.plan_status == "idle"
|
||
assert saved_run.execution_outcome["handled"] is True
|
||
assert saved_run.execution_outcome["semantic_packet_summary"]["packet_version"] == "v1"
|
||
assert saved_run.execution_outcome["semantic_decision"]["mode_decision"] == "stay_plan"
|
||
assert saved_run.execution_outcome["semantic_gate_result"]["next_plan_status"] == "collecting"
|
||
assert saved_run.execution_outcome["semantic_diff"]["legacy_route"] == "query.answer"
|
||
assert saved_run.execution_outcome["source_message_preview"] == "先继续整理语义上下文，不要直接执行。"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_finalize_run_persists_requirement_plan_context_and_rag_result(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_plan_context_1",
|
||
message_text="确认A，继续往下收。",
|
||
sender="林宇航",
|
||
channel_ref="chat_plan_context",
|
||
thread_context={"root_message_id": "om_root_plan_context_1"},
|
||
source_sender_open_id="ou_plan_context_1",
|
||
)
|
||
|
||
await service.finalize_run(
|
||
tracking=tracking,
|
||
result={
|
||
"handled": True,
|
||
"status": "answered",
|
||
"classification": "question",
|
||
"work_item_id": 901,
|
||
"workflow_route": "query.answer",
|
||
"direct_reply_text": "已确认选项 A，继续收下一轮信息。",
|
||
"rag_result": {
|
||
"answer": "已确认选项 A，继续收下一轮信息。",
|
||
"query_type": "plan",
|
||
},
|
||
"requirement_plan_context": {
|
||
"question_round": 2,
|
||
"last_round_options": {
|
||
"field": "target_user",
|
||
"options": [
|
||
{"key": "A", "title": "单一层级区域经理"},
|
||
{"key": "B", "title": "区域经理 + 店长"},
|
||
{"key": "C", "title": "总部 + 区域 + 门店多层级"},
|
||
],
|
||
"recommendation": {"key": "A"},
|
||
},
|
||
"round_option_maps": {
|
||
"target_user": {
|
||
"A": "单一层级区域经理",
|
||
"B": "区域经理 + 店长",
|
||
"C": "总部 + 区域 + 门店多层级",
|
||
}
|
||
},
|
||
},
|
||
},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one()
|
||
|
||
assert saved_run.final_reply == "已确认选项 A，继续收下一轮信息。"
|
||
assert saved_run.execution_outcome["rag_result"]["query_type"] == "plan"
|
||
assert saved_run.execution_outcome["requirement_plan_context"]["question_round"] == 2
|
||
assert saved_run.execution_outcome["requirement_plan_context"]["last_round_options"]["field"] == "target_user"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_finalize_run_reconciles_semantic_packet_summary_requirement_card_with_requirement_plan_context(
|
||
db_session,
|
||
) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_plan_context_sync",
|
||
message_text="确认A，继续往下收。",
|
||
sender="林宇航",
|
||
channel_ref="chat_plan_context_sync",
|
||
thread_context={"root_message_id": "om_root_plan_context_sync"},
|
||
source_sender_open_id="ou_plan_context_sync",
|
||
)
|
||
|
||
tracking.conversation_run.execution_outcome = {
|
||
"semantic_packet_summary": {
|
||
"draft": {
|
||
"requirement_card": {
|
||
"project_nature": "轻量工具型",
|
||
"target_user": "轻量工具型",
|
||
"question_round": 3,
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
await service.finalize_run(
|
||
tracking=tracking,
|
||
result={
|
||
"handled": True,
|
||
"status": "answered",
|
||
"classification": "question",
|
||
"workflow_route": "query.answer",
|
||
"direct_reply_text": "继续收敛第一版范围。",
|
||
"requirement_plan_context": {
|
||
"project_nature": "轻量定时简报",
|
||
"target_user": "单一层级区域经理",
|
||
"question_round": 4,
|
||
"last_round_options": {
|
||
"field": "first_scope",
|
||
"options": [
|
||
{"key": "A", "title": "全量门店+核心异常类型"},
|
||
{"key": "B", "title": "单城市试点+全异常维度"},
|
||
{"key": "C", "title": "TOP30高动销门店+时效性异常"},
|
||
],
|
||
"recommendation": {"key": "A"},
|
||
},
|
||
},
|
||
},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one()
|
||
|
||
packet_card = saved_run.execution_outcome["semantic_packet_summary"]["draft"]["requirement_card"]
|
||
assert packet_card["project_nature"] == "轻量定时简报"
|
||
assert packet_card["target_user"] == "单一层级区域经理"
|
||
assert packet_card["question_round"] == 4
|
||
assert packet_card["last_round_options"]["field"] == "first_scope"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_finalize_run_updates_topic_last_workflow_route_from_requirement_plan_result(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_plan_route_sync",
|
||
message_text="继续往下收。",
|
||
sender="林宇航",
|
||
channel_ref="chat_plan_route_sync",
|
||
thread_context={"root_message_id": "om_root_plan_route_sync"},
|
||
source_sender_open_id="ou_plan_route_sync",
|
||
)
|
||
tracking.topic_workspace.metadata_json = {
|
||
**(tracking.topic_workspace.metadata_json if isinstance(tracking.topic_workspace.metadata_json, dict) else {}),
|
||
"last_workflow_route": "query.answer",
|
||
"plan_mode": True,
|
||
}
|
||
tracking.topic_workspace.mode = "plan"
|
||
tracking.topic_workspace.plan_status = "collecting"
|
||
await db_session.flush()
|
||
|
||
await service.finalize_run(
|
||
tracking=tracking,
|
||
result={
|
||
"handled": True,
|
||
"status": "answered",
|
||
"classification": "question",
|
||
"workflow_route": "requirement_plan.collect",
|
||
"work_item_id": 902,
|
||
"direct_reply_text": "我先继续收下一轮关键信息。",
|
||
"rag_result": {
|
||
"answer": "我先继续收下一轮关键信息。",
|
||
"query_type": "plan",
|
||
},
|
||
"requirement_plan_context": {"question_round": 3},
|
||
},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_topic = (await db_session.execute(select(MountedTopicWorkspace))).scalar_one()
|
||
saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one()
|
||
|
||
assert saved_topic.metadata_json["last_workflow_route"] == "requirement_plan.collect"
|
||
assert saved_run.execution_outcome["workflow_route"] == "requirement_plan.collect"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_finalize_run_rebinds_conversation_run_to_canonical_draft_artifact_topic(db_session) -> None:
|
||
service = MountedAIRuntimeWriteService(db_session)
|
||
|
||
tracking = await service.begin_ingress_run(
|
||
tenant_id=1,
|
||
source_system="feishu",
|
||
source_message_id="om_message_plan_canonical_rebind",
|
||
message_text="想聊个新需求。",
|
||
sender="林宇航",
|
||
channel_ref="chat_plan_canonical_rebind",
|
||
thread_context=None,
|
||
source_sender_open_id="ou_plan_canonical_rebind",
|
||
)
|
||
assert tracking is not None
|
||
ingress_topic_id = int(tracking.topic_workspace.id)
|
||
|
||
canonical_topic = MountedTopicWorkspace(
|
||
tenant_id=1,
|
||
topic_key="feishu:chat_plan_canonical_rebind:message:om_existing_requirement_root",
|
||
source_system="feishu",
|
||
channel_ref="chat_plan_canonical_rebind",
|
||
conversation_scope_key="feishu:chat_plan_canonical_rebind:ou_plan_canonical_rebind",
|
||
root_message_id=None,
|
||
mode="plan",
|
||
status="active",
|
||
plan_status="collecting",
|
||
metadata_json={"plan_mode": True},
|
||
)
|
||
db_session.add(canonical_topic)
|
||
await db_session.flush()
|
||
|
||
await service.finalize_run(
|
||
tracking=tracking,
|
||
result={
|
||
"handled": True,
|
||
"status": "captured",
|
||
"classification": "command",
|
||
"workflow_route": "project_create.create",
|
||
"direct_reply_text": "已整理成草稿，等待确认。",
|
||
"reply_template_key": "draft_pending_confirm",
|
||
"draft_artifact": {
|
||
"type": "requirement_plan_draft",
|
||
"topic_workspace_id": int(canonical_topic.id),
|
||
"draft_workspace_id": 36,
|
||
"draft_version": 1,
|
||
"promote_to_confirmation": True,
|
||
},
|
||
},
|
||
)
|
||
await db_session.commit()
|
||
|
||
saved_run = (await db_session.execute(select(MountedConversationRun))).scalar_one()
|
||
traces = (
|
||
await db_session.execute(
|
||
select(MountedAgentTrace).order_by(MountedAgentTrace.id.asc())
|
||
)
|
||
).scalars().all()
|
||
|
||
assert int(saved_run.topic_workspace_id) == int(canonical_topic.id)
|
||
assert int(saved_run.topic_workspace_id) != ingress_topic_id
|
||
assert saved_run.execution_outcome["draft_artifact"]["topic_workspace_id"] == int(canonical_topic.id)
|
||
assert traces[-1].agent_name == "ReplyComposer"
|