"""The row contract: canonical accessors and per-lane validation.

A Row is one JSONL object for one measured operation — the durable source of

truth (CONTEXT.md). This module is the single definition of how consumers read

the best-available token/tool/output values from any lane's rows, and of what

every lane's emitter must include. ResultsStore validates at write time, so a

contract violation fails the producing run instead of corrupting a dashboard.

Null means unmeasured (ADR-0002): validation checks presence and types, and

treats null as a legitimate value wherever a signal is probe-dependent.

"""

from __future__ import annotations

import math

from typing import Any, Optional

LANES = ("core", "workflow", "agent", "contention", "mount", "task-loop", "platform", "robustness")

# Required top-level keys per lane. Deliberately the intersection that every

# row of the lane carries — including diagnostic rows (determinism probes,

# skip rows), which is why token fields are not required at this layer.

LANE_REQUIRED_KEYS: dict[str, tuple[str, ...]] = {

    "core": (

        "bench_id",

        "profile",

        "scenario",

        "operation",

        "run",

        "subject",

        "subject_kind",

        "elapsed_ms",

        "returncode",

        "command",

    ),

    "workflow": (

        "bench_id",

        "profile",

        "scenario",

        "workflow",

        "phase",

        "operation",

        "run",

        "subject",

        "subject_kind",

        "elapsed_ms",

        "returncode",

        "process_returncode",

        "command",

    ),

    "agent": (

        "schema_version",

        "bench_id",

        "scenario",

        "operation",

        "run",

        "subject",

        "subject_kind",

        "elapsed_ms",

        "returncode",

        "token_metrics",

        "turn_metrics",

        "tool_call_metrics",

    ),

    "contention": (

        "bench_id",

        "profile",

        "scenario",

        "contention_mode",

        "workers",

        "operation",

        "run",

        "subject",

        "subject_kind",

        "elapsed_ms",

        "returncode",

    ),

    "mount": (

        "bench_id",

        "profile",

        "scenario",

        "operation",

        "run",

        "subject",

        "elapsed_ms",

        "returncode",

    ),

    "task-loop": (

        "bench_id",

        "profile",

        "scenario",

        "operation",

        "run",

        "subject",

        "subject_kind",

        "elapsed_ms",

        "returncode",

        "command",

    ),

    # Platform lane: hosted-workflow rows are only comparable when both sides

    # ran the same semantics class under the same protection variant, so that

    # fairness context is part of every row — including skip rows. Runner

    # identity (ADR-0007) is required because this lane measures network

    # services from a specific vantage point.

    "platform": (

        "bench_id",

        "profile",

        "scenario",

        "operation",

        "run",

        "platform",

        "driver",

        "platform_semantics_version",

        "platform_semantics_class",

        "protection_variant",

        "platform_comparison_key",

        "semantic_contract_match",

        "elapsed_ms",

        "returncode",

        "command",

        "runner_id",

        "runner_class",

    ),

    # Robustness lane (suites/robustness): one row per fault case, judged by

    # independent readers. classification is held | violated | unsupported |

    # harness_error | fault_not_fired; unsupported and fault_not_fired cases are

    # skip rows (returncode 77). See ROBUSTNESS_CLASS_RETURNCODE.

    "robustness": (

        "bench_id",

        "scenario",

        "operation",

        "run",

        "subject",

        "subject_kind",

        "elapsed_ms",

        "returncode",

        "classification",

        "fault_family",

        "fault_point",

    ),

}

NUMERIC_KEYS = ("elapsed_ms",)

INTEGER_KEYS = ("returncode", "run")

SKIP_RETURNCODE = 77

# Optional metrics with constrained types/ranges when present. None is always

# legal (ADR-0002); a present value must be well-formed so consumers never

# need defensive parsing of the durable format.

OPTIONAL_NON_NEGATIVE_INT_KEYS = (

    "peak_rss_bytes",

    "tool_call_count",

    "probe_output_bytes",

    "changed_files_expected",

    "changed_files_recovered",

    "changed_files_reported",

    "changed_files_false_positive",

    "first_payload_bytes_server_to_client",

    "poll_iterations",

    "false_close_count",

    "close_eligible_n",

    "false_vcs_merge_safe_count",

    "vcs_merge_safe_n",

    "false_merge_allowed_count",

    "merge_allowed_n",

    "branches_scored",

    "branches_expected",

)

OPTIONAL_NON_NEGATIVE_NUMBER_KEYS = (

    "ttfd_ms",

    "observed_wait_ms",

    "settle_elapsed_ms",

    "poll_interval_s",

    "poll_quantization_ms",

    "status_post_elapsed_ms",

    "pacing_injected_ms",

)

OPTIONAL_UNIT_INTERVAL_KEYS = ("information_recall", "information_precision", "triage_recall", "triage_precision")

OPTIONAL_STRING_KEYS = (

    "runner_id",

    "runner_class",

    "fixture_id",

    "fixture_generator",

    "fixture_version",

    "fixture_seed",

    "fixture_manifest_sha256",

    "binary_sha256",

    "env_isolation_version",

    "platform_semantics_version",

    "platform_semantics_class",

    "protection_variant",

    "platform_comparison_key",

)

