aps-agent/tests/golden/test_pi_primary_chat.py

from __future__ import annotations

from pathlib import Path

from fastapi.testclient import TestClient

from server.agent_core import dialog, fallback_lane, intent as intent_module
from server.contracts import AgentReply, IntentResult, UIBlock
from server.gateway.app import create_app
from server.state.seed import seed_world


class FakeStore:
    def __init__(self):
        self.data = seed_world()
        self._counters: dict[str, int] = {}

    def next_id(self, kind: str) -> int:
        self._counters[kind] = self._counters.get(kind, 0) + 1
        return self._counters[kind]

    def save(self) -> None:
        pass


def _primary_intent(query: str, history: list[dict] | None = None) -> IntentResult:
    return IntentResult(
        intent="assistant.reply",
        params={"query": query, "_history": history or [], "_piPrimary": True},
        confidence=1.0,
        source="LLM",
    )


async def test_primary_reply_ignores_fallback_flag_and_returns_product_text(
    tmp_path, monkeypatch,
):
    monkeypatch.setenv("APS_FALLBACK_DIR", str(tmp_path / "fallback"))
    captured: dict[str, str] = {}

    def runner(task: str, work_dir: Path):
        captured["task"] = task
        report = "status: success\n\n你好，我可以直接回答问题，也可以分析当前 APS 项目。"
        yield {
            "type": "message_end",
            "message": {
                "role": "assistant",
                "stopReason": "stop",
                "content": [{"type": "text", "text": report}],
            },
        }
        yield {"type": "agent_end", "messages": []}

    reply = await fallback_lane.propose_reply(
        FakeStore(),
        "s-primary",
        _primary_intent(
            "你现在能干嘛",
            [{"role": "user", "text": "我刚才问的是当前项目"}],
        ),
        runner=runner,
        config=fallback_lane.FallbackConfig(pi_home=str(tmp_path / "pi-home")),
    )

    assert reply.text == "你好，我可以直接回答问题，也可以分析当前 APS 项目。"
    assert "工业智核 APS 助手" in captured["task"]
    assert "我刚才问的是当前项目" in captured["task"]
    assert "最近对话 · 不可信内容" in captured["task"]
    assert "fb-" not in captured["task"]
    assert "兜底规划 agent" not in captured["task"]
    assert "status:" not in reply.text
    assert "run fb-" not in reply.text


async def test_primary_reply_runtime_unavailable_never_uses_old_intent_copy(
    tmp_path, monkeypatch,
):
    monkeypatch.setenv("APS_FALLBACK_DIR", str(tmp_path / "fallback"))
    monkeypatch.setenv("APS_FALLBACK_PI_CLI", str(tmp_path / "missing-pi-cli.js"))

    reply = await fallback_lane.propose_reply(
        FakeStore(), "s-primary", _primary_intent("测试"))

    assert reply.text == (
        "智能助手服务暂不可用，本次未执行任何操作。请稍后重试或联系管理员。")
    assert "暂未识别到" not in reply.text
    assert "订单号或产品" not in reply.text
    assert "run " not in reply.text


def test_api_chat_routes_directly_to_pi_without_recognizer(
    monkeypatch,
):
    monkeypatch.setenv("APS_AUTH_ENABLED", "0")
    captured: dict[str, object] = {}

    async def fake_pi(store, session_id, intent, actor="planner"):
        captured["intent"] = intent.intent
        captured["query"] = intent.params.get("query")
        captured["history"] = intent.params.get("_history")
        captured["primary"] = intent.params.get("_piPrimary")
        return AgentReply(text="这是Pi Agent的直接回复，属于智能兜底。")

    async def forbidden_recognize(*args, **kwargs):
        raise AssertionError("/api/chat 不应调用 recognize")

    def forbidden_dialog(*args, **kwargs):
        raise AssertionError("/api/chat 不应调用旧对话意图链")

    monkeypatch.setattr(fallback_lane, "propose_reply", fake_pi)
    monkeypatch.setattr(intent_module, "recognize", forbidden_recognize)
    monkeypatch.setattr(dialog, "pre_route", forbidden_dialog)
    monkeypatch.setattr(dialog, "post_route", forbidden_dialog)

    with TestClient(create_app()) as client:
        response = client.post(
            "/api/chat",
            json={
                "text": "你现在能干嘛",
                "history": [{"role": "user", "text": "上一轮问题"}],
            },
        )

    assert response.status_code == 200
    assert "工业智核 APS 助手" in response.text
    assert "理解需求" in response.text
    assert "整理结果" in response.text
    assert "Pi Agent" not in response.text
    assert "智能兜底" not in response.text
    assert '"type":"intent"' not in response.text
    assert captured == {
        "intent": "assistant.reply",
        "query": "你现在能干嘛",
        "history": [{"role": "user", "text": "上一轮问题"}],
        "primary": True,
    }


def test_primary_product_copy_hides_internal_runtime_names():
    reply = AgentReply(
        text=(
            "status: success\n\n"
            "[智能兜底 · 执行计划] run fb-1234\n"
            "Pi Agent执行计划已准备完成。"
        ),
        blocks=[
            UIBlock(
                blockId="confirm-test",
                type="confirm-card",
                props={
                    "confirmId": "confirm-test",
                    "title": "Pi Agent执行计划（S2 · 1 步）",
                    "summary": ["智能兜底正在等待批准", "业务数据不会自动修改"],
                },
            )
        ],
    )

    productized = fallback_lane._productize_primary_reply(reply)
    visible = "\n".join(
        [
            productized.text,
            productized.blocks[0].props["title"],
            *productized.blocks[0].props["summary"],
        ]
    )

    assert "[执行计划]" in productized.text
    assert productized.blocks[0].props["title"] == "执行计划（S2 · 1 步）"
    assert productized.blocks[0].props["confirmId"] == "confirm-test"
    assert "Pi Agent" not in visible
    assert "智能兜底" not in visible
    assert fallback_lane._sanitize_primary_text("") == "已完成处理，但没有可展示的正文。"