validate.py

"""
validate.py — Pre-submission OpenEnv compliance validator.
Run this before submitting to catch any issues.

Usage:
    python validate.py
"""

from __future__ import annotations

import json
import sys
import traceback
from typing import Any, Dict, List, Tuple

PASS = "✓"
FAIL = "✗"
WARN = "⚠"

results: List[Tuple[str, str, str]] = []   # (status, check, detail)


def check(label: str, condition: bool, detail: str = "", warn_only: bool = False) -> bool:
    status = PASS if condition else (WARN if warn_only else FAIL)
    results.append((status, label, detail))
    return condition


# ---------------------------------------------------------------------------
# 1. File presence
# ---------------------------------------------------------------------------

import os

REQUIRED_FILES = [
    "inference.py",
    "openenv.yaml",
    "Dockerfile",
    "README.md",
    "requirements.txt",
    "env/environment.py",
    "env/models.py",
    "env/tasks.py",
    "env/grader.py",
    "env/__init__.py",
    "app.py",
    "LICENSE",
]

print("\n── File presence ──────────────────────────────────────")
for f in REQUIRED_FILES:
    check(f"File exists: {f}", os.path.isfile(f))

# ---------------------------------------------------------------------------
# 2. openenv.yaml
# ---------------------------------------------------------------------------

print("\n── openenv.yaml ───────────────────────────────────────")
try:
    import yaml
    with open("openenv.yaml") as fh:
        spec = yaml.safe_load(fh)

    check("Has 'name' field",        "name"    in spec, spec.get("name", ""))
    check("Has 'version' field",     "version" in spec, spec.get("version", ""))
    check("Has 'tasks' list",        "tasks"   in spec and len(spec["tasks"]) >= 3,
          f"{len(spec.get('tasks', []))} task(s)")
    check("Has 'openenv' tag",       "openenv" in spec.get("tags", []))
    check("Has 'reward' section",    "reward"  in spec)
    check("Has 'endpoints' section", "endpoints" in spec)

    tasks = spec.get("tasks", [])
    difficulties = [t.get("difficulty") for t in tasks]
    check("Has easy task",   "easy"   in difficulties)
    check("Has medium task", "medium" in difficulties)
    check("Has hard task",   "hard"   in difficulties)
except Exception as e:
    check("openenv.yaml parseable", False, str(e))

# ---------------------------------------------------------------------------
# 3. Pydantic models
# ---------------------------------------------------------------------------

print("\n── Pydantic models ────────────────────────────────────")
try:
    from env.models import (
        ModerationAction, ModerationObservation, ModerationReward,
        ModerationDecision, ViolationCategory, Severity,
        ContentItem, UserContext, PolicyRule,
    )
    check("ModerationObservation importable", True)
    check("ModerationAction importable",      True)
    check("ModerationReward importable",      True)

    # Instantiation test
    from env.models import ContentType, UserReputation
    content = ContentItem(
        content_id="v001", content_type=ContentType.TEXT,
        text="test", reported_count=0, view_count=100, engagement_rate=0.1, has_links=False,
    )
    user = UserContext(user_id="u001", reputation=UserReputation.TRUSTED, account_age_days=100)
    obs = ModerationObservation(
        content=content, user_context=user, applicable_rules=[],
        queue_position=1, queue_total=5, items_processed=0,
        correct_so_far=0, step_number=0, task_id="test", task_difficulty="easy",
    )
    check("ModerationObservation instantiates", True)

    action = ModerationAction(
        content_id="v001",
        decision=ModerationDecision.APPROVE,
        violation_category=ViolationCategory.NONE,
        severity_assessment=Severity.LOW,
        confidence=0.9,
    )
    check("ModerationAction instantiates", True)

    reward = ModerationReward(
        decision_score=0.6, category_score=0.25, severity_score=0.15,
        efficiency_bonus=0.0, consistency_bonus=0.0, penalty=0.0,
        total=1.0, explanation="test",
    )
    check("ModerationReward instantiates", True)

except Exception as e:
    check("Models import/instantiate", False, traceback.format_exc(limit=3))

# ---------------------------------------------------------------------------
# 4. Environment API — reset / step / state
# ---------------------------------------------------------------------------

print("\n── Environment API ────────────────────────────────────")
try:
    from env.environment import ContentModerationEnv
    env = ContentModerationEnv()
    check("ContentModerationEnv instantiates", True)

    for task_name in ["easy", "medium", "hard"]:
        obs = env.reset(task=task_name)
        check(f"reset('{task_name}') returns ModerationObservation",
              isinstance(obs, ModerationObservation),
              f"type={type(obs).__name__}")

        # Step with a valid action
        action = ModerationAction(
            content_id=obs.content.content_id,
            decision=ModerationDecision.ESCALATE,
            violation_category=ViolationCategory.NONE,
            severity_assessment=Severity.LOW,
            confidence=0.5,
        )
        result = env.step(action)
        check(f"step() returns 4-tuple for '{task_name}'", len(result) == 4)
        _obs, reward, done, info = result
        check(f"reward is float for '{task_name}'",     isinstance(reward, float))
        check(f"done is bool for '{task_name}'",        isinstance(done, bool))
        check(f"info is dict for '{task_name}'",        isinstance(info, dict))
        check(f"reward in [-1, 1] for '{task_name}'",  -1.0 <= reward <= 1.0,
              f"reward={reward:.4f}")

        state = env.state()
        check(f"state() returns dict for '{task_name}'", isinstance(state, dict))
        check(f"state() has 'task_id' for '{task_name}'", "task_id" in state)

