Blame for src/nviq/cli.py

import argparse
import json
import os
import sys
from pathlib import Path

from .adapters.openai_realtime import (
    adapt_openai_realtime_events,
    adapt_openai_realtime_trace_manifest,
    load_realtime_events,
)
from .adapters.gemini_live import adapt_gemini_live_trace_manifest
from .adapters.realtime_venus import adapt_realtime_venus_trace_manifest
from .audit.adapters import adapt_precomputed_observations
from .audit.compare import compare_audits
from .audit.manifest import load_manifest, validate_manifest
from .audit.remediations import load_remediations
from .audit.runner import run_audit, write_audit_artifacts
from .cases import load_cases, load_schema, validate_cases
from .errors import EvidenceValidationError
from .execution.evidence import load_evidence, validate_evidence
from .execution.runner import execute_audit
from .execution.target import capture_http_target
from .observations import load_observations
from . import __version__
from .errors import NVIQError
from .reporting import write_report
from .resources import (
    AUDIT_MANIFEST_SCHEMA,
    CAUSAL_LEDGER_SCHEMA,
    EVIDENCE_BUNDLE_SCHEMA,
    REMEDIATION_REGISTRY,
    TEST_CASE_SCHEMA,
)
from .realtime_venus_qualification import qualify_realtime_venus_runtime
from .runner import run_benchmark


# Process exit codes (documented in README).
EXIT_OK = 0
EXIT_ERROR = 1  # invalid input or NVIQ error; one-line message on stderr
EXIT_USAGE_OR_NOT_QUALIFIED = 2  # argparse usage error, or `qualify` receipt not qualified
EXIT_GATE_FAILED = 3  # --fail-on threshold reached
EXIT_CAPTURE_INCOMPLETE = 4  # `target run` could not capture every case

_SEVERITY_RANK = {"critical": 0, "high": 1, "medium": 2, "low": 3}
_FAIL_ON_CHOICES = ["critical", "high", "medium", "low", "none"]

_TRACKS = ["transcript_only", "transcript_metadata", "audio_native", "lsr_cvf", "oracle"]


