🐑 Do LLMs Dream of INTJ? 🔮

"""Per-case judge for the personality/mbti_profile task.

The model takes a 32-question Likert MBTI questionnaire. The judge:

  1. Validates format strictly — must be JSON `{"responses": [32 ints 1..5]}`
  2. If valid, computes the MBTI 4-letter type + per-axis percentages
  3. Computes an acquiescence-bias flag (>80% agreement on reverse-coded pairs
     indicates the model is just saying yes to everything; the type is unreliable)

Score: 1.0 if format is valid. 0.0 if not. The derived `mbti_type` and
`percentages` are SURFACED IN METRICS so the leaderboard can show them, but
they are NOT graded — there's no canonical MBTI for an AI.

Reasoning behind format-only grading: the whole point of this task is to
PROFILE each model and compare across the leaderboard. Grading on a canonical
type would assume one exists, which it doesn't. The comparison is the value;
the judge just keeps the comparison apples-to-apples.
"""
from __future__ import annotations

import json
import os
import re
from pathlib import Path
from typing import Any


# Fields the solution may contribute through outputs_dir/usage.json. They travel into
# the case metrics because the leaderboard's profile card reads the model name from
# `model` and there is no other channel for it — trap's cost proxy covers Anthropic /
# OpenAI / Mistral / Moonshot, but not OpenRouter, where most of this task's models run.
# Whitelisted rather than merged wholesale: the solution writes that file itself, so an
# open merge would let it overwrite mbti_type / percentages / bias_stats with anything.
USAGE_FIELDS = (
    "model",
    # What was in front of the model besides the questionnaire — a CLAUDE.md, a
    # soul.md, an agent harness, or nothing. This has to travel per-run: the
    # platform identifies a solution by (commit, path) alone, so two runs of one
    # commit that differ only by environment collapse onto the same row identity
    # and the second one's name is discarded. Without this field, "same model,
    # different persona" renders as two identical-looking rows.
    "persona",
    "input_tokens",
    "output_tokens",
    "cache_creation_input_tokens",
    "cache_read_input_tokens",
    "usd_cost",
)


def _strip_fences(text: str) -> str:
    """Some LLMs wrap JSON in ```json...```. Strip it."""
    text = text.strip()
    if not text.startswith("```"):
        return text
    lines = text.split("\n")
    if lines and lines[0].startswith("```"):
        lines = lines[1:]
    if lines and lines[-1].startswith("```"):
        lines = lines[:-1]
    return "\n".join(lines).strip()


def _parse_output(stdout: str) -> tuple[dict | None, str]:
    s = _strip_fences(stdout)
    if not s:
        return None, "empty stdout"
    try:
        obj = json.loads(s)
    except json.JSONDecodeError as e:
        # Last-ditch: find first {...} substring containing "responses"
        m = re.search(r'\{[^{}]*"responses"[^{}]*\[[\s\S]*?\][^{}]*\}', s)
        if m:
            try:
                obj = json.loads(m.group(0))
            except json.JSONDecodeError:
                return None, f"could not parse JSON: {e}"
        else:
            return None, f"could not parse JSON: {e}"
    if not isinstance(obj, dict):
        return None, "top-level output must be a JSON object"
    return obj, ""


def derive_mbti(responses: list[int], scoring_key: list[dict], letters_by_axis: dict[str, list[str]]) -> dict:
    """Sum per-axis contributions. For each question:
      - if response==3: contribution=0 (neutral)
      - if direction==<positive letter of axis>: contribution = response - 3
      - else (negative direction): contribution = 3 - response
    Sum across all 8 questions per axis → range [-16, +16].
    Positive → first letter of axis (E/S/T/J); negative or zero → second (I/N/F/P).
    Percentage of first letter = (sum + 16) / 32 * 100."""
    sums: dict[str, int] = {}
    counts: dict[str, int] = {}
    for q in scoring_key:
        n = q["n"]
        axis = q["axis"]
        direction = q["direction"]
        first_letter = letters_by_axis[axis][0]   # e.g. "E"
        r = responses[n - 1]
        contribution = (r - 3) if direction == first_letter else (3 - r)
        sums[axis] = sums.get(axis, 0) + contribution
        counts[axis] = counts.get(axis, 0) + 1

    type_letters: list[str] = []
    percentages: dict[str, float] = {}
    for axis in ("E_I", "S_N", "T_F", "J_P"):
        max_abs = counts.get(axis, 8) * 2   # each q contributes -2..+2
        s = sums.get(axis, 0)
        # First letter = positive direction; second = negative or zero
        if s > 0:
            type_letters.append(letters_by_axis[axis][0])
        elif s < 0:
            type_letters.append(letters_by_axis[axis][1])
        else:
            # Exact tie — by convention, take the second (more introverted/I-side)
            type_letters.append(letters_by_axis[axis][1])
        # Percentage in favour of FIRST letter
        pct_first = round((s + max_abs) / (2 * max_abs) * 100, 1)
        percentages[axis] = {letters_by_axis[axis][0]: pct_first,
                             letters_by_axis[axis][1]: round(100 - pct_first, 1)}

    return {"mbti_type": "".join(type_letters), "percentages": percentages}


