#!/usr/bin/env python3
"""BEIS conformance harness. Point it at ANY system, including ours.

MIT Licence. Copyright (c) 2026 BvLogic Solutions LLC. Permission is hereby
granted, free of charge, to any person obtaining a copy of this software to
use, copy, modify, merge, publish, distribute, sublicense and/or sell copies,
subject to the copyright notice and this permission notice being included.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND. For a
measurement tool that last clause is the honest position rather than
boilerplate: it reports what your system returned and cannot know whether
that was correct.

The specification it implements is separate: BEIS 0.1 is CC BY 4.0.
https://bvlogic.com/standard/licence/

WHY THIS FILE EXISTS, stated plainly because it is a correction.

The benchmark page says we publish the method and offer the harness so a third
party can run it themselves, on the grounds that a result the reader produces is
worth more than one we publish. That was true of the METHOD and not of the code:
tools/eip_benchmark.py hardcodes bvlogic.com and reads our own golden set, so
nobody outside this company could actually run it. The offer was real and
unexecutable, which is a gap between a published claim and the software behind
it, and the whole site exists to argue against exactly that.

This is the executable version. One file, standard library only, no BvLogic
content, no network calls to us. Point it at your own endpoint with your own
questions and it will score your system, including scoring it badly.

    python beis_conformance.py --endpoint https://your.system/ask \\
                               --questions your-questions.json

QUESTION FILE FORMAT
    {
      "questions": [
        { "q": "When do the EU AI Act high-risk rules apply?",
          "expect_source": "/regulation/eu-ai-act/",     # optional
          "should_refuse": false }                        # optional
      ]
    }

    A question with should_refuse true is one your system SHOULD decline, because
    the answer is not in its corpus. Include some. A benchmark with no refusal
    cases measures confidence, not accuracy.

WHAT THIS CAN AND CANNOT MEASURE FROM OUTSIDE
    Measurable  retrieval@1, retrieval@3, citation coverage, refusal accuracy,
                response speed
    NOT         permission correctness, freshness, action safety, human
                oversight. These depend on your access model, your data pipeline
                and your approval flow, none of which are visible over HTTP.

    Those four are reported as NOT MEASURED rather than skipped or scored zero.
    A conformance report that silently omits what it could not test is the
    failure BEIS was written to prevent, and it would be an odd thing for this
    file to do.
"""
from __future__ import annotations

import argparse
import json
import ssl
import sys
import time
import urllib.error
import urllib.request

MEASURABLE = ("retrieval-1", "retrieval-3", "citation", "refusal", "speed")
NOT_MEASURABLE = {
    "permission": "depends on your access model, which is not visible over HTTP",
    "freshness": "depends on your data pipeline, not on any answer it returns",
    "action-safety": "requires a system that takes actions; this harness only asks",
    "oversight": "requires observing your approval flow, not your answers",
    "reasoning": "cross-document reasoning cannot be judged from a single response",
    "hallucination": "requires ground truth we do not have for your corpus",
}


def ask(endpoint, question, timeout, insecure):
    """POST {"q": ...}. Returns (payload, milliseconds, error)."""
    body = json.dumps({"q": question}).encode("utf-8")
    req = urllib.request.Request(endpoint, body,
                                 headers={"Content-Type": "application/json",
                                          "User-Agent": "BEIS-conformance/0.1"})
    ctx = None
    if insecure:
        ctx = ssl.create_default_context()
        ctx.check_hostname = False
        ctx.verify_mode = ssl.CERT_NONE
    t0 = time.time()
    try:
        with urllib.request.urlopen(req, timeout=timeout, context=ctx) as r:
            payload = json.loads(r.read().decode("utf-8", "replace"))
    except urllib.error.HTTPError as e:
        return None, int((time.time() - t0) * 1000), "HTTP %s" % e.code
    except Exception as e:
        return None, int((time.time() - t0) * 1000), str(e)
    return payload, int((time.time() - t0) * 1000), None


def sources_of(payload):
    """Pull a list of source URLs out of a response, tolerantly.

    Deliberately permissive about SHAPE and strict about substance: a system
    should not have to adopt our JSON to be measured. If no sources can be
    found at all, that is itself the finding - an answer with no citation is
    unverifiable, and BEIS scores citation coverage for that reason.
    """
    if not isinstance(payload, dict):
        return []
    for key in ("sources", "citations", "refs", "documents", "results"):
        v = payload.get(key)
        if isinstance(v, list):
            out = []
            for item in v:
                if isinstance(item, str):
                    out.append(item)
                elif isinstance(item, dict):
                    for k in ("url", "href", "path", "source", "id"):
                        if isinstance(item.get(k), str):
                            out.append(item[k]); break
            return out
    return []


