group_fqcd_jr/tools/acceptance_check.py

"""独立验收脚本：真实 HTTP + 真实 JWT + 真实 MySQL + 显式注册的测试 Agent。

覆盖 P0-1 越权拦截、P0-2 未注册拒绝、P0-3 端到端闭环、P1-3 SSE delta。
生产代码不注册业务 Agent，本脚本按 TODO 阶段 8 的约定显式注入测试注册表。
"""

from __future__ import annotations

import asyncio
import datetime as dt
import sys
import uuid
from pathlib import Path
from typing import Any

import httpx
import jwt
from sqlalchemy import delete, select

import app.service.admin_service as admin_service
import app.service.agent_run_application_service as run_service
import app.service.public_platform_service as public_platform_service
from app.core.contracts import (
    AgentDefinition,
    AgentRequest,
    CoreResult,
    RequestContext,
    ResolvedAgentConfig,
)
from app.infrastructure.db import SessionFactory
from app.main import create_app
from app.model.audit import InteractionAudit
from app.model.conversation import ConversationMessage
from app.model.platform import AgentRun, DomainEventOutbox, OutboxDelivery, RequestIdempotency
from app.service.agent.base import BaseAgent
from app.service.agent.bootstrap import get_agent_factory
from app.service.agent.factory import AgentFactory
from app.worker.runtime import WorkerRuntime

PRIVATE_KEY = Path("config/jwt/jwt-private.pem").read_text(encoding="utf-8")
CUSTOMER_ID = "9001"  # tools/seed_test_rbac.py 造的数据

_results: list[tuple[str, str, str]] = []


def record(name: str, expected: str, actual: str) -> None:
    _results.append(("PASS" if expected == actual else "FAIL", name, f"期望 {expected} / 实际 {actual}"))


def token(sub: str) -> str:
    now = dt.datetime.now(dt.UTC)
    return jwt.encode(
        {
            "sub": sub,
            "iss": "jr-local",
            "aud": "jr-agent-platform",
            "exp": now + dt.timedelta(minutes=30),
            "nbf": now - dt.timedelta(seconds=5),
            "jti": str(uuid.uuid4()),
        },
        PRIVATE_KEY,
        algorithm="RS256",
    )


class StubGovernance:
    async def resolve(self, definition: AgentDefinition, context: RequestContext) -> ResolvedAgentConfig:
        del definition, context
        return ResolvedAgentConfig(
            config_version="acceptance", prompt_version="acceptance", model_endpoint="stub"
        )

    async def recall(self, context: RequestContext) -> tuple[Any, ...]:
        del context
        return ()

    async def review(
        self, result: Any, context: RequestContext, config: Any, memories: Any
    ) -> Any:
        del context, config, memories
        return result


class AcceptanceAgent(BaseAgent):
    async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
        del request, context
        return CoreResult(text="独立验收通过")


def register_probes(factory: AgentFactory, *, version: str = "acceptance") -> None:
    """向工厂注册两个探针 Agent（customer_service / risk）。

    生产代码不注册业务 Agent（底座只提供框架），而验收需要走完整 HTTP 链路，
    因此通过工厂的公开 `register` 接口注册——这与业务组员接入 Agent 的方式一致。
    """
    for agent_type, roles in (
        ("customer_service", ("customer",)),
        ("risk", ("risk_operator",)),
    ):
        definition = AgentDefinition(
            agent_type=agent_type,
            version=version,
            allowed_roles=roles,
            allowed_portals=("api",),
        )
        factory.register(definition, lambda _ctx, _def=definition: AcceptanceAgent(_def))


def build_registry() -> AgentFactory:
    """替身治理装配：供 HTTP 契约验收使用（**不验证生产装配**）。"""
    factory = AgentFactory(StubGovernance())
    register_probes(factory)
    return factory


def production_factory() -> AgentFactory:
    """生产装配 + 探针 Agent，并**断言生产装配完整**。

    与旧做法的关键差别：不再无条件把测试工厂塞进生产模块的全局名字——那样做会让
    "生产装配到底可不可用"完全验收不到。这里直接用生产工厂，先断言模型服务、
    工具执行器、意图分类器都已注入，再把探针 Agent 注册进去。
    """
    factory = get_agent_factory()
    missing = [
        name
        for name, value in (
            ("model_service", factory._model_service),
            ("tool_executor", factory._tool_executor),
            ("intent_classifier", factory._intent_classifier),
        )
        if value is None
    ]
    if missing:
        raise SystemExit(f"生产装配不完整，缺少：{missing}")
    register_probes(factory, version="acceptance-production")
    return factory


