"""Verify the tutorial against a new, isolated local Redis process."""
import importlib.metadata
import json
import os
from pathlib import Path
import socket
import subprocess
import sys
import tempfile
import time
from datetime import datetime, timezone

import redis
import memory_agent as agent

ROOT = Path.cwd()
checks = []


def check(name, condition, detail=None):
    checks.append({"name": name, "passed": bool(condition), "detail": detail})
    if not condition:
        raise AssertionError(name)


def run():
    root = ROOT / "test-runs"
    root.mkdir(exist_ok=True)
    run_dir = Path(tempfile.mkdtemp(prefix="redis-", dir=root))
    with socket.socket() as sock:
        sock.bind(("127.0.0.1", 0))
        port = sock.getsockname()[1]
    proc = subprocess.Popen([
        "redis-server", "--bind", "127.0.0.1", "--port", str(port),
        "--protected-mode", "yes", "--dir", str(run_dir), "--save", "",
        "--appendonly", "yes", "--appendfsync", "everysec",
        "--maxmemory", "64mb", "--maxmemory-policy", "allkeys-lru",
    ], stdout=(run_dir / "redis.log").open("w"), stderr=subprocess.STDOUT)
    env = os.environ.copy()
    env["REDIS_URL"] = f"redis://127.0.0.1:{port}/0"
    env_file = ROOT / ".env"
    for line in (env_file.read_text().splitlines() if env_file.exists() else []):
        if line.startswith("OPENROUTER_API_KEY="):
            env["OPENROUTER_API_KEY"] = line.split("=", 1)[1].strip().strip("\"'")
    client = redis.Redis.from_url(env["REDIS_URL"], decode_responses=True)

    def cli(*args, ok=True):
        result = subprocess.run([sys.executable, str(Path(__file__).with_name("memory_agent.py")), *args],
                                env=env, capture_output=True, text=True, timeout=60)
        if ok and result.returncode:
            raise RuntimeError(result.stderr.strip())
        return result

    try:
        for _ in range(50):
            try:
                if client.ping():
                    break
            except redis.ConnectionError:
                time.sleep(0.1)
        version = client.info("server")["redis_version"]
        cli("remember", "--user", "alice", "--preference", "vegetarian")
        saved = json.loads(cli("inspect", "--user", "alice").stdout)
        check("Preference survives separate Python processes", saved["profile"] == {"dietary_preference": "vegetarian"})
        profile, history = agent.keys("alice", "session-1")
        check("Profile gets 30-day expiry", 2591940 <= client.ttl(profile) <= 2592000)
        first = cli("chat", "--user", "alice", "--session", "session-1", "--message", "What dietary preference have I saved?").stdout.strip()
        check("Live model recalls stored preference", "vegetarian" in first.lower(), first)
        check("Chat writes a user and assistant pair", client.llen(history) == 2)
        check("Chat gets 24-hour expiry", 86340 <= client.ttl(history) <= 86400)
        session2 = json.loads(cli("inspect", "--user", "alice", "--session", "session-2").stdout)
        check("New session has profile and no old transcript", session2["profile"] == saved["profile"] and session2["history"] == [])
        second = cli("chat", "--user", "alice", "--session", "session-2", "--message", "What dietary preference have I saved?").stdout.strip()
        check("Live model recalls preference in a new process and session", "vegetarian" in second.lower(), second)
        bob = json.loads(cli("inspect", "--user", "bob", "--session", "session-1").stdout)
        check("Another user's context is empty", bob == {"profile": {}, "history": []})
        bob_answer = cli("chat", "--user", "bob", "--session", "session-1", "--message", "What dietary preference have I saved? If unknown, say unknown.").stdout.strip()
        check("Live model has no saved preference for another user", "unknown" in bob_answer.lower(), bob_answer)
        for i in range(14):
            agent.save_turn(client, "bounded", "chat", f"question {i}", f"answer {i}")
        _, rows = agent.context(client, "bounded", "chat")
        check("History retains 10 complete turns", len(rows) == 20 and rows[0]["content"] == "question 4" and rows[-1]["content"] == "answer 13")
        invalid = cli("inspect", "--user", "alice}:*", ok=False)
        check("IDs cannot inject a key prefix or scan pattern", invalid.returncode != 0)
        agent.remember(client, "expires", "vegan")
        ep, _ = agent.keys("expires", "chat")
        client.expire(ep, 1)
        agent.save_turn(client, "expires", "chat", "hello", "hi")
        _, eh = agent.keys("expires", "chat")
        client.expire(eh, 1)
        time.sleep(1.1)
        check("Expired profile and session are absent", agent.context(client, "expires", "chat") == ({}, []))
        cli("forget-user", "--user", "alice")
        check("Forget removes profile and both sessions", list(client.scan_iter(match="agentmem:v1:{alice}:*")) == [])
        check("Forget leaves another user's session intact", len(agent.context(client, "bob", "session-1")[1]) == 2)
        forgotten = cli("chat", "--user", "alice", "--session", "session-3", "--message", "What dietary preference have I saved? If unknown, say unknown.").stdout.strip()
        check("Live model no longer sees forgotten profile", "unknown" in forgotten.lower(), forgotten)
        before = agent.context(client, "alice", "session-3")
        rejected = cli("chat", "--user", "alice", "--session", "session-3", "--message", "x" * 2001, ok=False)
        check("Rejected oversized input leaves history unchanged",
              rejected.returncode != 0 and agent.context(client, "alice", "session-3") == before)
        print(json.dumps({"checks": len(checks), "passed": sum(x["passed"] for x in checks), "redis": version}))
    finally:
        proc.terminate()
        proc.wait(timeout=10)
        (ROOT / "validation.json").write_text(json.dumps({
            "testedAt": datetime.now(timezone.utc).isoformat(),
            "python": sys.version.split()[0],
            "redisServer": locals().get("version"), "redisPy": importlib.metadata.version("redis"),
            "model": env.get("OPENROUTER_MODEL", "openai/gpt-4.1-mini"),
            "scope": "Isolated local Redis; four live OpenRouter calls; separate CLI processes. No hosted instance, web authentication, concurrent turns or disaster recovery test.",
            "checks": checks,
        }, indent=2) + "\n")


if __name__ == "__main__":
    run()