def refused(payload):
    if not isinstance(payload, dict):
        return False
    for k in ("refused", "declined", "no_answer"):
        if payload.get(k) is True:
            return True
    conf = str(payload.get("confidence", "")).lower()
    if conf in ("none", "no"):
        return True
    text = str(payload.get("answer") or payload.get("text") or "")
    return bool(text) and any(p in text.lower() for p in (
        "i do not have", "not published", "no information", "cannot answer",
        "we have not published"))


def run(endpoint, questions, timeout, insecure):
    at1 = at3 = cited = 0
    refusal_right = refusal_total = 0
    graded = 0
    lat, errors = [], []

    for case in questions:
        q = case["q"]
        payload, ms, err = ask(endpoint, q, timeout, insecure)
        if err:
            errors.append((q, err))
            continue
        lat.append(ms)
        srcs = sources_of(payload)
        did_refuse = refused(payload)

        if case.get("should_refuse"):
            refusal_total += 1
            if did_refuse:
                refusal_right += 1
            continue

        if srcs:
            cited += 1
        want = case.get("expect_source")
        if want:
            graded += 1
            norm = [s.rstrip("/") for s in srcs]
            w = want.rstrip("/")
            if norm and norm[0].endswith(w):
                at1 += 1
            if any(s.endswith(w) for s in norm[:3]):
                at3 += 1

    answered = len([c for c in questions if not c.get("should_refuse")])
    lat.sort()
    return {
        "questions": len(questions),
        "errors": len(errors),
        "error_detail": errors[:5],
        "graded": graded,
        "retrieval-1": round(100 * at1 / graded, 1) if graded else None,
        "retrieval-3": round(100 * at3 / graded, 1) if graded else None,
        "citation": round(100 * cited / answered, 1) if answered else None,
        "refusal": round(100 * refusal_right / refusal_total, 1) if refusal_total else None,
        "refusal_cases": refusal_total,
        "p50_ms": lat[len(lat) // 2] if lat else None,
        "p95_ms": lat[int(len(lat) * 0.95)] if lat else None,
    }


def validate_statement(stmt):
    """Check a conformance statement somebody handed you. Returns problems.

    This is the half that makes the format worth anything. Emitting a statement
    is easy and self-serving; being able to check one you were GIVEN is what
    stops a vendor publishing the three metrics that flattered them and being
    truthful in every individual number while the overall impression is false.

    A statement missing a metric is INVALID, not partial. There is no partial.
    """
    problems = []
    for f in ("system", "date", "beis_version", "question_set", "measured",
              "not_measured", "run_by"):
        if not stmt.get(f):
            problems.append("missing required field: %s" % f)

    measured = stmt.get("measured") or {}
    absent = [m for m in MEASURABLE if m not in measured]
    if absent:
        problems.append("INVALID: metrics missing from `measured`: %s. A statement is not "
                        "partial, it is invalid - this is where a flattering one omits its "
                        "worst number." % ", ".join(absent))

    nm = stmt.get("not_measured") or {}
    for k, why in nm.items():
        if not why or len(str(why)) < 10:
            problems.append("`not_measured.%s` gives no reason. Not-measured without a why "
                            "is indistinguishable from measured badly." % k)

    qs = stmt.get("question_set") or {}
    src = str(qs.get("source", "")).lower()
    if src in ("", "unstated", "internal", "standard"):
        problems.append("question_set.source is %r. If a reader cannot tell where the "
                        "questions came from, they cannot weigh the result." % qs.get("source"))
    if qs.get("refusal_cases", 0) == 0:
        problems.append("no refusal cases: the accuracy figures measure confidence rather "
                        "than accuracy, and the statement must say so")

    if str(stmt.get("run_by", "")).lower() in ("", "unstated"):
        problems.append("run_by is unstated. A self-run result is not worthless; concealing "
                        "that it was self-run is.")

    for key in stmt:
        if key.lower() in ("overall", "overall_score", "total", "score"):
            problems.append("INVALID: an aggregate %r is present. BEIS has no overall score, "
                            "and inventing one is how a bad metric gets averaged away." % key)

    d = str(stmt.get("date", ""))
    if d:
        import datetime
        try:
            age = (datetime.date.today() - datetime.date.fromisoformat(d)).days
            if age > 365:
                problems.append("the run is %d days old. Re-run it before presenting it as "
                                "current." % age)
        except ValueError:
            problems.append("date %r is not ISO format" % d)
    return problems


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--endpoint")
    ap.add_argument("--questions")
    ap.add_argument("--validate", metavar="FILE",
                    help="check a conformance statement you were GIVEN")
    ap.add_argument("--timeout", type=float, default=30)
    ap.add_argument("--insecure", action="store_true",
                    help="skip TLS verification. Use only when a local proxy intercepts "
                         "HTTPS; it is a property of your machine, not of the system tested.")
    ap.add_argument("--json", action="store_true")
    ap.add_argument("--statement", metavar="FILE",
                    help="write a BEIS conformance statement. It carries EVERY metric "
                         "measured and every metric that could not be, because a partial "
                         "statement is invalid rather than partial.")
    ap.add_argument("--system", help="what was measured, for the statement")
    ap.add_argument("--run-by", default="unstated",
                    help="vendor, customer, or third party. Saying it is self-run does not "
                         "invalidate a result; concealing it does.")
    ap.add_argument("--question-source", default="unstated",
                    help="where the questions came from. A vendor-written set on a vendor's "
                         "own corpus is close to a self-graded exam and a reader must be able "
                         "to tell.")
    args = ap.parse_args()

    if args.validate:
        stmt = json.load(open(args.validate, encoding="utf-8"))
        problems = validate_statement(stmt)
        print("BEIS conformance statement: %s" % args.validate)
        print("  system  %s" % stmt.get("system"))
        print("  run by  %s" % stmt.get("run_by"))
        print("")
        for p in problems:
            print("  INVALID  " + p)
        if not problems:
            print("  VALID. Every metric measured is present, every unmeasured metric gives a")
            print("  reason, and the question set and runner are declared.")
            print("")
            print("  That does not mean the SYSTEM is good. It means the statement can be read")
            print("  honestly, which is a different and smaller claim.")
        return 1 if problems else 0

    if not args.endpoint or not args.questions:
        ap.print_help()
        return 2
    blob = json.load(open(args.questions, encoding="utf-8"))
    questions = blob.get("questions", blob)
    if not isinstance(questions, list) or not questions:
        print("no questions found. See the format in this file's header.")
        return 2

    r = run(args.endpoint, questions, args.timeout, args.insecure)

    if args.json:
        print(json.dumps({"endpoint": args.endpoint, "measured": r,
                          "not_measured": NOT_MEASURABLE}, indent=2))
        return 0

    print("BEIS conformance")
    print("  endpoint   %s" % args.endpoint)
    print("  questions  %d   graded for retrieval: %d   refusal cases: %d"
          % (r["questions"], r["graded"], r["refusal_cases"]))
    if r["errors"]:
        print("  errors     %d" % r["errors"])
        for q, e in r["error_detail"]:
            print("      %-46s %s" % (q[:46], e))
    print("")
    for k in MEASURABLE:
        if k == "speed":
            print("  %-14s p50 %sms, p95 %sms" % (k, r["p50_ms"], r["p95_ms"]))
        else:
            v = r[k]
            print("  %-14s %s" % (k, "%.1f%%" % v if v is not None else
                                  "no cases supplied"))
    print("")
    print("  NOT MEASURED from outside, and reported rather than skipped:")
    for k, why in NOT_MEASURABLE.items():
        print("    %-14s %s" % (k, why))
    print("")
    if r["refusal_cases"] == 0:
        print("  ⚠ No refusal cases were supplied. A benchmark with none measures")
        print("    confidence rather than accuracy: a system that answers everything")
        print("    scores perfectly and tells you nothing.")
    print("  This harness has no connection to BvLogic and sends nothing to us. The")
    print("  result is yours, including a bad one.")

    if args.statement:
        import datetime
        # EVERY metric, including the bad ones. There is deliberately no
        # parameter selecting which to include: a statement that can be
        # filtered is a slide, not a measurement.
        measured = {k: (("p50 %sms, p95 %sms" % (r["p50_ms"], r["p95_ms"])) if k == "speed"
                        else ("%.1f%%" % r[k] if r[k] is not None else "no cases supplied"))
                    for k in MEASURABLE}
        stmt = {
            "beis_version": "0.1",
            "statement_format": "0.1",
            "system": args.system or args.endpoint,
            "endpoint": args.endpoint,
            "date": datetime.date.today().isoformat(),
            "run_by": args.run_by,
            "question_set": {"count": r["questions"], "source": args.question_source,
                             "refusal_cases": r["refusal_cases"]},
            "measured": measured,
            "not_measured": dict(NOT_MEASURABLE),
            "errors": r["errors"],
        }
        if r["refusal_cases"] == 0:
            stmt["caveat"] = ("No refusal cases were supplied, so the accuracy figures "
                              "measure confidence rather than accuracy. Weigh them accordingly.")
        with open(args.statement, "w", encoding="utf-8") as fh:
            json.dump(stmt, fh, indent=2)
        print("")
        print("  statement written: %s" % args.statement)
        print("  It carries every metric measured and every metric that could not be.")
    return 0


if __name__ == "__main__":
    sys.exit(main())
