#!/usr/bin/env python3
"""Handoff Check v0.1 — Python 3.9+, standard library only.

Usage: python handoff_check.py handoff.json [--json]
       python handoff_check.py --example > handoff.json
       python handoff_check.py --self-test

Schema: exact top-level fields shown by --example. Text must be nonempty.
constraints/decisions/uncertainties: lists of nonempty strings (may be empty).
checks: objects with unique id, scope, status (passed/failed/not_run).
claims: objects with text and supporting_check_ids (unique string references).
All fields are required; unknown fields are errors to catch misspellings.
Empty evidence and non-passing cited checks produce warnings. Unknown evidence
IDs are errors. Mixed passed/failed evidence still warns. Duplicate JSON keys
and nonstandard numeric constants are rejected. Input limit: 1 MiB.

Exit: 0 = no findings, 1 = warnings, 2 = invalid input/schema/references.
No result certifies task completion, truth, or adequate test scope. This tool
cannot detect omitted constraints, fabricated evidence, or semantic mismatch
between a claim and its checks. All example data is fictional. No network calls.
"""
import argparse
import copy
import json
import sys
import unittest

LIMIT = 1024 * 1024
EXAMPLE = {
    "goal": "Export the visible fictional task rows",
    "constraints": ["Preserve saved manual order"],
    "decisions": ["Export a snapshot of the visible rows"],
    "checks": [{"id": "csv-unit", "scope": "Direct exporter comma quoting", "status": "passed"}],
    "claims": [{"text": "Direct exporter comma quoting passed", "supporting_check_ids": ["csv-unit"]}],
    "uncertainties": ["Preview-button wiring and reload persistence remain untested"],
    "next_action": "Export through the preview button and compare saved order after reload",
}


def validate(doc):
    findings = []

    def issue(path, code, message, severity="error"):
        findings.append(dict(path=path, code=code, message=message, severity=severity))

    def obj(value, fields, path):
        if not isinstance(value, dict):
            issue(path, "type", "Expected an object")
            return False
        for key in sorted(fields - value.keys()):
            issue(path + "." + key, "missing_field", "Required field missing")
        for key in sorted(value.keys() - fields):
            issue(path + "." + key, "unknown_field", "Unknown field")
        return True

    def text(value, path):
        if not isinstance(value, str) or not value.strip():
            issue(path, "text", "Expected a nonempty string")
            return False
        return True

    def array(value, path):
        if not isinstance(value, list):
            issue(path, "type", "Expected an array")
            return []
        return value

    if not obj(doc, set(EXAMPLE), "$"):
        return findings
    for key in ("goal", "next_action"):
        if key in doc:
            text(doc[key], "$." + key)
    for key in ("constraints", "decisions", "uncertainties"):
        if key in doc:
            for i, value in enumerate(array(doc[key], "$." + key)):
                text(value, f"$.{key}[{i}]")
    checks = {}
    ambiguous = set()
    for i, check in enumerate(array(doc.get("checks", []), "$.checks")):
        path = f"$.checks[{i}]"
        if not obj(check, {"id", "scope", "status"}, path):
            continue
        ident = check.get("id")
        valid_id = text(ident, path + ".id")
        text(check.get("scope"), path + ".scope")
        status = check.get("status")
        if status not in ("passed", "failed", "not_run"):
            issue(path + ".status", "status", "Expected passed, failed, or not_run")
        if valid_id:
            if ident in checks:
                ambiguous.add(ident)
                issue(path + ".id", "duplicate_id", "Check IDs must be unique")
            checks[ident] = status
    for i, claim in enumerate(array(doc.get("claims", []), "$.claims")):
        path = f"$.claims[{i}]"
        if not obj(claim, {"text", "supporting_check_ids"}, path):
            continue
        text(claim.get("text"), path + ".text")
        if "supporting_check_ids" not in claim:
            continue
        refs = array(claim["supporting_check_ids"], path + ".supporting_check_ids")
        if not refs and isinstance(claim["supporting_check_ids"], list):
            issue(path, "no_evidence", "Claim cites no checks", "warning")
        seen = set()
        for j, ref in enumerate(refs):
            rp = f"{path}.supporting_check_ids[{j}]"
            if not text(ref, rp):
                continue
            if ref in seen:
                issue(rp, "duplicate_reference", "Repeated evidence reference")
            seen.add(ref)
            if ref not in checks:
                issue(rp, "unknown_reference", "Referenced check does not exist")
            elif ref in ambiguous:
                issue(rp, "ambiguous_reference", "Referenced check ID is duplicated")
            elif checks[ref] != "passed":
                issue(rp, "nonpassing_evidence", "Cited check did not pass", "warning")
    return findings


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("Duplicate JSON object key")
        result[key] = value
    return result