OPTIONAL_OBJECT_KEYS = ("runner_profile",)

class RowContractViolation(SystemExit):

    """Raised after rows are persisted; fails the producing run loudly."""

# Robustness lane: every classification and the returncode it must carry.

ROBUSTNESS_CLASS_RETURNCODE = {

    "held": 0,

    "violated": 1,

    "harness_error": 2,

    "unsupported": SKIP_RETURNCODE,

    "fault_not_fired": SKIP_RETURNCODE,

}

def _robustness_row_errors(index: int, row: dict[str, Any]) -> list[str]:

    op = row.get("operation", "?")

    classification = row.get("classification")

    if classification not in ROBUSTNESS_CLASS_RETURNCODE:

        return [f"row {index} ({op}): unknown robustness classification {classification!r}; "

                f"known: {', '.join(ROBUSTNESS_CLASS_RETURNCODE)}"]

    errors: list[str] = []

    expected = ROBUSTNESS_CLASS_RETURNCODE[classification]

    if row.get("returncode") != expected:

        errors.append(f"row {index} ({op}): classification {classification!r} requires returncode {expected}, "

                      f"got {row.get('returncode')!r}")

    if classification in ("held", "violated") and row.get("fault_fired") is False:

        errors.append(f"row {index} ({op}): a case whose fault did not fire cannot be {classification!r} "

                      "(use fault_not_fired)")

    violations = row.get("violations")

    if classification == "held" and violations:

        errors.append(f"row {index} ({op}): held row lists violations {violations!r}")

    if classification == "violated" and "violations" in row and not violations:

        errors.append(f"row {index} ({op}): violated row lists no violations")

    return errors

def validate_rows(rows: list[dict[str, Any]], lane: str) -> list[str]:

    if lane not in LANE_REQUIRED_KEYS:

        return [f"unknown lane {lane!r}; known: {', '.join(LANES)}"]

    required = LANE_REQUIRED_KEYS[lane]

    errors: list[str] = []

    for index, row in enumerate(rows, 1):

        if not isinstance(row, dict):

            errors.append(f"row {index}: expected object, got {type(row).__name__}")

            continue

        for key in required:

            if key not in row:

                errors.append(f"row {index} ({row.get('operation', '?')}): missing required key {key!r}")

        for key in NUMERIC_KEYS:

            value = row.get(key)

            diagnostic_timing = (lane == "core" and key == "elapsed_ms" and value is None

                and row.get("measurement_phase") == "output_diagnostic"

                and row.get("operation") in {

                    f"{operation}.{probe}" for operation in ("status.clean", "status.dirty", "diff.dirty", "diff.full")

                    for probe in ("inforecall", "determinism")})

            if diagnostic_timing:

                continue

            if key in row and (

                isinstance(value, bool)

                or not isinstance(value, (int, float))

                or not math.isfinite(float(value))

                or float(value) < 0

            ):

                errors.append(f"row {index}: {key} must be numeric, got {type(value).__name__}")

        for key in INTEGER_KEYS:

            value = row.get(key)

            if key in row and (isinstance(value, bool) or not isinstance(value, int)):

                errors.append(f"row {index}: {key} must be an integer, got {type(value).__name__}")

        for key in OPTIONAL_NON_NEGATIVE_INT_KEYS:

            value = row.get(key)

            if key in row and value is not None:

                if isinstance(value, bool) or not isinstance(value, int) or value < 0:

                    errors.append(f"row {index}: {key} must be a non-negative integer or null, got {value!r}")

        for key in OPTIONAL_NON_NEGATIVE_NUMBER_KEYS:

            value = row.get(key)

            if key in row and value is not None:

                if isinstance(value, bool) or not isinstance(value, (int, float)) or not math.isfinite(value) or value < 0:

                    errors.append(f"row {index}: {key} must be a non-negative finite number or null, got {value!r}")

        for key in OPTIONAL_UNIT_INTERVAL_KEYS:

            value = row.get(key)

            if key in row and value is not None:

                if not isinstance(value, (int, float)) or isinstance(value, bool) or not 0.0 <= float(value) <= 1.0:

                    errors.append(f"row {index}: {key} must be in [0, 1] or null, got {value!r}")

        for key in OPTIONAL_STRING_KEYS:

            value = row.get(key)

            if key in row and value is not None and not isinstance(value, str):

                errors.append(f"row {index}: {key} must be a string or null, got {type(value).__name__}")

        for key in OPTIONAL_OBJECT_KEYS:

            value = row.get(key)

            if key in row and value is not None and not isinstance(value, dict):

                errors.append(f"row {index}: {key} must be an object or null, got {type(value).__name__}")

        # Skip-row coherence: a skip is a recorded coverage gap, and consumers

        # key off both signals. Half-marked skips silently become failures

        # (or worse, successes) downstream.

        returncode = row.get("returncode")

        if returncode == SKIP_RETURNCODE:

            if not row.get("skipped"):

                errors.append(f"row {index} ({row.get('operation', '?')}): returncode 77 requires skipped: true")

            if not str(row.get("skip_reason") or "").strip():

                errors.append(f"row {index} ({row.get('operation', '?')}): skip rows require a non-empty skip_reason")

        elif row.get("skipped"):

            errors.append(

                f"row {index} ({row.get('operation', '?')}): skipped: true requires returncode {SKIP_RETURNCODE}"

            )

        if lane == "robustness":

            errors.extend(_robustness_row_errors(index, row))

    return errors

