scripts/full_integration_test.py

"""
Full system integration test — kernel, public feeds, engines, gateway.
Run after both servers are up (gateway :3000, SRC API :8080).
"""

from __future__ import annotations

import asyncio
import json
import sys
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List

import httpx
import numpy as np

SRC_URL = "http://localhost:8080"
GW_URL = "http://localhost:3000"
GW_TOKEN = "pilot-console-primary-jwt-fallback-x992z"


@dataclass
class TestResult:
    name: str
    passed: bool
    detail: str = ""
    data: Dict[str, Any] = field(default_factory=dict)


results: List[TestResult] = []


def record(name: str, passed: bool, detail: str = "", **data: Any) -> None:
    results.append(TestResult(name, passed, detail, dict(data)))
    status = "PASS" if passed else "FAIL"
    print(f"  [{status}] {name}" + (f" — {detail}" if detail else ""))


async def wait_for(url: str, path: str = "/health", timeout: float = 30.0) -> bool:
    deadline = time.time() + timeout
    async with httpx.AsyncClient() as client:
        while time.time() < deadline:
            try:
                r = await client.get(f"{url}{path}", timeout=3.0)
                if r.status_code == 200:
                    return True
            except Exception:
                pass
            await asyncio.sleep(1.0)
    return False


async def test_src_health() -> None:
    async with httpx.AsyncClient(timeout=10.0) as c:
        r = await c.get(f"{SRC_URL}/health")
        d = r.json()
        record("SRC API health", r.status_code == 200 and d.get("status") == "ok", json.dumps(d))


async def test_gateway_health() -> None:
    async with httpx.AsyncClient(timeout=10.0) as c:
        r = await c.get(f"{GW_URL}/v1/health")
        d = r.json()
        record("Gateway health", r.status_code == 200 and d.get("status") == "ok", json.dumps(d))


async def test_kernel_examples() -> None:
    async with httpx.AsyncClient(timeout=15.0) as c:
        r = await c.get(f"{SRC_URL}/api/examples/worked")
        d = r.json()
        ok = r.status_code == 200 and len(d) == 5
        record("Worked examples (5 forcing functions)", ok, f"modes={list(d.keys())}")


async def test_benchmarks() -> None:
    async with httpx.AsyncClient(timeout=30.0) as c:
        r = await c.get(f"{SRC_URL}/api/benchmarks")
        d = r.json()
        n = d.get("summary", {}).get("operators_tested", 0)
        record("20-operator benchmark", r.status_code == 200 and n == 20, f"mean_gain={d.get('summary',{}).get('mean_gain_pct')}%")


async def test_live_feed() -> None:
    async with httpx.AsyncClient(timeout=30.0) as c:
        r = await c.post(f"{SRC_URL}/api/feed/live", json={"latitude": 38.627, "longitude": -90.199, "hours": 1})
        d = r.json()
        snap = d.get("snapshot", {})
        ok = r.status_code == 200 and "Q_t" in snap and "DT" in snap
        record("Live public sensor feed", ok, f"source={snap.get('source')}, Q_t={snap.get('Q_t')}")


async def test_feed_series_pipeline() -> None:
    async with httpx.AsyncClient(timeout=60.0) as c:
        r = await c.post(f"{SRC_URL}/api/feed/series", json={"latitude": 38.627, "longitude": -90.199, "hours": 12})
        d = r.json()
        engines = d.get("primal_bridge", {})
        atlas = d.get("atlas_fusion", {})
        ok = (
            r.status_code == 200
            and d.get("points", 0) >= 1
            and "trust_anchor_sha512" in d
            and "coherence_retention" in d
        )
        record(
            "Full feed→Dx→Atlas→Primal pipeline",
            ok,
            f"points={d.get('points')}, coherence={round(d.get('coherence_retention',0),3)}",
            engines_available=d.get("primal_bridge", {}).get("ewic_control") is not None,
            atlas_converged=atlas.get("converged"),
        )


async def test_compute_dx_meta() -> None:
    q = [1.0, 2.5, -0.3, 1.8, 0.5]
    async with httpx.AsyncClient(timeout=15.0) as c:
        rdx = await c.post(f"{SRC_URL}/api/compute/dx", json={"q_values": q, "a": 1.0, "dt": 1.0})
        rmeta = await c.post(f"{SRC_URL}/api/compute/meta", json={"q_values": q, "b": 0.091, "dt": 1.0})
        ok = rdx.status_code == 200 and rmeta.status_code == 200
        record("Dx + O(f) compute endpoints", ok, f"dx={rdx.json().get('dx_final')}, of={rmeta.json().get('of_final')}")


async def test_gateway_psi_rpo() -> None:
    headers = {"Authorization": f"Bearer {GW_TOKEN}"}
    async with httpx.AsyncClient(timeout=20.0) as c:
        psi = await c.post(f"{GW_URL}/api/compute/psi", headers=headers, json={"t": 3.5, "x0": 150.0})
        rpo = await c.post(f"{GW_URL}/api/compute/rpo", headers=headers, json={"n": 30, "x_init": 120.0})
        ok = psi.status_code == 200 and rpo.status_code == 200 and psi.json().get("status") == "STABLE"
        record("Gateway psi + rpo offload", ok, f"psi={psi.json().get('psi')}, converged={rpo.json().get('converged')}")