def _parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="nviq", description="Noct-Tech NVIQ benchmark and execution engine")
    parser.add_argument("--version", action="version", version=f"nviq {__version__}")
    sub = parser.add_subparsers(dest="command", required=True)

    validate = sub.add_parser("validate", help="validate NVIQ test cases")
    validate.add_argument("--schema", type=Path, default=TEST_CASE_SCHEMA, help="test-case schema (default: bundled v0.2)")
    validate.add_argument("--cases", type=Path, required=True)

    score = sub.add_parser("score", help="score track observations")
    score.add_argument("--schema", type=Path, default=TEST_CASE_SCHEMA, help="test-case schema (default: bundled v0.2)")
    score.add_argument("--cases", type=Path, required=True)
    score.add_argument("--observations", type=Path, required=True)
    score.add_argument("--output", type=Path, required=True)

    audit = sub.add_parser("audit", help="run NVIQ commercial Reliability Audits")
    audit_sub = audit.add_subparsers(dest="audit_command", required=True)

    audit_validate = audit_sub.add_parser("validate-manifest", help="validate a commercial audit manifest")
    audit_validate.add_argument("--manifest", type=Path, required=True)
    audit_validate.add_argument("--manifest-schema", type=Path, default=AUDIT_MANIFEST_SCHEMA)

    audit_run = audit_sub.add_parser("run", help="run a commercial NVIQ Reliability Audit from precomputed observations")
    audit_run.add_argument("--manifest", type=Path, required=True)
    audit_run.add_argument("--manifest-schema", type=Path, default=AUDIT_MANIFEST_SCHEMA)
    audit_run.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    audit_run.add_argument("--observations", type=Path, required=True)
    audit_run.add_argument("--remediations", type=Path, default=REMEDIATION_REGISTRY)
    audit_run.add_argument("--output-dir", type=Path, required=True)
    audit_run.add_argument(
        "--fail-on",
        choices=_FAIL_ON_CHOICES,
        default="none",
        help="exit 3 when any finding is at or above this severity (default: none)",
    )

    audit_compare = audit_sub.add_parser("compare", help="compare two canonical NVIQ audit reports")
    audit_compare.add_argument("--baseline", type=Path, required=True)
    audit_compare.add_argument("--candidate", type=Path, required=True)
    audit_compare.add_argument("--output-dir", type=Path, required=True)

    evidence = sub.add_parser("evidence", help="validate raw NVIQ execution evidence")
    evidence_sub = evidence.add_subparsers(dest="evidence_command", required=True)
    evidence_validate = evidence_sub.add_parser("validate", help="validate an Evidence Bundle v0.1")
    evidence_validate.add_argument("--evidence", type=Path, required=True)
    evidence_validate.add_argument("--schema", type=Path, default=EVIDENCE_BUNDLE_SCHEMA)
    evidence_validate.add_argument("--cases", type=Path)
    evidence_validate.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    evidence_validate.add_argument("--causal-ledger-schema", type=Path, default=CAUSAL_LEDGER_SCHEMA)

    ingest = sub.add_parser("ingest", help="convert provider traces into NVIQ Evidence Bundle records")
    ingest_sub = ingest.add_subparsers(dest="ingest_command", required=True)
    openai_realtime = ingest_sub.add_parser("openai-realtime", help="ingest captured OpenAI Realtime JSON/JSONL traces")
    openai_realtime.add_argument("--cases", type=Path, required=True)
    openai_realtime.add_argument("--case-id")
    openai_realtime.add_argument("--events", type=Path)
    openai_realtime.add_argument("--trace-manifest", type=Path)
    openai_realtime.add_argument("--track", choices=_TRACKS)
    openai_realtime.add_argument("--output", type=Path, required=True)
    openai_realtime.add_argument("--model")
    openai_realtime.add_argument("--session-id")
    openai_realtime.add_argument("--audio-ref")
    openai_realtime.add_argument(
        "--capture-origin",
        help='required with --events: "runtime" for live captures, "adapter_contract_fixture" for hand-built traces',
    )
    openai_realtime.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    openai_realtime.add_argument("--evidence-schema", type=Path, default=EVIDENCE_BUNDLE_SCHEMA)

    gemini_live = ingest_sub.add_parser("gemini-live", help="ingest instrumented Gemini Live capture manifests")
    gemini_live.add_argument("--cases", type=Path, required=True)
    gemini_live.add_argument("--trace-manifest", type=Path, required=True)
    gemini_live.add_argument("--output", type=Path, required=True)
    gemini_live.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    gemini_live.add_argument("--evidence-schema", type=Path, default=EVIDENCE_BUNDLE_SCHEMA)

    realtime_venus = ingest_sub.add_parser("realtime-venus", help="ingest instrumented Realtime-Venus capture manifests")
    realtime_venus.add_argument("--cases", type=Path, required=True)
    realtime_venus.add_argument("--trace-manifest", type=Path, required=True)
    realtime_venus.add_argument("--output", type=Path, required=True)
    realtime_venus.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    realtime_venus.add_argument("--evidence-schema", type=Path, default=EVIDENCE_BUNDLE_SCHEMA)
    realtime_venus.add_argument("--causal-ledger-schema", type=Path, default=CAUSAL_LEDGER_SCHEMA)

    execute = sub.add_parser("execute", help="derive NVIQ observations from raw evidence and run a Reliability Audit")
    execute.add_argument("--manifest", type=Path, required=True)
    execute.add_argument("--evidence", type=Path, required=True)
    execute.add_argument("--output-dir", type=Path, required=True)
    execute.add_argument(
        "--fail-on",
        choices=_FAIL_ON_CHOICES,
        default="none",
        help="exit 3 when any finding is at or above this severity (default: none)",
    )
    execute.add_argument("--manifest-schema", type=Path, default=AUDIT_MANIFEST_SCHEMA)
    execute.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    execute.add_argument("--evidence-schema", type=Path, default=EVIDENCE_BUNDLE_SCHEMA)
    execute.add_argument("--causal-ledger-schema", type=Path, default=CAUSAL_LEDGER_SCHEMA)
    execute.add_argument("--remediations", type=Path, default=REMEDIATION_REGISTRY)

    target = sub.add_parser("target", help="execute benchmark cases against a live target adapter")
    target_sub = target.add_subparsers(dest="target_command", required=True)
    target_run = target_sub.add_parser("run", help="capture raw evidence from an HTTP JSON target and run NVIQ")
    target_run.add_argument("--manifest", type=Path, required=True)
    target_run.add_argument("--endpoint", required=True)
    target_run.add_argument("--output-dir", type=Path, required=True)
    target_run.add_argument(
        "--fail-on",
        choices=_FAIL_ON_CHOICES,
        default="none",
        help="exit 3 when any finding is at or above this severity (default: none)",
    )
    target_run.add_argument("--track")
    target_run.add_argument("--token-env", default="NVIQ_TARGET_BEARER_TOKEN")
    target_run.add_argument("--timeout", type=float, default=30.0)
    target_run.add_argument("--retries", type=int, default=0, help="retries per case on transport failure (default 0)")
    target_run.add_argument("--retry-delay", type=float, default=1.0, help="seconds between retries (default 1.0)")
    target_run.add_argument("--manifest-schema", type=Path, default=AUDIT_MANIFEST_SCHEMA)
    target_run.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    target_run.add_argument("--evidence-schema", type=Path, default=EVIDENCE_BUNDLE_SCHEMA)
    target_run.add_argument("--causal-ledger-schema", type=Path, default=CAUSAL_LEDGER_SCHEMA)
    target_run.add_argument("--remediations", type=Path, default=REMEDIATION_REGISTRY)

    qualify = sub.add_parser("qualify", help="issue deterministic frontier qualification receipts")
    qualify_sub = qualify.add_subparsers(dest="qualify_command", required=True)
    venus_qualify = qualify_sub.add_parser(
        "realtime-venus",
        help="qualify authentic Realtime-Venus runtime evidence",
    )
    venus_qualify.add_argument("--cases", type=Path, required=True)
    venus_qualify.add_argument("--evidence", type=Path, required=True)
    venus_qualify.add_argument("--execution-dir", type=Path, required=True)
    venus_qualify.add_argument("--output", type=Path, required=True)
    venus_qualify.add_argument("--case-schema", type=Path, default=TEST_CASE_SCHEMA)
    venus_qualify.add_argument("--evidence-schema", type=Path, default=EVIDENCE_BUNDLE_SCHEMA)
    venus_qualify.add_argument("--causal-ledger-schema", type=Path, default=CAUSAL_LEDGER_SCHEMA)
    return parser