def reject_constant(value):
    raise ValueError("Nonstandard JSON constant")


def parse(raw):
    if len(raw) > LIMIT:
        raise ValueError("Input exceeds 1 MiB")
    return json.loads(raw.decode("utf-8"), object_pairs_hook=unique_object,
                      parse_constant=reject_constant)


def self_test():
    class ContractTests(unittest.TestCase):
        def test_contract_cases(self):
            cases = [
                (lambda d: None, set()),
                (lambda d: d.pop("goal"), {"missing_field"}),
                (lambda d: d.update(next_action=" "), {"text"}),
                (lambda d: d.update(constraints="oops"), {"type"}),
                (lambda d: d.update(checks=[None]), {"type", "unknown_reference"}),
                (lambda d: d["checks"].append(copy.deepcopy(d["checks"][0])), {"duplicate_id", "ambiguous_reference"}),
                (lambda d: d["checks"][0].update(status="failed"), {"nonpassing_evidence"}),
                (lambda d: d["checks"][0].update(status="not_run"), {"nonpassing_evidence"}),
                (lambda d: d["claims"][0].update(supporting_check_ids=[]), {"no_evidence"}),
                (lambda d: d["claims"][0].update(supporting_check_ids=["missing"]), {"unknown_reference"}),
                (lambda d: d["claims"][0].update(supporting_check_ids=[{}]), {"text"}),
                (lambda d: d["claims"][0].update(supporting_check_ids=["csv-unit", "csv-unit"]), {"duplicate_reference"}),
                (lambda d: d.update(cliams=[]), {"unknown_field"}),
                (lambda d: d["checks"][0].update(status=[]), {"status", "nonpassing_evidence"}),
            ]
            for index, (mutate, expected) in enumerate(cases):
                with self.subTest(case=index):
                    doc = copy.deepcopy(EXAMPLE)
                    mutate(doc)
                    self.assertEqual({f["code"] for f in validate(doc)}, expected)

        def test_mixed_evidence(self):
            doc = copy.deepcopy(EXAMPLE)
            doc["checks"].append(dict(id="browser", scope="Preview button", status="failed"))
            doc["claims"][0]["supporting_check_ids"].append("browser")
            self.assertEqual([f["code"] for f in validate(doc)], ["nonpassing_evidence"])

        def test_unrelated_pass_does_not_hide_warning(self):
            doc = copy.deepcopy(EXAMPLE)
            doc["checks"][0]["status"] = "failed"
            doc["checks"].append(dict(id="other", scope="Unrelated", status="passed"))
            self.assertEqual([f["code"] for f in validate(doc)], ["nonpassing_evidence"])

        def test_strict_json(self):
            for raw in (b'{"a":1,"a":2}', b'{"a":NaN}', b'not json', b'\xff', b' ' * (LIMIT + 1)):
                with self.subTest(raw_prefix=raw[:20]):
                    with self.assertRaises(ValueError):
                        parse(raw)

        def test_root_and_roundtrip(self):
            self.assertEqual(validate(parse(json.dumps(EXAMPLE).encode())), [])
            self.assertEqual(validate([])[0]["code"], "type")

    return 0 if unittest.TextTestRunner().run(unittest.defaultTestLoader.loadTestsFromTestCase(ContractTests)).wasSuccessful() else 2


def main():
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("file", nargs="?", help="JSON input file, or - for stdin")
    parser.add_argument("--json", action="store_true", help="Machine-readable report")
    parser.add_argument("--example", action="store_true")
    parser.add_argument("--self-test", action="store_true")
    args = parser.parse_args()
    if args.self_test:
        return self_test()
    if args.example:
        print(json.dumps(EXAMPLE, indent=2))
        return 0
    if not args.file:
        parser.error("provide a JSON file, --example, or --self-test")
    try:
        if args.file == "-":
            raw = sys.stdin.buffer.read(LIMIT + 1)
        else:
            with open(args.file, "rb") as handle:
                raw = handle.read(LIMIT + 1)
        findings = validate(parse(raw))
    except (OSError, ValueError, RecursionError):
        findings = [dict(path="$", code="input_error", severity="error",
                         message="Cannot read input: require UTF-8 JSON, unique keys, finite constants, <=1 MiB, reasonable nesting")]
    code = 2 if any(f["severity"] == "error" for f in findings) else 1 if findings else 0
    report = dict(structurally_valid=code != 2, findings=findings,
                  semantic_review_required=True, completion_certified=False)
    if args.json:
        print(json.dumps(report, indent=2))
    else:
        print("Invalid handoff" if code == 2 else "Valid structure with warnings" if code else "No structural findings")
        for finding in findings:
            print("{severity}: {path}: {code}: {message}".format(**finding))
        print("Semantic review is required; completion is not certified.")
    return code


if __name__ == "__main__":
    sys.exit(main())