"""Reproducible synthetic functional checks; not a product benchmark.

Run using Python 3.12 with ray==2.59.0 and wandb==0.30.0 installed.
No cloud account, production data, GPU, API key or cloud sync is used.
"""
import hashlib
import importlib.metadata
import json
import os
from pathlib import Path
import platform
import sys
import tempfile
import time
import traceback

os.environ["RAY_USAGE_STATS_ENABLED"] = "0"
os.environ["WANDB_MODE"] = "offline"
os.environ["WANDB_SILENT"] = "true"
out = Path(sys.argv[2]).resolve()
out.mkdir(parents=True, exist_ok=True)
product = sys.argv[1]
result = {
    "product": product,
    "started_utc": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
    "environment": {"python": platform.python_version(), "os": platform.system(),
                    "os_version": platform.mac_ver()[0], "architecture": platform.machine()},
    "package_version": importlib.metadata.version(product),
    "scope": "Synthetic local functional smoke test; no cloud or production benchmark",
    "checks": [],
}
try:
    if product == "ray":
        import ray
        ray.init(num_cpus=2, num_gpus=0, include_dashboard=False,
                 _node_ip_address="127.0.0.1", _temp_dir=tempfile.mkdtemp(prefix="review-ray-", dir="/tmp"),
                 log_to_driver=False)
        @ray.remote
        def square(n):
            return n * n
        values = ray.get([square.remote(n) for n in range(8)], timeout=30)
        assert values == [0, 1, 4, 9, 16, 25, 36, 49]
        result["checks"].append({"check": "eight CPU task results", "passed": True, "observed": values})
        @ray.remote
        class Counter:
            def __init__(self):
                self.value = 0
            def increment(self):
                self.value += 1
                return self.value
        actor = Counter.remote()
        counts = ray.get([actor.increment.remote() for _ in range(3)], timeout=30)
        assert counts == [1, 2, 3]
        result["checks"].append({"check": "actor state retained across calls", "passed": True, "observed": counts})
        @ray.remote
        def fail():
            raise ValueError("intentional synthetic failure")
        try:
            ray.get(fail.remote(), timeout=30)
            raise AssertionError("Expected task error was not propagated")
        except ray.exceptions.RayTaskError as error:
            assert "intentional synthetic failure" in str(error)
            result["checks"].append({"check": "task exception reaches caller", "passed": True})
        result["limitations"] = ["One local machine with two CPU workers", "No GPU or multiple-node scaling",
            "No performance comparison, Ray Serve deployment, dashboard or production reliability assessment"]
        ray.shutdown()
    elif product == "wandb":
        import wandb
        synthetic = out / "synthetic-input.json"
        synthetic.write_text(json.dumps({"kind": "synthetic", "values": [1, 2, 3]}))
        run = wandb.init(project="review-synthetic-offline", mode="offline", dir=str(out),
            config={"dataset": "synthetic", "learning_rate": 0.1},
            settings=wandb.Settings(disable_code=True, disable_git=True,
                x_disable_meta=True, x_disable_stats=True, x_disable_machine_info=True))
        assert run.settings.mode == "offline"
        result["checks"].append({"check": "offline mode selected", "passed": True})
        observations = [{"step": 0, "loss": 1.0}, {"step": 1, "loss": 0.5}, {"step": 2, "loss": 0.25}]
        for item in observations:
            run.log({"synthetic_loss": item["loss"]}, step=item["step"])
        run.summary["synthetic_final_loss"] = 0.25
        assert run.config["learning_rate"] == 0.1
        assert run.summary["synthetic_final_loss"] == 0.25
        artifact = wandb.Artifact("synthetic-input", type="dataset")
        artifact.add_file(str(synthetic))
        run.log_artifact(artifact)
        run.finish()
        records = list(out.glob("wandb/offline-run-*/run-*.wandb"))
        assert len(records) == 1 and records[0].stat().st_size > 0
        # Check that the local persisted stream contains the submitted metric and artifact identifiers.
        # This is deliberately not a cloud retrieval, UI or cross-device reproduction test.
        raw = records[0].read_bytes()
        assert b"synthetic_loss" in raw and b"synthetic_final_loss" in raw
        assert b"synthetic-input" in raw
        result["checks"].append({"check": "synthetic configuration and summary accepted", "passed": True,
                                  "observed_final_loss": 0.25})
        result["checks"].append({"check": "nonempty local run stream contains metric and artifact identifiers",
            "passed": True, "record_bytes": len(raw), "record_sha256": hashlib.sha256(raw).hexdigest(),
            "submitted_metrics": observations})
        result["limitations"] = ["Offline SDK only; no cloud login or sync", "Synthetic values, not model training",
            "No cloud dashboard, team collaboration, registry retrieval, Weave or production benchmark",
            "Persisted stream checked for identifiers, not independently reconstructed metric history"]
    else:
        raise ValueError(product)
    result["passed"] = True
except Exception:
    result["passed"] = False
    result["error"] = traceback.format_exc()
finally:
    if product == "ray":
        try:
            import ray
            ray.shutdown()
        except Exception:
            pass
    result["finished_utc"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
    (out / (product + "-results.json")).write_text(json.dumps(result, indent=2))
    print(json.dumps(result, indent=2))
sys.exit(0 if result.get("passed") else 1)