def _read_json(path: Path) -> dict:
    data = json.loads(path.read_text(encoding="utf-8"))
    if not isinstance(data, dict):
        raise ValueError(f"expected JSON object: {path}")
    return data


def _resolve_case_set(manifest_path: Path, case_set: str) -> Path:
    """Resolve a manifest case_set as given, then relative to the manifest and its parent folders."""
    supplied = Path(case_set)
    if supplied.is_absolute() or supplied.exists():
        return supplied
    for base in (manifest_path.parent, *manifest_path.resolve().parent.parents):
        candidate = base / supplied
        if candidate.exists():
            return candidate
    return supplied


def _gate_exit(args: argparse.Namespace, report: dict) -> int:
    threshold = getattr(args, "fail_on", "none")
    if threshold == "none":
        return EXIT_OK
    limit = _SEVERITY_RANK[threshold]
    hits = [item for item in report.get("findings", []) if _SEVERITY_RANK[item["severity"].lower()] <= limit]
    if hits:
        print(f"nviq: gate failed: {len(hits)} finding(s) at or above {threshold}", file=sys.stderr)
        return EXIT_GATE_FAILED
    return EXIT_OK


def _load_manifest_and_cases(
    manifest_path: Path,
    manifest_schema_path: Path,
    case_schema_path: Path,
) -> tuple[dict, list[dict]]:
    manifest = load_manifest(manifest_path)
    validate_manifest(manifest, load_schema(manifest_schema_path))
    cases = load_cases(_resolve_case_set(manifest_path, manifest["case_set"]))
    validate_cases(cases, load_schema(case_schema_path))
    return manifest, cases


def _run_audit_command(args: argparse.Namespace) -> int:
    if args.audit_command == "validate-manifest":
        manifest = load_manifest(args.manifest)
        validate_manifest(manifest, load_schema(args.manifest_schema))
        print(f"NVIQ Commercial Audit v0.1: validated audit manifest {manifest['audit_id']}")
        return 0

    if args.audit_command == "compare":
        comparison = compare_audits(_read_json(args.baseline), _read_json(args.candidate))
        args.output_dir.mkdir(parents=True, exist_ok=True)
        output = args.output_dir / "audit-comparison.json"
        write_report(comparison, output)
        print(f"NVIQ Reliability Audit comparison: {comparison['recommendation']}; report={output}")
        return 0

    manifest, cases = _load_manifest_and_cases(args.manifest, args.manifest_schema, args.case_schema)
    observations = load_observations(args.observations, {case["case_id"] for case in cases})
    observations = adapt_precomputed_observations(
        observations,
        provider="precomputed",
        model=manifest["target_name"],
        configuration={"target_version": manifest["target_version"]},
    )
    registry = load_remediations(args.remediations)
    report = run_audit(manifest, cases, observations, registry)
    paths = write_audit_artifacts(report, args.output_dir)
    print(
        f"NVIQ Reliability Audit v0.1: status={report['summary']['overall_status']}; "
        f"findings={len(report['findings'])}; report={paths['json']}"
    )
    return _gate_exit(args, report)