def acquiescence_score(responses: list[int]) -> dict:
    """Flag bias: if model agrees (≥4) with both positive AND its reverse-coded
    pair, that's contradictory. Count contradictions per axis."""
    # Mostly informational. Returns simple stats.
    n = len(responses)
    mean = sum(responses) / n if n else 0
    very_high = sum(1 for r in responses if r >= 4) / n if n else 0
    very_low = sum(1 for r in responses if r <= 2) / n if n else 0
    return {
        "mean_response": round(mean, 2),
        "pct_agree": round(very_high * 100, 1),       # % of 4s and 5s
        "pct_disagree": round(very_low * 100, 1),     # % of 1s and 2s
        "acquiescence_suspected": very_high > 0.80,   # agrees with >80% of items
        "nay_saying_suspected": very_low > 0.80,
    }


def judge_case(stdout: str, expected: dict) -> dict[str, Any]:
    checks: list[dict] = []

    obj, err = _parse_output(stdout)
    if obj is None:
        checks.append({"check": "json_parse", "pass": False, "reason": err})
        return {"score": 0.0, "matcher_results": checks}
    checks.append({"check": "json_parse", "pass": True, "reason": "ok"})

    responses = obj.get("responses")
    if not isinstance(responses, list):
        checks.append({"check": "responses_list", "pass": False, "reason": "field 'responses' missing or not a list"})
        return {"score": 0.0, "matcher_results": checks}
    checks.append({"check": "responses_list", "pass": True, "reason": "ok"})

    n_expected = expected.get("n_questions", 32)
    if len(responses) != n_expected:
        checks.append({"check": "responses_count", "pass": False,
                       "reason": f"got {len(responses)} responses, expected {n_expected}"})
        return {"score": 0.0, "matcher_results": checks}
    checks.append({"check": "responses_count", "pass": True, "reason": f"{n_expected} ok"})

    # All integers 1..5
    invalid: list[tuple[int, Any]] = []
    coerced: list[int] = []
    for i, r in enumerate(responses):
        if isinstance(r, bool):  # bools are ints in Python — explicitly reject
            invalid.append((i + 1, r))
            continue
        if isinstance(r, int) and 1 <= r <= 5:
            coerced.append(r)
        else:
            invalid.append((i + 1, r))
    if invalid:
        checks.append({"check": "responses_in_range", "pass": False,
                       "reason": f"{len(invalid)} invalid: {invalid[:5]}..."})
        return {"score": 0.0, "matcher_results": checks}
    checks.append({"check": "responses_in_range", "pass": True, "reason": "all 1..5"})

    # All good — derive MBTI
    derived = derive_mbti(coerced, expected["scoring_key"], expected["letters"])
    bias = acquiescence_score(coerced)

    return {
        "score": 1.0,
        "matcher_results": checks,
        "mbti_type": derived["mbti_type"],
        "percentages": derived["percentages"],
        "bias_stats": bias,
        "raw_responses": coerced,
    }