except Exception as e:
    check("Environment API", False, traceback.format_exc(limit=5))

# ---------------------------------------------------------------------------
# 5. Graders
# ---------------------------------------------------------------------------

print("\n── Graders ────────────────────────────────────────────")
try:
    from env.grader import GRADERS, grade_single_action
    from env.tasks import ALL_TASKS

    for difficulty, grader_fn in GRADERS.items():
        task_def = ALL_TASKS[difficulty]
        items    = task_def["items"]

        # Build a set of decisions using ground-truth answers
        decisions = [
            {
                "action": ModerationAction(
                    content_id=item["content"].content_id,
                    decision=item["gt_decision"],
                    violation_category=item["gt_category"],
                    severity_assessment=item["gt_severity"],
                    confidence=0.95,
                ),
                "gt_decision": item["gt_decision"],
                "gt_category": item["gt_category"],
                "gt_severity": item["gt_severity"],
            }
            for item in items
        ]
        result = grader_fn(decisions)
        check(f"grade_{difficulty}() returns EpisodeResult", hasattr(result, "final_score"))
        check(f"grade_{difficulty}() score in [0,1]",
              0.0 <= result.final_score <= 1.0, f"score={result.final_score:.4f}")
        check(f"grade_{difficulty}() score is deterministic",
              grader_fn(decisions).final_score == result.final_score)

        # Perfect run should score > 0.9
        check(f"grade_{difficulty}() perfect run scores >0.90",
              result.final_score >= 0.90, f"got {result.final_score:.4f}", warn_only=True)

        # Random (bad) decisions should score lower than perfect
        bad_decisions = [
            {
                "action": ModerationAction(
                    content_id=item["content"].content_id,
                    decision=ModerationDecision.APPROVE,
                    violation_category=ViolationCategory.NONE,
                    severity_assessment=Severity.LOW,
                    confidence=0.1,
                ),
                "gt_decision": item["gt_decision"],
                "gt_category": item["gt_category"],
                "gt_severity": item["gt_severity"],
            }
            for item in items
        ]
        bad_result = grader_fn(bad_decisions)
        check(f"grade_{difficulty}() bad run scores lower than perfect",
              bad_result.final_score < result.final_score,
              f"bad={bad_result.final_score:.4f} vs perfect={result.final_score:.4f}")

except Exception as e:
    check("Graders", False, traceback.format_exc(limit=5))

# ---------------------------------------------------------------------------
# 6. Task dataset integrity
# ---------------------------------------------------------------------------

print("\n── Task dataset ───────────────────────────────────────")
try:
    from env.tasks import ALL_TASKS

    for difficulty, task_def in ALL_TASKS.items():
        items = task_def["items"]
        check(f"Task '{difficulty}' has ≥3 items", len(items) >= 3, f"{len(items)} items")

        ids = [item["content"].content_id for item in items]
        check(f"Task '{difficulty}' all content_ids unique",
              len(ids) == len(set(ids)), f"ids={ids}")

        for item in items:
            assert "gt_decision" in item, f"Missing gt_decision in {item}"
            assert "gt_category" in item, f"Missing gt_category in {item}"
            assert "gt_severity" in item, f"Missing gt_severity in {item}"
        check(f"Task '{difficulty}' all items have ground truth", True)

except Exception as e:
    check("Task dataset integrity", False, traceback.format_exc(limit=3))

# ---------------------------------------------------------------------------
# 7. Dockerfile basics
# ---------------------------------------------------------------------------

print("\n── Dockerfile ─────────────────────────────────────────")
with open("Dockerfile") as fh:
    df = fh.read()
check("Dockerfile EXPOSE 7860",  "EXPOSE 7860"  in df)
check("Dockerfile HEALTHCHECK",  "HEALTHCHECK"  in df)
check("Dockerfile non-root user", "useradd"     in df or "USER" in df)
check("Dockerfile copies requirements", "COPY" in df and "requirements" in df)

# ---------------------------------------------------------------------------
# 8. inference.py compliance
# ---------------------------------------------------------------------------

print("\n── inference.py ───────────────────────────────────────")
with open("inference.py") as fh:
    inf = fh.read()
check("Uses 'from openai import OpenAI'", "from openai import OpenAI" in inf)
check("Reads API_BASE_URL env var",       "API_BASE_URL" in inf and "os.getenv" in inf)
check("Reads MODEL_NAME env var",         "MODEL_NAME"   in inf)
check("Reads HF_TOKEN env var",           "HF_TOKEN"     in inf)
check("Has TEMPERATURE constant",         "TEMPERATURE"  in inf)
check("Covers all 3 tasks",
      all(t in inf for t in ['"easy"', '"medium"', '"hard"']))

# ---------------------------------------------------------------------------
# Summary
# ---------------------------------------------------------------------------

print("\n" + "="*56)
print("  VALIDATION SUMMARY")
print("="*56)
passed = sum(1 for s, _, _ in results if s == PASS)
warned = sum(1 for s, _, _ in results if s == WARN)
failed = sum(1 for s, _, _ in results if s == FAIL)

for status, label, detail in results:
    suffix = f"  [{detail}]" if detail else ""
    print(f"  {status}  {label}{suffix}")

print(f"\n  Passed: {passed}  |  Warnings: {warned}  |  Failed: {failed}")

if failed == 0:
    print("\n  🎉 ALL CHECKS PASSED — submission is ready!")
    sys.exit(0)
else:
    print(f"\n  ❌ {failed} check(s) failed — fix before submitting.")
    sys.exit(1)