PMFlow/backend/tests/test_ai_pmo_signal_source_o...

715 lines
25 KiB
Python

from __future__ import annotations
from types import SimpleNamespace
from unittest.mock import AsyncMock
import pytest
from app.services.ai_pmo.ai_pmo_intent import CanonicalIntent
from app.services.ai_pmo.ai_pmo_kernel.ai_pmo_typed_executor_mixins_parse import (
TypedExecutorParseMixin,
)
from app.services.ai_pmo.ai_pmo_typed_executor_discovery.mixin_discover import (
DiscoveryFlowMixin,
)
from app.services.ai_pmo.ai_pmo_intent import WorkflowRoute
from app.services.ai_pmo.ai_pmo_kernel.ai_pmo_reply_renderer import ReplyRenderer
from app.services.ai_pmo.ai_pmo_kernel.mixins.ingress.ai_pmo_kernel_mixins_ingress_storage import (
KernelIngressStorageMixin,
)
from app.services.ai_pmo.ai_pmo_kernel.mixins.followup.ai_pmo_kernel_mixins_followup_project_signal_resume import (
KernelFollowupProjectSignalResumeMixin,
)
from app.services.ai_pmo.ai_pmo_typed_executor_execute import mixin_execute_signal_source as signal_source_execute_module
from app.services.ai_pmo.ai_pmo_typed_executor_execute.mixin_execute_signal_source import (
ExecuteSignalSourceMixin,
)
from app.services.signal.signal_source_orchestration import SignalSourceOrchestrationService
def test_signal_source_onboard_idempotency_key_prefers_message_key() -> None:
key_one = ExecuteSignalSourceMixin._build_signal_source_onboard_idempotency_key(
source_system="feishu",
source_message_id="msg-123",
)
key_two = ExecuteSignalSourceMixin._build_signal_source_onboard_idempotency_key(
source_system="feishu",
source_message_id="msg-123",
)
bare_key = ExecuteSignalSourceMixin._build_signal_source_onboard_idempotency_key(
source_system="",
source_message_id="msg-123",
)
assert key_one == "feishu:msg-123"
assert key_two == "feishu:msg-123"
assert bare_key == "msg-123"
@pytest.mark.asyncio
async def test_signal_source_onboard_replays_existing_ledger_without_reexecution() -> None:
class _FakeLedger:
def __init__(self) -> None:
self.status = "applied"
self.result_payload = {
"status": "succeeded",
"signal_id": 88,
"project_id": 11,
"signal_source_workflow": {
"status": "completed",
"run_status": "completed",
},
"signal_source_orchestration": {
"signal_source": {
"registry_id": 501,
"source_name": "Webhook A",
}
},
"auto_tasks": [],
}
class _FakeWorkflow:
def __init__(self) -> None:
self.called = False
async def run(self, **kwargs):
self.called = True
raise AssertionError("signal_source_workflow.run should not be called on replay")
class _FakeDB:
def __init__(self) -> None:
self.committed = False
async def commit(self):
self.committed = True
class _FakeExecutor(ExecuteSignalSourceMixin):
def __init__(self) -> None:
self.db = _FakeDB()
self.tenant_id = 1
self.signal_source_workflow = _FakeWorkflow()
self.loaded_key = None
async def _load_signal_source_onboard_ledger(self, *, idempotency_key):
self.loaded_key = idempotency_key
return _FakeLedger()
async def _claim_signal_source_onboard_ledger(self, **kwargs):
raise AssertionError("ledger claim should not be reached when replaying an applied result")
def _build_signal_source_semantic_payload(self, **kwargs):
raise AssertionError("semantic payload should not be rebuilt on replay")
async def _create_signal(self, **kwargs):
raise AssertionError("signal creation should not be reached on replay")
@staticmethod
def _normalize_title_candidate(value):
return str(value or "").strip()
@staticmethod
def _is_acceptable_title(text):
return bool(text)
@staticmethod
def _extract_auto_tasks(control_plane_payload):
return []
kernel = _FakeExecutor()
result = await kernel._execute_signal_source_onboard_intent(
intent=type("Intent", (), {"project_id": 11, "signal_summary": "摘要", "signal_title": "标题"})(),
message_text="请接入这个信号源",
operator_user_id=9,
source_system="feishu",
source_message_id="msg-abc",
channel_ref="oc-1",
)
assert result["status"] == "succeeded"
assert result["signal_id"] == 88
assert kernel.loaded_key == "feishu:msg-abc"
assert kernel.db.committed is False
assert kernel.signal_source_workflow.called is False
@pytest.mark.asyncio
async def test_signal_source_onboard_materializes_reply_anchor_candidate_refs_into_multiple_signals(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _FakeLedger:
def __init__(self) -> None:
self.status = "processing"
self.result_payload = None
class _FakeWorkflow:
def __init__(self) -> None:
self.calls: list[int] = []
async def run(self, **kwargs):
self.calls.append(int(kwargs["signal_id"]))
return {
"status": "completed",
"raw_result": {
"signal_source": {"source_name": f"registry-{kwargs['signal_id']}"},
"execution": {},
},
}
class _FakeDB:
def __init__(self) -> None:
self.commits = 0
async def commit(self):
self.commits += 1
async def flush(self):
return None
class _FakeContextCache:
async def invalidate_system_context(self, *, tenant_id):
_ = tenant_id
return None
class _FakeExecutor(ExecuteSignalSourceMixin, TypedExecutorParseMixin):
def __init__(self) -> None:
self.db = _FakeDB()
self.tenant_id = 1
self.signal_source_workflow = _FakeWorkflow()
self.created_titles: list[str] = []
async def _load_signal_source_onboard_ledger(self, *, idempotency_key):
_ = idempotency_key
return None
async def _claim_signal_source_onboard_ledger(self, **kwargs):
_ = kwargs
return _FakeLedger(), True
async def _create_signal(self, **kwargs):
self.created_titles.append(str(kwargs["title"]))
signal_id = 100 + len(self.created_titles)
return SimpleNamespace(id=signal_id, title=kwargs["title"])
async def _finalize_signal_source_onboard_ledger(self, **kwargs):
_ = kwargs
return None
@staticmethod
def _normalize_title_candidate(value):
return str(value or "").strip()
@staticmethod
def _is_acceptable_title(text):
return bool(text)
@staticmethod
def _extract_auto_tasks(control_plane_payload):
_ = control_plane_payload
return []
monkeypatch.setattr(signal_source_execute_module, "context_cache_service", _FakeContextCache(), raising=False)
intent = CanonicalIntent(
name="signal_source.onboard",
confidence=0.95,
project_id=11,
signal_title="",
signal_summary="",
metadata={
"operation_intent_contract": {
"intent_phase": "execute_write",
"project_binding": 11,
"candidate_set_binding": "recset_invoice_01",
"candidate_refs": ["cand_retry", "cand_platform"],
},
"reply_anchor_qa_context": {
"project_id": 11,
"signal_source_recommendation": {
"candidate_set_id": "recset_invoice_01",
"project_binding": 11,
"candidates": [
{
"candidate_ref": "cand_retry",
"display_name": "开票失败或重试回执",
"business_value": "最早看到票开不出去",
"materialization_payload": {
"display_name": "开票失败或重试回执",
"business_value": "最早看到票开不出去",
},
},
{
"candidate_ref": "cand_platform",
"display_name": "税控或发票平台状态变化",
"business_value": "能区分是业务失败还是外部平台波动",
"materialization_payload": {
"display_name": "税控或发票平台状态变化",
"business_value": "能区分是业务失败还是外部平台波动",
},
},
],
},
},
},
)
kernel = _FakeExecutor()
result = await kernel._execute_signal_source_onboard_intent(
intent=intent,
message_text="把刚才推荐的那两类外部信号源写进自动开票系统。",
operator_user_id=9,
source_system="feishu",
source_message_id="msg-execute-write-1",
channel_ref="oc-1",
)
assert result["status"] == "succeeded"
assert result["project_id"] == 11
assert result["signal_ids"] == [101, 102]
assert result["signal_id"] == 101
assert result["requested_count"] == 2
assert result["candidate_set_binding"] == "recset_invoice_01"
assert result["candidate_refs"] == ["cand_retry", "cand_platform"]
assert result["reply_anchor_qa_context"]["signal_source_recommendation"]["candidate_set_id"] == "recset_invoice_01"
assert kernel.created_titles == ["开票失败或重试回执", "税控或发票平台状态变化"]
assert kernel.signal_source_workflow.calls == [101, 102]
@pytest.mark.asyncio
async def test_signal_source_onboard_direct_write_stays_succeeded_when_downstream_is_pending_manual(
monkeypatch: pytest.MonkeyPatch,
) -> None:
class _FakeLedger:
def __init__(self) -> None:
self.status = "processing"
self.result_payload = None
class _FakeWorkflow:
async def run(self, **kwargs):
_ = kwargs
return {
"status": "pending_manual",
"raw_result": {
"signal_source": {"source_name": "开票失败或重试回执"},
"execution": {},
},
}
class _FakeDB:
async def commit(self):
return None
async def flush(self):
return None
class _FakeContextCache:
async def invalidate_system_context(self, *, tenant_id):
_ = tenant_id
return None
class _FakeExecutor(ExecuteSignalSourceMixin, TypedExecutorParseMixin):
def __init__(self) -> None:
self.db = _FakeDB()
self.tenant_id = 1
self.signal_source_workflow = _FakeWorkflow()
async def _load_signal_source_onboard_ledger(self, *, idempotency_key):
_ = idempotency_key
return None
async def _claim_signal_source_onboard_ledger(self, **kwargs):
_ = kwargs
return _FakeLedger(), True
async def _create_signal(self, **kwargs):
_ = kwargs
return SimpleNamespace(id=301, title="开票失败或重试回执")
async def _finalize_signal_source_onboard_ledger(self, **kwargs):
_ = kwargs
return None
@staticmethod
def _normalize_title_candidate(value):
return str(value or "").strip()
@staticmethod
def _is_acceptable_title(text):
return bool(text)
@staticmethod
def _extract_auto_tasks(control_plane_payload):
_ = control_plane_payload
return []
monkeypatch.setattr(signal_source_execute_module, "context_cache_service", _FakeContextCache(), raising=False)
kernel = _FakeExecutor()
result = await kernel._execute_signal_source_onboard_intent(
intent=CanonicalIntent(
name="signal_source.onboard",
confidence=0.95,
project_id=11,
signal_title="",
signal_summary="",
metadata={
"operation_intent_contract": {
"intent_phase": "execute_write",
"project_binding": 11,
}
},
),
message_text="把开票失败或重试回执加入自动开票系统。",
operator_user_id=9,
source_system="feishu",
source_message_id="msg-execute-write-pending",
channel_ref="oc-1",
)
assert result["status"] == "succeeded"
assert result["post_process_status"] == "pending_manual"
assert result["signal_source_workflow"]["status"] == "pending_manual"
assert result["signal_id"] == 301
@pytest.mark.asyncio
async def test_signal_source_followup_resume_reuses_stored_candidate_binding_for_confirm_write() -> None:
captured: dict[str, object] = {}
class _EmptyResult:
@staticmethod
def scalar_one_or_none():
return None
class _FakeDB:
async def execute(self, stmt):
_ = stmt
return _EmptyResult()
class _FakeExecutorService:
async def execute(self, **kwargs):
captured["intent"] = kwargs["intent"]
return {
"status": "succeeded",
"project_id": 11,
"signal_id": 101,
"signal_ids": [101, 102],
"signal_source_workflow": {"status": "completed"},
}
class _FakeKernel(KernelFollowupProjectSignalResumeMixin):
def __init__(self) -> None:
self.db = _FakeDB()
self.tenant_id = 1
self.executor = _FakeExecutorService()
self.intent_resolver = SimpleNamespace(
resolve=lambda text, project_id: CanonicalIntent(
name="signal_source.onboard",
confidence=0.92,
project_id=project_id,
signal_title="",
signal_summary=text[:120],
metadata={},
)
)
self.review_service = SimpleNamespace(review_work_item=AsyncMock(return_value={}))
self.reply_renderer = SimpleNamespace(render=lambda **kwargs: {"text": "已继续处理"})
async def _verify_execution(self, **kwargs):
_ = kwargs
return {"verified": True, "issues": []}
async def _project_success(self, **kwargs):
_ = kwargs
return None
async def _maybe_emit_risk_cadence_snapshot(self, **kwargs):
_ = kwargs
return None
@staticmethod
def _normalize_route_from_stored_intent(value):
return str(value or "")
@staticmethod
def _enrich_signal_source_intent_from_closure(*, intent, message_text, closure_payload):
_ = (message_text, closure_payload)
return intent
kernel = _FakeKernel()
target_work_item = SimpleNamespace(
id=41,
project_id=11,
state="waiting_manual",
intent="signal_source.onboard",
signal_id=None,
review_summary={},
context_snapshot={},
execution_result={
"status": "pending_manual",
"project_id": 11,
"candidate_set_binding": "recset_invoice_01",
"candidate_refs": ["cand_retry", "cand_platform"],
"reply_anchor_qa_context": {
"project_id": 11,
"signal_source_recommendation": {
"candidate_set_id": "recset_invoice_01",
"project_binding": 11,
"candidates": [
{
"candidate_ref": "cand_retry",
"display_name": "开票失败或重试回执",
"business_value": "最早看到票开不出去",
},
{
"candidate_ref": "cand_platform",
"display_name": "税控或发票平台状态变化",
"business_value": "能区分是业务失败还是平台波动",
},
],
},
},
"action_contract": {
"operation_code": "signal_source.onboard",
"params": {
"project_id": 11,
"candidate_set_binding": "recset_invoice_01",
"candidate_refs": ["cand_retry", "cand_platform"],
},
},
},
)
target_ingest = SimpleNamespace(payload={"message_text": "把刚才推荐的那几类外部信号源加入自动开票系统。"})
await kernel._resume_followup_target(
target_work_item=target_work_item,
target_ingest=target_ingest,
followup_intent=CanonicalIntent(
name="followup.confirm",
confidence=0.99,
project_id=11,
signal_title="",
signal_summary="",
metadata={},
),
followup_sender="林宇航",
operator_user_id=9,
source_system="feishu",
source_message_id="om-followup-confirm-1",
world_model={"resolved_project_id": 11},
message_text="确认入库",
workflow_route=WorkflowRoute(
family="signal_source",
action="onboard",
project_id=11,
requested_count=1,
requires_confirmation=True,
metadata={"workflow_route": "signal_source.onboard"},
),
)
rebuilt_intent = captured["intent"]
assert rebuilt_intent.metadata["operation_intent_contract"]["candidate_set_binding"] == "recset_invoice_01"
assert rebuilt_intent.metadata["operation_intent_contract"]["candidate_refs"] == ["cand_retry", "cand_platform"]
assert (
rebuilt_intent.metadata["reply_anchor_qa_context"]["signal_source_recommendation"]["candidate_set_id"]
== "recset_invoice_01"
)
def test_reply_renderer_formats_multi_signal_source_onboard_success_naturally() -> None:
renderer = ReplyRenderer(
state_waiting_manual="waiting_manual",
state_failed="failed",
receipt_phase_action="action",
receipt_phase_pending="pending",
receipt_phase_error="error",
)
intent = CanonicalIntent(
name="signal_source.onboard",
confidence=0.96,
project_id=11,
signal_title="",
signal_summary="",
metadata={},
)
rendered = renderer.render(
work_state="succeeded",
intent=intent,
decision_record={"approval_requirement": {}},
execution_result={
"status": "succeeded",
"post_process_status": "pending_manual",
"project_id": 11,
"signal_id": 101,
"signal_ids": [101, 102],
"signal_source_workflow": {"status": "pending_manual"},
},
)
assert rendered["template_key"] == "action_succeeded"
assert "项目#11" in rendered["text"]
assert "2条信号源" in rendered["text"]
assert "后续仍有待处理步骤" in rendered["text"]
@pytest.mark.asyncio
async def test_project_signal_discovery_uses_project_id_without_name_error() -> None:
class _FakeDiscovery(DiscoveryFlowMixin):
tenant_id = 1
@staticmethod
def _extract_requested_signal_count(*, message_text, metadata):
_ = (message_text, metadata)
return 1
async def _analyze_signal_batch_with_high_tier_model(self, **kwargs):
_ = kwargs
return {
"items": [
{
"title": "候选信号",
"summary": "摘要",
"content": "内容",
"signal_type": "新需求",
"priority": "medium",
}
]
}
async def _collect_project_signal_discovery_context(self, **kwargs):
_ = kwargs
return {
"retrieval_items": [{"content": "evidence"}],
"retrieval_metrics": {},
"entity_matches": [],
"tool_trace": [],
"source_registries": [],
"truth_nodes": [],
"experiments": [],
"recent_signals": [],
"capability_bundle": {},
}
async def _refine_signal_title_with_ai(self, **kwargs):
_ = kwargs
return "信号A"
@staticmethod
def _build_candidate_dedupe_key(*, project_id, candidate):
return f"{project_id}:{candidate.get('title')}"
@staticmethod
def _build_signal_title_refine_error_message(exc):
return str(exc)
kernel = _FakeDiscovery()
result = await kernel.discover_project_signal_candidates(
intent=CanonicalIntent(
name="project_signal.discover",
confidence=0.9,
project_id=2,
signal_title=None,
signal_summary=None,
metadata={},
),
message_text="给项目 2 先出一条候选信号",
source_system="feishu",
source_message_id="msg-1",
channel_ref="oc-chat-1",
)
assert result["status"] == "pending_confirmation"
assert result["project_id"] == 2
assert result["candidate_signals"][0]["title"] == "信号A"
def test_signal_source_kind_uses_explicit_repo_locator_without_hint_table() -> None:
assert (
SignalSourceOrchestrationService._normalize_source_kind(
None,
text="请把这个来源接进来",
requested_via="",
source_payload={"repo": "acme/reco-engine"},
)
== "github"
)
def test_signal_source_kind_uses_channel_config_without_hint_table() -> None:
assert (
SignalSourceOrchestrationService._normalize_source_kind(
None,
text="先挂到当前群",
requested_via="",
source_payload={"config": {"channel_id": "oc_abc123"}},
)
== "feishu"
)
def test_execution_action_contract_preserves_signal_source_candidate_binding_and_reply_anchor_context() -> None:
class _FakeStorage(KernelIngressStorageMixin):
@staticmethod
def _primary_operation_code(value):
_ = value
return "signal_source.onboard"
intent = CanonicalIntent(
name="signal_source.onboard",
confidence=0.96,
project_id=11,
signal_title="",
signal_summary="",
metadata={
"operation_intent_contract": {
"intent_phase": "execute_write",
"project_binding": 11,
"candidate_set_binding": "recset_invoice_01",
"candidate_refs": ["cand_retry", "cand_platform"],
},
"reply_anchor_qa_context": {
"project_id": 11,
"signal_source_recommendation": {
"candidate_set_id": "recset_invoice_01",
},
},
},
)
contract = _FakeStorage()._build_execution_action_contract(
intent=intent,
workflow_route=SimpleNamespace(key="signal_source.onboard", action="onboard", project_id=11, requested_count=2),
decision_payload={
"execution_plan": {
"actions": [
{
"operation_code": "signal_source.onboard",
"mode": "write",
}
]
}
},
execution_result={
"status": "succeeded",
"project_id": 11,
"signal_ids": [101, 102],
"candidate_set_binding": "recset_invoice_01",
"candidate_refs": ["cand_retry", "cand_platform"],
"reply_anchor_qa_context": {
"project_id": 11,
"signal_source_recommendation": {
"candidate_set_id": "recset_invoice_01",
},
},
},
source_system="feishu",
source_message_id="om-signal-source-write-1",
channel_ref="chat-1",
)
assert contract["params"]["candidate_set_binding"] == "recset_invoice_01"
assert contract["params"]["candidate_refs"] == ["cand_retry", "cand_platform"]
assert contract["params"]["intent_phase"] == "execute_write"
assert contract["metadata"]["reply_anchor_qa_context"]["signal_source_recommendation"]["candidate_set_id"] == "recset_invoice_01"