def number_or_none(value: Any) -> Optional[float]:

    if value is None or isinstance(value, bool):

        return None

    try:

        parsed = float(value)

    except (TypeError, ValueError):

        return None

    if not math.isfinite(parsed):

        return None

    return parsed

def row_returncode(row: dict[str, Any]) -> int:

    """Row returncode with missing/null treated as failure, never a crash."""

    value = row.get("returncode")

    if value is None:

        return 1

    try:

        return int(value)

    except (TypeError, ValueError):

        return 1

def transcript_metrics_complete(row: dict[str, Any]) -> bool:

    source = row.get("source")

    execution = source.get("agent_execution") if isinstance(source, dict) else None

    is_agent = isinstance(row.get("agent"), dict) or str(row.get("operation", "")).startswith("agent.") or isinstance(execution, dict)

    return not ((is_agent and row.get("output_truncated") is True) or (isinstance(execution, dict) and

        (execution.get("usage_complete") is False or execution.get("transcript_metrics_complete") is False)))

def token_total(row: dict[str, Any]) -> Optional[float]:

    """Best-available token total for a row, preferring provider-reported usage."""

    if not transcript_metrics_complete(row):

        return None

    token_metrics = row.get("token_metrics")

    if isinstance(token_metrics, dict):

        value = number_or_none(token_metrics.get("total_tokens_reported"))

        if value is not None:

            return value

        reported_parts = [

            number_or_none(token_metrics.get("input_tokens_reported")),

            number_or_none(token_metrics.get("output_tokens_reported")),

            number_or_none(token_metrics.get("reasoning_tokens_reported")),

        ]

        if any(part is not None for part in reported_parts):

            return sum(part or 0.0 for part in reported_parts)

    for key in ("estimated_tokens_total", "token_count", "total_tokens"):

        value = number_or_none(row.get(key))

        if value is not None:

            return value

    token_estimate = row.get("token_estimate")

    if isinstance(token_estimate, dict):

        value = number_or_none(token_estimate.get("total"))

        if value is not None:

            return value

    tokens = row.get("tokens")

    if isinstance(tokens, dict):

        for key in ("total", "total_tokens", "provider_total", "estimated_total"):

            value = number_or_none(tokens.get(key))

            if value is not None:

                return value

        provider = tokens.get("provider_reported")

        if isinstance(provider, dict):

            value = number_or_none(provider.get("total"))

            if value is not None:

                return value

    if isinstance(token_metrics, dict):

        chars = number_or_none(token_metrics.get("transcript_chars"))

        if chars is None:

            input_chars = number_or_none(token_metrics.get("normalized_input_chars"))

            output_chars = number_or_none(token_metrics.get("normalized_output_chars"))

            if input_chars is not None or output_chars is not None:

                chars = (input_chars or 0.0) + (output_chars or 0.0)

        if chars is not None:

            return max(1.0, math.ceil(chars / 4.0))

    return None

def tool_calls(row: dict[str, Any]) -> Optional[float]:

    if not transcript_metrics_complete(row):

        return None

    for key in ("tool_call_count", "total_tool_calls"):

        value = number_or_none(row.get(key))

        if value is not None:

            return value

    nested = row.get("tool_calls")

    if isinstance(nested, dict):

        for key in ("total", "all"):

            value = number_or_none(nested.get(key))

            if value is not None:

                return value

    tool_call_metrics = row.get("tool_call_metrics")

    if isinstance(tool_call_metrics, dict):

        value = number_or_none(tool_call_metrics.get("tool_calls_total"))

        if value is not None:

            return value

    return None

def output_bytes(row: dict[str, Any]) -> Optional[float]:

    value = number_or_none(row.get("raw_output_bytes"))

    if value is not None:

        return value

    stdout = number_or_none(row.get("stdout_bytes"))

    stderr = number_or_none(row.get("stderr_bytes"))

    if stdout is not None or stderr is not None:

        return (stdout or 0.0) + (stderr or 0.0)

    token_estimate = row.get("token_estimate")

    if isinstance(token_estimate, dict):

        stdout = number_or_none(token_estimate.get("stdout_bytes"))

        stderr = number_or_none(token_estimate.get("stderr_bytes"))

        if stdout is not None or stderr is not None:

            return (stdout or 0.0) + (stderr or 0.0)

    return None

def output_truncated(row: dict[str, Any]) -> bool:

    if bool(row.get("output_truncated")):

        return True

    if bool(row.get("stdout_truncated")) or bool(row.get("stderr_truncated")):

        return True

    token_estimate = row.get("token_estimate")

    if isinstance(token_estimate, dict):

        return bool(token_estimate.get("stdout_truncated")) or bool(token_estimate.get("stderr_truncated"))

    return False