tests/test_integration.py

"""Gate 2 — the four organs compose and the BackboneAdapter contract drives the full loop end-to-end
on a trivial backbone (no real model). Proves the runtime works before expensive backbones.

Author - Deepak Soni
"""
import os, sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))
import torch
from antahkarana.core import Antahkarana
from antahkarana.domains.mock import MockBackbone, make_synthetic_stream

R = []
def check(name, ok):
    R.append(ok); print(("PASS" if ok else "FAIL") + "  " + name)

torch.manual_seed(0)
stream = make_synthetic_stream(T=3, seed=0)
mind = Antahkarana(MockBackbone(), epochs=3, bs=16, samskara=True, replay=True, sleep=True)
res = mind.run(stream)

check("loop returns result dict", isinstance(res, dict))
check("matrix is T×T", len(res["matrix"]) == 3 and all(len(r) == 3 for r in res["matrix"]))
check("no NaN in matrix", all(v == v for r in res["matrix"] for v in r))
check("final_avg in [0,1]", 0.0 <= res["final_avg"] <= 1.0)
check("forgetting is finite float", isinstance(res["forgetting"], float))
check("risk_coverage populated", "1.0" in res["risk_coverage"] and "0.7" in res["risk_coverage"])
check("trajectory length == T", len(res["trajectory"]) == 3)
check("all 4 organs fired per task", all({"nu", "guna", "boundary"} <= set(t) for t in res["trajectory"]))
check("guna: task0 -> rajas", res["trajectory"][0]["guna"] == "rajas")
check("ahamkara: task0 is boundary", res["trajectory"][0]["boundary"] is True)
check("citta sleep ran (recovery measured)", res["recovery"] is not None and "task0_delta" in res["recovery"])
check("learned above chance (0.5)", res["final_avg"] > 0.55)

ok = all(R)
print("\nG2 INTEGRATION: " + ("ALL PASS (%d/%d)" % (sum(R), len(R)) if ok else "FAIL (%d/%d)" % (sum(R), len(R))))
print("  final_avg=%.3f  forgetting=%+.4f  guna_traj=%s" %
      (res["final_avg"], res["forgetting"], [t["guna"] for t in res["trajectory"]]))
sys.exit(0 if ok else 1)