async def test_gateway_src_offload() -> None:
    headers = {"Authorization": f"Bearer {GW_TOKEN}"}
    q = [1.95, 4.25, -0.05, 2.1, -1.3, 0.8]
    async with httpx.AsyncClient(timeout=20.0) as c:
        r = await c.post(
            f"{GW_URL}/api/compute/src",
            headers=headers,
            json={"q_values": q, "a": 1.0, "b": 0.091, "mu": 0.16905, "dt": 3600.0},
        )
        d = r.json()
        ok = r.status_code == 200 and d.get("status") == "SYMBOLIC_RECURSION_STABLE"
        record("Gateway SRC offload endpoint", ok, f"dx_final={d.get('dx_final')}, converged={d.get('converged')}")


async def test_gateway_stk() -> None:
    async with httpx.AsyncClient(timeout=60.0) as c:
        r = await c.post(f"{GW_URL}/api/physics/stk/simulate", json={"steps": 500, "dt": 0.05})
        d = r.json()
        ok = r.status_code == 200 and "trustAnchor" in d
        record("Gateway STK simulation", ok, f"finalState={d.get('finalState')}, anchor={str(d.get('trustAnchor',''))[:16]}...")


async def test_src_gateway_bridge() -> None:
    async with httpx.AsyncClient(timeout=20.0) as c:
        r = await c.get(f"{SRC_URL}/api/gateway/status")
        d = r.json()
        record("SRC→Gateway status bridge", d.get("connected") is True, json.dumps(d.get("health", {})))


async def test_src_gateway_offload() -> None:
    async with httpx.AsyncClient(timeout=60.0) as c:
        r = await c.post(f"{SRC_URL}/api/gateway/offload", params={"t": 3.5, "x0": 150.0})
        d = r.json()
        ok = r.status_code == 200 and "psi" in d
        record("SRC gateway offload proxy", ok, f"psi_status={d.get('psi',{}).get('status')}")


async def test_primal_engines_direct() -> None:
    from symbolic_recursion.integrations.primal_bridge import PrimalBridge
    from symbolic_recursion.integrations.atlas_bridge import AtlasBridge
    from symbolic_recursion.kernel import SymbolicRecursionKernel

    bridge = PrimalBridge()
    avail = bridge.engines_available
    q = np.array([1.0, 2.0, -0.5, 1.5, 0.3])
    result = bridge.process_q_series(q.tolist(), dt=1.0)
    dx = SymbolicRecursionKernel().integrate_series(q, dt=1.0)
    atlas = AtlasBridge().fuse_with_symbolic_recursion(q, dx)
    ok = avail["primal_trading"] and avail["primallang_physics"] and "physics" in result
    record(
        "Direct primal-trading + primallang physics",
        ok,
        f"ewic={result.get('ewic_control')}, eigenfreq={result.get('physics',{}).get('eigenfrequency_hz')}",
        atlas_offset=atlas.get("offset_from_attractor"),
    )


async def test_unit_tests() -> None:
    import subprocess
    proc = subprocess.run(
        [sys.executable, "-m", "pytest", "tests/", "-q"],
        cwd=r"C:\Users\stlta\symbolic-recursion-coherence",
        capture_output=True,
        text=True,
    )
    ok = proc.returncode == 0
    record("Unit test suite", ok, proc.stdout.strip().split("\n")[-1] if proc.stdout else proc.stderr[:80])


async def main() -> int:
    print("=" * 60)
    print("SYMBOLIC RECURSION COHERENCE — FULL INTEGRATION TEST")
    print("=" * 60)

    print("\n[1/3] Waiting for services...")
    src_up = await wait_for(SRC_URL)
    gw_up = await wait_for(GW_URL, "/v1/health")
    record("SRC API reachable", src_up, SRC_URL)
    record("Gateway reachable", gw_up, GW_URL)
    if not src_up:
        print("\nABORT: Start SRC API with: python -m symbolic_recursion.cli serve")
        return 1

    print("\n[2/3] Running integration tests...")
    await test_unit_tests()
    await test_src_health()
    if gw_up:
        await test_gateway_health()
    await test_kernel_examples()
    await test_benchmarks()
    await test_compute_dx_meta()
    await test_live_feed()
    await test_feed_series_pipeline()
    await test_primal_engines_direct()
    if gw_up:
        await test_gateway_psi_rpo()
        await test_gateway_src_offload()
        await test_gateway_stk()
        await test_src_gateway_bridge()
        await test_src_gateway_offload()
    else:
        record("Gateway tests", False, "skipped — gateway not running")

    print("\n[3/3] Summary")
    passed = sum(1 for r in results if r.passed)
    failed = sum(1 for r in results if not r.passed)
    print(f"  Total: {len(results)} | Passed: {passed} | Failed: {failed}")
    print("=" * 60)

    report_path = r"C:\Users\stlta\symbolic-recursion-coherence\integration_report.json"
    with open(report_path, "w") as f:
        json.dump(
            [{"name": r.name, "passed": r.passed, "detail": r.detail, "data": r.data} for r in results],
            f,
            indent=2,
        )
    print(f"Report saved: {report_path}")
    return 0 if failed == 0 else 1


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