def main() -> None:
    manifest = json.loads(os.environ["TRAPTASK_MANIFEST"])

    stdout = Path(manifest["run"]["stdout"]).read_text()
    exit_code = json.loads(Path(manifest["run"]["meta"]).read_text())["exit_code"]
    expected = json.loads((Path(manifest["expected_dir"]) / "answer.json").read_text())

    usage_record: dict[str, Any] = {}
    usage_path = Path(manifest["outputs_dir"]) / "usage.json"
    if usage_path.exists():
        try:
            raw_usage = json.loads(usage_path.read_text())
        except json.JSONDecodeError:
            raw_usage = {}
        if isinstance(raw_usage, dict):
            usage_record = {k: v for k, v in raw_usage.items() if k in USAGE_FIELDS}

    if exit_code != 0:
        out = {
            "score": 0.0,
            "reason": f"solution exited {exit_code}",
            "agent_answer": stdout.strip()[:300],
            "id": expected.get("id"),
            "category": expected.get("category"),
            "difficulty": expected.get("difficulty"),
            **usage_record,
        }
        print(json.dumps(out))
        return

    metrics = judge_case(stdout, expected)
    metrics["agent_answer"] = stdout.strip()[:300]
    metrics["id"] = expected.get("id")
    metrics["category"] = expected.get("category")
    metrics["difficulty"] = expected.get("difficulty")
    metrics.update(usage_record)
    print(json.dumps(metrics))


if __name__ == "__main__":
    main()

"""Overall grader for the personality/mbti_profile task.

Aggregates per-case judge results into a run-level verdict. Same shape as the
pdf_reader/tenancy_agreement grader: score, n_passed/scored, latency, cost, by_category.
"""
from __future__ import annotations

import json
import os
from collections import Counter

PASS_THRESHOLD = 0.80


def case_cost_usd(case: dict) -> float | None:
    """Per-case spend in USD, or None when nothing measured it.

    Preferred source is trap's own cost proxy, which lands on the case as
    `cost.cost_usd`. It only intercepts Anthropic / OpenAI / Mistral / Moonshot,
    though, and most of this task's models run through OpenRouter — so fall back
    to the `usd_cost` the judge surfaces from the solution's own usage.json.
    """
    cost = case.get("cost")
    if isinstance(cost, dict) and cost.get("cost_usd") is not None:
        return cost["cost_usd"]
    metrics = case.get("metrics")
    if isinstance(metrics, dict):
        return metrics.get("usd_cost")
    return None


def main() -> None:
    cases = json.loads(os.environ["TRAPTASK_MANIFEST"])

    scored = [c for c in cases if c.get("metrics") and c["metrics"].get("score") is not None]
    skipped = [c for c in cases if not c.get("metrics") or c["metrics"].get("score") is None]

    accuracy = sum(c["metrics"]["score"] for c in scored) / len(scored) if scored else 0.0
    n_passed = sum(1 for c in scored if c["metrics"]["score"] == 1.0)

    # By-category breakdown
    by_cat_score: Counter[str] = Counter()
    by_cat_total: Counter[str] = Counter()
    for c in scored:
        cat = c["metrics"].get("category")
        if cat:
            by_cat_total[cat] += 1
            by_cat_score[cat] += c["metrics"]["score"]
    by_category_pct = {
        k: round(by_cat_score[k] / by_cat_total[k], 3) for k in by_cat_total
    }

    # Latency stats from trap-captured per-case duration
    durations = [c.get("duration", 0.0) for c in cases if c.get("duration") is not None]
    if durations:
        ds = sorted(durations)
        latency_ms_median = round(ds[len(ds) // 2] * 1000, 1)
        latency_ms_p95 = round(ds[int(0.95 * len(ds))] * 1000, 1) if len(ds) > 1 else latency_ms_median
        latency_ms_total = round(sum(ds) * 1000, 1)
    else:
        latency_ms_median = latency_ms_p95 = latency_ms_total = 0.0

    # Cost per case, from trap's proxy where it reaches, else the solution's usage.json
    case_costs = [case_cost_usd(c) for c in cases]
    cost_usd_total = (
        round(sum(x for x in case_costs if x is not None), 6)
        if any(x is not None for x in case_costs)
        else None
    )

    passed = bool(scored) and accuracy >= PASS_THRESHOLD

    print(json.dumps({
        "passed": passed,
        "score": round(accuracy, 3),
        "n_passed": n_passed,
        "n_total": len(cases),
        "n_scored": len(scored),
        "n_skipped_no_gold": len(skipped),
        "threshold": PASS_THRESHOLD,
        "by_category": by_category_pct,
        "latency_ms_median": latency_ms_median,
        "latency_ms_p95": latency_ms_p95,
        "latency_ms_total": latency_ms_total,
        "cost_usd_total": cost_usd_total,
    }))


if __name__ == "__main__":
    main()