def inject_factory(factory: AgentFactory) -> None:
    """把替身工厂绑定到服务模块的全局名字——**仅限默认的 HTTP 契约验收模式**。

    这是本脚本唯一的注入手段：Controller 直接构造 `AgentRunApplicationService(session)`
    而不接收工厂，因此没有依赖注入点可用。生产装配的验证由 `--production` 模式与
    `tools/memory_chain_probe.py` 承担，不会被这个注入掩盖。
    """
    for module in (run_service, public_platform_service, admin_service):
        if hasattr(module, "get_agent_factory"):
            module.get_agent_factory = lambda: factory  # type: ignore[attr-defined]


async def _collect_events(
    client: httpx.AsyncClient, run_id: str, auth: dict[str, str]
) -> list[str]:
    events: list[str] = []
    async with client.stream(
        "GET", f"/api/v1/agent-runs/{run_id}/events", headers=auth, timeout=30
    ) as stream:
        async for line in stream.aiter_lines():
            if line.startswith("event:"):
                events.append(line.split(":", 1)[1].strip())
                if events[-1] in {"done", "error"}:
                    break
    return events


async def cleanup(session_id: str, run_id: str | None) -> None:
    async with SessionFactory() as session, session.begin():
        if run_id:
            event_ids = select(DomainEventOutbox.event_id).where(
                DomainEventOutbox.aggregate_id == run_id
            )
            await session.execute(
                delete(OutboxDelivery).where(OutboxDelivery.event_id.in_(event_ids))
            )
            await session.execute(
                delete(DomainEventOutbox).where(DomainEventOutbox.aggregate_id == run_id)
            )
        await session.execute(
            delete(InteractionAudit).where(InteractionAudit.session_id == session_id)
        )
        await session.execute(delete(AgentRun).where(AgentRun.session_id == session_id))
        await session.execute(
            delete(RequestIdempotency).where(RequestIdempotency.session_id == session_id)
        )
        await session.execute(
            delete(ConversationMessage).where(ConversationMessage.session_id == session_id)
        )


async def main() -> None:
    if "--production" in sys.argv:
        factory = production_factory()
        print("模式：生产装配验收（真实治理链 + 探针 Agent）")
    else:
        factory = build_registry()
        inject_factory(factory)
        print("模式：HTTP 契约验收（替身治理）——未验证生产装配；加 --production 验生产链路")

    app = create_app()
    session_id = f"acceptance-{uuid.uuid4()}"
    run_id: str | None = None
    try:
        async with httpx.AsyncClient(
            transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=30
        ) as client:
            auth = {"Authorization": f"Bearer {token(CUSTOMER_ID)}"}

            response = await client.post(
                "/api/v1/agent-runs",
                json={
                    "agent_type": "customer_service",
                    "message": "独立验收消息",
                    "session_id": session_id,
                    "idempotency_key": uuid.uuid4().hex,
                },
                headers=auth,
            )
            record("客户调用已授权 Agent", "202", str(response.status_code))
            if response.status_code == 202:
                run_id = response.json().get("data", {}).get("run_id")

            response = await client.post(
                "/api/v1/agent-runs",
                json={
                    "agent_type": "risk",
                    "message": "越权尝试",
                    "session_id": session_id,
                    "idempotency_key": uuid.uuid4().hex,
                },
                headers=auth,
            )
            record("客户越权调用 risk Agent", "403", str(response.status_code))

            response = await client.post(
                "/api/v1/agent-runs",
                json={
                    "agent_type": "no_such_agent",
                    "message": "x",
                    "session_id": session_id,
                    "idempotency_key": uuid.uuid4().hex,
                },
                headers=auth,
            )
            record("未注册 agent_type", "404", str(response.status_code))

            if run_id:
                runtime = WorkerRuntime(factory)
                await runtime.dispatch_one(run_id=run_id)

                # 先建立 SSE 订阅（此时 run 仍为 queued），再执行，才能观察到 delta 分块
                subscription = asyncio.create_task(_collect_events(client, run_id, auth))
                await asyncio.sleep(1.0)
                await runtime.execute(run_id)
                events = await subscription

                record("SSE 含 delta 事件", "yes", "yes" if "delta" in events else "no")
                record("SSE 终止事件", "done", events[-1] if events else "无事件")

                response = await client.get(f"/api/v1/agent-runs/{run_id}", headers=auth)
                # 文档 §3.3：单资源成功响应是 {data, meta} 信封。
                body = response.json().get("data") or {}
                record("运行终态", "succeeded", str(body.get("status")))
                record(
                    "运行结果内容",
                    "独立验收通过",
                    str((body.get("result") or {}).get("content")),
                )
    finally:
        await cleanup(session_id, run_id)

    width = max(len(name) for _, name, _ in _results)
    print()
    for verdict, name, detail in _results:
        print(f"[{verdict}] {name.ljust(width)}  {detail}")
    failed = sum(1 for verdict, _, _ in _results if verdict == "FAIL")
    print(f"\n合计 {len(_results)} 项，失败 {failed} 项")


if __name__ == "__main__":
    asyncio.run(main())