validate.py

"""
validate.py — OpenEnv Spec Compliance Validator
================================================

Run with:   python validate.py
Checks:
  1. All Pydantic models have correct fields and types
  2. reset() returns a valid InterviewObservation
  3. step() returns a valid StepResult with reward in range
  4. state() returns a valid InterviewState
  5. All 3 tasks complete full episodes
  6. Reward is always in [-0.05, 1.0]
  7. done=True only after final question
  8. Partial rewards are non-trivial (not all same score)
  9. Episode boundary: step() after done returns done=True
 10. openenv.yaml parses correctly and has required fields
"""
from __future__ import annotations

import sys
import yaml
from pathlib import Path
from typing import List

from interview_env import (
    InterviewEnv,
    InterviewAction,
    InterviewObservation,
    InterviewState,
    StepResult,
    TASK_REGISTRY,
)
from interview_env.models import InterviewReward

PASS = "  ✅"
FAIL = "  ❌"
results: List[tuple[str, bool, str]] = []


def check(name: str, condition: bool, detail: str = ""):
    status = PASS if condition else FAIL
    results.append((name, condition, detail))
    print(f"{status}  {name}" + (f"  ({detail})" if detail else ""))


def run_checks():
    print("\n══════════════════════════════════════════════")
    print("  OpenEnv Validator — AI Interview Simulator")
    print("══════════════════════════════════════════════\n")

    # ── 1. Model field existence ───────────────────────────────────────────
    print("── Model structure ──")
    obs_fields = InterviewObservation.model_fields.keys()
    check("InterviewObservation has current_question",   "current_question"   in obs_fields)
    check("InterviewObservation has task_id",            "task_id"            in obs_fields)
    check("InterviewObservation has stage",              "stage"              in obs_fields)
    check("InterviewObservation has difficulty",         "difficulty"         in obs_fields)
    check("InterviewObservation has interview_complete", "interview_complete" in obs_fields)

    reward_fields = InterviewReward.model_fields.keys()
    check("InterviewReward has total_score",      "total_score"      in reward_fields)
    check("InterviewReward has relevance",        "relevance"        in reward_fields)
    check("InterviewReward has structure",        "structure"        in reward_fields)
    check("InterviewReward has depth",            "depth"            in reward_fields)
    check("InterviewReward has professionalism",  "professionalism"  in reward_fields)
    check("InterviewReward has feedback",         "feedback"         in reward_fields)

    state_fields = InterviewState.model_fields.keys()
    check("InterviewState has qa_history",     "qa_history"     in state_fields)
    check("InterviewState has episode_done",   "episode_done"   in state_fields)
    check("InterviewState has cumulative_reward", "cumulative_reward" in state_fields)

    step_fields = StepResult.model_fields.keys()
    check("StepResult has observation", "observation" in step_fields)
    check("StepResult has reward",      "reward"      in step_fields)
    check("StepResult has done",        "done"        in step_fields)
    check("StepResult has info",        "info"        in step_fields)
    print()

    # ── 2. reset() contract ───────────────────────────────────────────────
    print("── reset() contract ──")
    env = InterviewEnv(task_id="junior_behavioral")
    obs = env.reset()
    check("reset() returns InterviewObservation",        isinstance(obs, InterviewObservation))
    check("reset() sets interview_complete=False",       obs.interview_complete is False)
    check("reset() question_number starts at 1",         obs.question_number == 1)
    check("reset() cumulative_score starts at 0.0",      obs.cumulative_score == 0.0)
    check("reset() previous_qa is empty",                obs.previous_qa == [])
    check("reset() current_question is non-empty string",isinstance(obs.current_question, str) and len(obs.current_question) > 5)

    # Double-reset should restart cleanly
    obs2 = env.reset()
    check("double reset() restarts cleanly",             obs2.question_number == 1 and obs2.cumulative_score == 0.0)
    print()

    # ── 3. step() contract ───────────────────────────────────────────────
    print("── step() contract ──")
    env = InterviewEnv(task_id="junior_behavioral")
    env.reset()
    action = InterviewAction(response="I once worked on a project where I fixed a bug.")
    result = env.step(action)

    check("step() returns StepResult",          isinstance(result, StepResult))
    check("step() reward in [-0.05, 1.0]",      -0.05 <= result.reward <= 1.0, f"got {result.reward}")
    check("step() done is bool",                 isinstance(result.done, bool))
    check("step() observation is InterviewObservation", isinstance(result.observation, InterviewObservation))
    check("step() info is dict",                 isinstance(result.info, dict))
    check("step() info has grader_breakdown",    "grader_breakdown" in result.info)
    check("step() info has feedback",            "feedback" in result.info)
    check("step() info grader has 4 dimensions", all(
        k in result.info["grader_breakdown"]
        for k in ["relevance", "structure", "depth", "professionalism"]
    ))
    print()

    # ── 4. state() contract ──────────────────────────────────────────────
    print("── state() contract ──")
    state = env.state()
    check("state() returns InterviewState",      isinstance(state, InterviewState))
    check("state() is JSON-serialisable",        bool(state.model_dump()))
    check("state() qa_history has 1 entry",      len(state.qa_history) == 1)
    check("state() episode_done is False mid-episode", state.episode_done is False)
    print()

    # ── 5. Full episode for all 3 tasks ──────────────────────────────────
    print("── Full episode per task ──")
    ANSWERS = {
        "easy":   (
            "In that situation I was on a team building a new feature. "
            "My task was to implement the backend API endpoint. "
            "I took action by writing clean code with tests and documentation. "
            "The result was a successful delivery appreciated by stakeholders."
        ),
        "medium": (
            "First, I would analyse the architecture and identify bottlenecks. "
            "The trade-off between REST and GraphQL is flexibility vs simplicity. "
            "I used Redis for caching and PostgreSQL with proper indexing to improve "
            "query performance. The result reduced latency by 40ms on the p99."
        ),
        "hard":   (
            "For this distributed system design, first I would implement a Redis-based "
            "sliding window rate limiter with Lua scripts for atomicity. "
            "The architecture uses sharded Redis cluster with replication for fault tolerance. "
            "Trade-off: sliding window uses more memory but provides accurate rate limiting. "
            "Fallback: local token bucket when Redis is partitioned. "
            "Result: handles 100k RPS with sub-millisecond overhead."
        ),
    }

    for task_id, task in TASK_REGISTRY.items():
        env = InterviewEnv(task_id=task_id)
        obs = env.reset()
        rewards = []
        steps = 0

        while not obs.interview_complete:
            answer  = ANSWERS[task.difficulty.value]
            result  = env.step(InterviewAction(response=answer))
            rewards.append(result.reward)
            steps  += 1
            obs     = result.observation

        final_state = env.state()
        episode_score = sum(rewards) / len(rewards) if rewards else 0.0

        check(
            f"[{task_id}] completes in {task.total_questions} steps",
            steps == task.total_questions,
            f"got {steps}",
        )
        check(
            f"[{task_id}] episode_done=True after last step",
            final_state.episode_done,
        )
        check(
            f"[{task_id}] all rewards in range",
            all(-0.05 <= r <= 1.0 for r in rewards),
            f"rewards={[round(r,2) for r in rewards]}",
        )
        check(
            f"[{task_id}] partial rewards (not all identical)",
            len(set(round(r, 3) for r in rewards)) > 1 or steps == 1,
            f"score={episode_score:.3f}",
        )
        check(
            f"[{task_id}] qa_history length matches steps",
            len(final_state.qa_history) == steps,
        )
    print()

    # ── 6. Episode boundary ───────────────────────────────────────────────
    print("── Episode boundary ──")
    env = InterviewEnv(task_id="junior_behavioral")
    env.reset()
    for _ in range(10):  # exhaust all questions (4 + extra)
        r = env.step(InterviewAction(response="test answer here for boundary check"))
    check("step() after done still returns done=True", r.done is True)
    check("step() after done reward is 0.0",           r.reward == 0.0)
    print()

    # ── 7. Reward shaping — quality sensitivity ───────────────────────────
    print("── Reward shaping quality ──")
    from interview_env.graders import grade_response
    from interview_env.tasks import TASK_JUNIOR_BEHAVIORAL
    q = TASK_JUNIOR_BEHAVIORAL.questions[1]  # debug question

    bad   = grade_response("idk", q).total_score
    med   = grade_response(
        "I debugged a problem once by checking the logs and fixed it.", q
    ).total_score
    good  = grade_response(
        "In a situation where our API was returning 500 errors, my task was to diagnose "
        "the root cause. I took action by reproducing the bug locally, adding debug logging, "
        "and isolating it to a race condition in the database connection pool. "
        "I fixed it by implementing proper connection timeouts. The result was zero errors "
        "in production and I learned to always add integration tests for concurrency.", q
    ).total_score

    check("Bad answer scores lower than medium",  bad  < med,  f"{bad:.3f} < {med:.3f}")
    check("Medium answer scores lower than good", med  < good, f"{med:.3f} < {good:.3f}")
    check("Bad answer score < 0.35",              bad  < 0.35, f"got {bad:.3f}")
    check("Good answer score > 0.65",             good > 0.65, f"got {good:.3f}")
    print()

    # ── 8. openenv.yaml ───────────────────────────────────────────────────
    print("── openenv.yaml ──")
    yaml_path = Path("openenv.yaml")
    check("openenv.yaml exists", yaml_path.exists())
    if yaml_path.exists():
        with open(yaml_path) as f:
            meta = yaml.safe_load(f)
        check("yaml has 'name'",    "name"    in meta)
        check("yaml has 'version'", "version" in meta)
        check("yaml has 'tasks'",   "tasks"   in meta)
        check("yaml has 3+ tasks",  len(meta.get("tasks", [])) >= 3)
        check("yaml has reward spec", "reward" in meta)
        check("yaml has interface",   "interface" in meta)
    print()

    # ── Summary ───────────────────────────────────────────────────────────
    passed = sum(1 for _, ok, _ in results if ok)
    total  = len(results)
    failed = [name for name, ok, _ in results if not ok]

    print("══════════════════════════════════════════════")
    print(f"  Result: {passed}/{total} checks passed")
    if failed:
        print(f"\n  FAILED:")
        for name in failed:
            print(f"    ❌  {name}")
    else:
        print("  🎉 All checks passed! Environment is OpenEnv compliant.")
    print("══════════════════════════════════════════════\n")

    return len(failed) == 0


if __name__ == "__main__":
    ok = run_checks()
    sys.exit(0 if ok else 1)