def _run_evidence_command(args: argparse.Namespace) -> int:
    records = load_evidence(args.evidence)
    known_case_ids = None
    if args.cases:
        cases = load_cases(args.cases)
        validate_cases(cases, load_schema(args.case_schema))
        known_case_ids = {case["case_id"] for case in cases}
    validate_evidence(
        records,
        load_schema(args.schema),
        known_case_ids,
        load_schema(args.causal_ledger_schema),
    )
    print(f"NVIQ Execution Engine v0.1: validated {len(records)} evidence records")
    return 0


def _run_ingest_command(args: argparse.Namespace) -> int:
    cases = load_cases(args.cases)
    validate_cases(cases, load_schema(args.case_schema))
    known_case_ids = {case["case_id"] for case in cases}

    if args.ingest_command == "openai-realtime":
        if args.trace_manifest:
            if args.case_id or args.events:
                raise ValueError("--trace-manifest cannot be combined with --case-id or --events")
            evidence_records = adapt_openai_realtime_trace_manifest(cases, args.trace_manifest)
        else:
            if not args.case_id or not args.events:
                raise ValueError("single-trace ingestion requires --case-id and --events")
            case = next((item for item in cases if item["case_id"] == args.case_id), None)
            if case is None:
                raise ValueError(f"unknown case_id: {args.case_id}")
            events = load_realtime_events(args.events)
            evidence_records = [adapt_openai_realtime_events(
                case,
                events,
                track=args.track or "audio_native",
                model=args.model,
                session_id=args.session_id,
                audio_ref=args.audio_ref,
                capture_origin=args.capture_origin,
            )]
        causal_schema = None
        label = "OpenAI Realtime adapter v0.1"
    elif args.ingest_command == "gemini-live":
        evidence_records = adapt_gemini_live_trace_manifest(cases, args.trace_manifest)
        causal_schema = None
        label = "Gemini Live adapter v0.1"
    elif args.ingest_command == "realtime-venus":
        evidence_records = adapt_realtime_venus_trace_manifest(cases, args.trace_manifest)
        causal_schema = load_schema(args.causal_ledger_schema)
        label = "Realtime-Venus adapter v0.1"
    else:
        raise ValueError(f"unsupported ingest command: {args.ingest_command}")

    validate_evidence(
        evidence_records,
        load_schema(args.evidence_schema),
        known_case_ids,
        causal_schema,
    )
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(evidence_records, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    print(f"NVIQ {label}: wrote {len(evidence_records)} evidence record(s); output={args.output}")
    return 0


def _run_execute_command(args: argparse.Namespace) -> int:
    manifest, cases = _load_manifest_and_cases(args.manifest, args.manifest_schema, args.case_schema)
    evidence = load_evidence(args.evidence)
    validate_evidence(
        evidence,
        load_schema(args.evidence_schema),
        {case["case_id"] for case in cases},
        load_schema(args.causal_ledger_schema),
    )
    registry = load_remediations(args.remediations)
    result = execute_audit(manifest, cases, evidence, registry, args.output_dir)
    report = result["report"]
    print(
        f"NVIQ Execution Engine v0.1: derived={len(result['observations'])}; "
        f"status={report['summary']['overall_status']}; report={result['paths']['json']}"
    )
    return _gate_exit(args, report)


def _run_target_command(args: argparse.Namespace) -> int:
    manifest, cases = _load_manifest_and_cases(args.manifest, args.manifest_schema, args.case_schema)
    track = args.track
    if track is None:
        if len(manifest["tracks"]) != 1:
            raise ValueError("--track is required when the manifest contains multiple tracks")
        track = manifest["tracks"][0]
    if track not in manifest["tracks"]:
        raise ValueError(f"target track {track} is not enabled by the manifest")

    token = os.environ.get(args.token_env) if args.token_env else None
    capture = capture_http_target(
        cases,
        args.endpoint,
        track,
        bearer_token=token,
        timeout_s=args.timeout,
        retries=args.retries,
        retry_delay_s=args.retry_delay,
    )
    evidence_schema = load_schema(args.evidence_schema)
    ledger_schema = load_schema(args.causal_ledger_schema)
    known_case_ids = {case["case_id"] for case in cases}
    evidence: list[dict] = []
    errors = list(capture["errors"])
    for row in capture["evidence"]:
        try:
            validate_evidence([row], evidence_schema, known_case_ids, ledger_schema)
        except EvidenceValidationError as exc:
            errors.append({"case_id": row.get("case_id"), "track": track, "attempts": None, "error": f"invalid evidence: {exc}"})
            continue
        evidence.append(row)

    args.output_dir.mkdir(parents=True, exist_ok=True)
    captured_path = args.output_dir / "captured-evidence.json"
    errors_path = args.output_dir / "capture-errors.json"
    captured_path.write_text(json.dumps(evidence, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    errors_path.write_text(json.dumps(sorted(errors, key=lambda item: str(item["case_id"])), indent=2, sort_keys=True) + "\n", encoding="utf-8")

    registry = load_remediations(args.remediations)
    result = execute_audit(manifest, cases, evidence, registry, args.output_dir)
    print(
        f"NVIQ target execution v0.1: captured={len(evidence)}; capture_errors={len(errors)}; derived={len(result['observations'])}; "
        f"status={result['report']['summary']['overall_status']}; evidence={captured_path}; report={result['paths']['json']}"
    )
    if errors:
        print(f"nviq: warning: {len(errors)} case(s) were not captured; see {errors_path}", file=sys.stderr)
        return EXIT_CAPTURE_INCOMPLETE
    return _gate_exit(args, result["report"])


def _run_qualify_command(args: argparse.Namespace) -> int:
    if args.qualify_command != "realtime-venus":
        raise ValueError(f"unsupported qualification command: {args.qualify_command}")

    cases = load_cases(args.cases)
    validate_cases(cases, load_schema(args.case_schema))
    known_case_ids = {case["case_id"] for case in cases}

    evidence = load_evidence(args.evidence)
    validate_evidence(
        evidence,
        load_schema(args.evidence_schema),
        known_case_ids,
        load_schema(args.causal_ledger_schema),
    )

    observations = load_observations(
        args.execution_dir / "derived-observations.json",
        known_case_ids,
    )
    audit_report = _read_json(args.execution_dir / "audit-report.json")
    receipt = qualify_realtime_venus_runtime(
        cases,
        evidence,
        observations,
        audit_report,
    )
    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(
        json.dumps(receipt, indent=2, sort_keys=True) + "\n",
        encoding="utf-8",
    )
    print(
        f"NVIQ Realtime-Venus runtime qualification: status={receipt['status']}; "
        f"receipt={args.output}"
    )
    return 0 if receipt["qualified"] else 2


def main(argv: list[str] | None = None) -> int:
    args = _parser().parse_args(argv)
    if args.command == "audit":
        return _run_audit_command(args)
    if args.command == "evidence":
        return _run_evidence_command(args)
    if args.command == "ingest":
        return _run_ingest_command(args)
    if args.command == "execute":
        return _run_execute_command(args)
    if args.command == "target":
        return _run_target_command(args)
    if args.command == "qualify":
        return _run_qualify_command(args)

    schema = load_schema(args.schema)
    cases = load_cases(args.cases)
    validate_cases(cases, schema)
    if args.command == "validate":
        print(f"NVIQ v0.2: validated {len(cases)} cases")
        return 0

    known_case_ids = {case["case_id"] for case in cases}
    observations = load_observations(args.observations, known_case_ids)
    report = run_benchmark(cases, observations)
    write_report(report, args.output)
    hard_gate_failures = sum(track["hard_gate_failures"] for track in report["tracks"].values())
    catastrophic_failures = sum(track["catastrophic_failures"] for track in report["tracks"].values())
    print(
        f"NVIQ v0.2: scored {len(observations)} observations; "
        f"hard-gate failures={hard_gate_failures}; catastrophic failures={catastrophic_failures}; "
        f"report={args.output}"
    )
    return 0


def run(argv: list[str] | None = None) -> int:
    """Console entry point: like main(), but prints one-line errors instead of tracebacks."""
    try:
        return main(argv)
    except KeyboardInterrupt:
        print("nviq: interrupted", file=sys.stderr)
        return 130
    except (NVIQError, ValueError, RuntimeError, OSError, KeyError) as exc:
        detail = str(exc) or exc.__class__.__name__
        if isinstance(exc, KeyError):
            detail = f"missing required field {detail}"
        elif isinstance(exc, json.JSONDecodeError):
            detail = f"invalid JSON input: {detail}"
        print(f"nviq: error: {detail}", file=sys.stderr)
        return EXIT_ERROR


if __name__ == "__main__":
    raise SystemExit(run())