aps-agent/tests/golden/test_trace_external_http.py

# ============================================================
# round-39 黄金测试：EXTERNAL 真实 HTTP skill 端到端（矩阵 114 剩余项）
# 覆盖：algo_skill_stub.create_stub_app 起真实 HTTP 服务 → 注册 HTTP skill →
#       flex.schedule.external 走真实 HTTP POST → ALGO_RUN 审计携带
#       traceChainHash/traceCount/traceSummary/traceItems → 可复算 + verify_audit_trace 校验通过。
# ============================================================
from __future__ import annotations

import asyncio
import json
import re
import socket
import threading
import time

import pytest

from server.agent_core.evidence import verify_audit_trace
from server.aps_domain.workflow import handle_intent
from server.contracts import IntentResult
from server.state.seed import ensure_flex_seed, seed_world

_HEX64 = re.compile(r"^[0-9a-f]{64}$")


class _MemStore:
    """最小内存 store：满足 _run_flex EXTERNAL 路径对 data/next_id/save 的依赖。"""

    def __init__(self, data):
        self.data = data

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

    def save(self):
        pass


def _free_port() -> int:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


@pytest.fixture
def http_stub(tmp_path, monkeypatch):
    """启动 algo_skill_stub 真实 HTTP 服务，并注册为外部算法 skill（algo.http.stub）。"""
    import httpx
    import uvicorn
    from server.integrations.algo_skill_stub import create_stub_app

    port = _free_port()
    server = uvicorn.Server(uvicorn.Config(
        create_stub_app(), host="127.0.0.1", port=port, log_level="warning"))
    thread = threading.Thread(target=server.run, daemon=True)
    thread.start()
    try:
        deadline = 100
        while not server.started and deadline > 0:
            time.sleep(0.05)
            deadline -= 1
        with httpx.Client(timeout=5.0) as client:      # 真实 HTTP 健康探测
            resp = client.get("http://127.0.0.1:{0}/health".format(port))
            assert resp.status_code == 200 and resp.json().get("ok") is True
        skills_file = tmp_path / "skills.json"
        skills_file.write_text(json.dumps({"skills": [{
            "skill_id": "algo.http.stub",
            "name": "HTTP 算法桩",
            "endpoint": "http://127.0.0.1:{0}".format(port),
            "track": "flex",
            "enabled": True,
            "timeout_sec": 10.0,
            "capabilities": ["flex_schedule"],
            "ragScopes": ["process", "sop"],
        }]}), encoding="utf-8")
        import server.agent_core.skills as sk
        monkeypatch.setenv("APS_SKILLS_PATH", str(skills_file))
        monkeypatch.setattr(sk, "_registry", None)     # 下次 get_skills() 按新配置重建
        yield {"port": port, "skills_file": skills_file}
    finally:
        server.should_exit = True
        thread.join(timeout=5.0)
        import server.agent_core.skills as sk
        sk._registry = None                            # 清理单例，避免污染其他测试


def _run_external_http_once(store) -> None:
    async def _run():
        await handle_intent(
            store, "t",
            IntentResult(intent="flex.schedule",
                         params={"skillId": "algo.http.stub", "useExternal": True},
                         confidence=1.0, source="RULE_FAST"))

    asyncio.run(_run())


def _external_audit(store) -> dict:
    runs = [e for e in store.data["auditEvents"]
            if e["category"] == "ALGO_RUN" and e["action"] == "flex.schedule.external"]
    assert runs, "应产生一条 ALGO_RUN flex.schedule.external 审计"
    return runs[-1]


def test_external_http_skill_trace_written_and_recomputable(http_stub):
    """矩阵 114：EXTERNAL 真实 HTTP skill 端到端 —— trace 写入且同输入可复算。"""
    store = _MemStore(seed_world())
    ensure_flex_seed(store.data)
    _run_external_http_once(store)
    ev = _external_audit(store)
    rat = ev["rationale"]
    assert isinstance(rat["traceChainHash"], str) and _HEX64.match(rat["traceChainHash"]), "traceChainHash 应为 64 位 SHA-256"
    assert rat["traceCount"] == 3, "链应含 schedule-version/run/algorithm 三条"
    assert [it["kind"] for it in rat["traceSummary"]] == ["schedule-version", "run", "algorithm"]
    assert rat["traceSummary"][2]["ref"] == "algo.http.stub", "algorithm 元素引用应为 skillId"
    assert isinstance(rat.get("traceItems"), list) and len(rat["traceItems"]) == 3, "traceItems 应落盘以便精确重算"

    store2 = _MemStore(seed_world())
    ensure_flex_seed(store2.data)
    _run_external_http_once(store2)
    rat2 = _external_audit(store2)["rationale"]
    assert rat2["traceChainHash"] == rat["traceChainHash"], "同输入 EXTERNAL trace 链应可复算"
    assert rat2["traceSummary"] == rat["traceSummary"]


def test_external_http_skill_trace_passes_verify_audit_trace(http_stub):
    """EXTERNAL HTTP 端到端产出的 ALGO_RUN 审计应通过 verify_audit_trace 全量校验。"""
    store = _MemStore(seed_world())
    ensure_flex_seed(store.data)
    _run_external_http_once(store)
    result = verify_audit_trace(store.data["auditEvents"])
    assert result["ok"] is True
    assert result["checked"] >= 1
    assert result["broken"] == []