{"name":"skill-drift-final","lang":"python3","entry":"spec-drift-final.py","script":"#!/usr/bin/env python3\n\"\"\"spec-drift-final.py — N5 (终检) verifier.\n\nMetaTask v1.3 competitive-mode verifier (protocol draft section 4.3 CI-style contract).\n\nArtifact contract: metafile bundle: join-report.json (独立重算 + 外部来源带标识 + 未解决项非空)\n\nEnvironment: METATASK_ARTIFACT_URI / METATASK_NODE / METATASK_TASKID /\n  METATASK_DOWNLOAD_BASE (metafile resolution: <base>/file/<urlencoded URI>) /\n  METATASK_RESOLVE=0 disables live link resolution (offline dry runs);\n  METATASK_INSECURE=1 allows an unverified TLS context when no CA store exists.\n\nExit codes: 0 pass (evidence on stdout) / 1 fail (reason on stderr) /\n2 invalid (null / empty / unresolvable input). The pass path asserts the\nenumeration closure: the check counter must equal EXPECTED_CHECKS.\n\"\"\"\n\nEXPECTED_CHECKS = 9\n\nimport io\nimport json\nimport os\nimport re\nimport sys\nimport tarfile\nimport urllib.error\nimport urllib.parse\nimport urllib.request\nimport zipfile\n\n\nchecks_done = 0\n\n\ndef ok(msg):\n    global checks_done\n    checks_done += 1\n    print(\"OK   \" + msg)\n\n\ndef emit(verdict, detail):\n    print(json.dumps({\"verdict\": verdict, \"detail\": detail, \"checks\": checks_done}, ensure_ascii=False))\n\n\ndef closure():\n    \"\"\"Assert the enumeration closure without consuming a check slot.\"\"\"\n    if checks_done != EXPECTED_CHECKS:\n        die_fail(\"enumeration closure: %d/%d checks\" % (checks_done, EXPECTED_CHECKS))\n    print(\"OK   enumeration closure: %d/%d checks\" % (checks_done, EXPECTED_CHECKS))\n\n\ndef die_fail(msg):\n    emit(\"fail\", msg)\n    print(\"FAIL: \" + msg, file=sys.stderr)\n    sys.exit(1)\n\n\ndef die_invalid(msg):\n    emit(\"invalid\", msg)\n    sys.exit(2)\n\n\ndef check(cond, msg):\n    if not cond:\n        die_fail(msg)\n    ok(msg)\n\n\ndef _guard(exc_type, exc, tb):\n    \"\"\"Any unhandled error is INVALID input, never a fail verdict.\"\"\"\n    if issubclass(exc_type, SystemExit):\n        return\n    emit(\"invalid\", \"unhandled %s: %s\" % (exc_type.__name__, exc))\n    sys.exit(2)\n\n\nsys.excepthook = _guard\n\n\ndef ssl_context():\n    \"\"\"CA bundle for HTTPS: certifi when available; METATASK_INSECURE=1 opts out.\n\n    The host Python on some macOS installs ships without the system CA store, so\n    a bare urlopen fails SSL verification while curl succeeds. Reviewers get a\n    documented opt-out, and every use is logged as reduced evidence.\n    \"\"\"\n    import ssl\n    try:\n        import certifi\n        ctx = ssl.create_default_context(cafile=certifi.where())\n    except Exception:  # noqa: BLE001\n        ctx = ssl.create_default_context()\n    if os.environ.get(\"METATASK_INSECURE\") == \"1\":\n        print(\"NOTE ssl verification disabled (METATASK_INSECURE=1) — resolution evidence is weaker\")\n        ctx.check_hostname = False\n        ctx.verify_mode = ssl.CERT_NONE\n    return ctx\n\n\ndef artifact_uri():\n    uri = os.environ.get(\"METATASK_ARTIFACT_URI\", \"\").strip()\n    if not uri:\n        die_invalid(\"METATASK_ARTIFACT_URI is empty\")\n    return uri\n\n\ndef fetch_bytes(uri):\n    if uri.startswith((\"metafile://\", \"pin://\")):\n        base = os.environ.get(\"METATASK_DOWNLOAD_BASE\", \"\").strip()\n        if not base:\n            die_invalid(\"METATASK_DOWNLOAD_BASE unset — cannot resolve \" + uri)\n        url = base.rstrip(\"/\") + \"/file/\" + urllib.parse.quote(uri, safe=\"\")\n        path = None\n    elif uri.startswith(\"file://\"):\n        url, path = None, uri[7:]\n    else:\n        url, path = None, uri\n    try:\n        if url:\n            req = urllib.request.Request(url, headers={\"User-Agent\": \"metatask-spec/1.0\"})\n            with urllib.request.urlopen(req, timeout=60, context=ssl_context()) as resp:\n                return resp.read()\n        with open(path, \"rb\") as fh:\n            return fh.read()\n    except Exception as exc:  # noqa: BLE001 - any fetch failure is an invalid run\n        die_invalid(\"artifact fetch failed for %s: %r\" % (uri, exc))\n\n\ndef load_members(data):\n    \"\"\"Return {name: bytes} for an archive, or {\"__raw__\": decoded text}.\"\"\"\n    if data[:2] != b\"\\x1f\\x8b\" and data[:2] != b\"PK\":\n        return {\"__raw__\": data.decode(\"utf-8\", \"replace\")}\n\n    members = {}\n    if data[:2] == b\"PK\":\n        with zipfile.ZipFile(io.BytesIO(data)) as arc:\n            names = [n for n in arc.namelist() if not n.endswith(\"/\")]\n            for name in names:\n                members[name] = arc.read(name)\n    else:\n        with tarfile.open(fileobj=io.BytesIO(data), mode=\"r:gz\") as arc:\n            names = [n for n in arc.getnames() if not n.endswith(\"/\")]\n            for name in names:\n                handle = arc.extractfile(name)\n                if handle is not None:\n                    members[name] = handle.read()\n\n    if not members:\n        die_invalid(\"artifact archive is empty — nothing to verify\")\n\n    # 仅在「所有成员共享同一个顶层目录、且该目录不是契约要求的目录名」时剥离。\n    # 契约里的 experiment/ 与 repro/ 是语义目录，绝不能当作包装目录剥掉。\n    RESERVED = {\"experiment\", \"repro\", \"fixtures\"}\n    if members and all(\"/\" in n for n in members):\n        tops = {n.split(\"/\")[0] for n in members}\n        if len(tops) == 1:\n            top = sorted(tops)[0]\n            if top not in RESERVED:\n                members = {n[len(top) + 1:]: v for n, v in members.items()}\n    return members\n\n\ndef text_of(members, name):\n    if name in members:\n        return members[name].decode(\"utf-8\", \"replace\")\n    raw = members.get(\"__raw__\")\n    if raw is not None and name.endswith(\".md\"):\n        return raw\n    return None\n\n\ndef as_text(value):\n    if isinstance(value, (bytes, bytearray)):\n        return value.decode(\"utf-8\", \"replace\")\n    return value\n\n\ndef json_of(members, name):\n    for key in (name, \"__raw__\"):\n        if key not in members:\n            continue\n        text = as_text(members[key])\n        if not isinstance(text, str):\n            return None\n        try:\n            return json.loads(text)\n        except Exception:  # noqa: BLE001\n            if key == name:\n                die_fail(\"%s is present but is not valid JSON\" % name)\n            return None\n    return None\n\n\nMETA_BLOCK_RE = re.compile(r\"<!--\\\\s*meta:.*?-->\", re.S)\n\n\ndef body_only(doc):\n    \"\"\"The document without its machine-readable meta declarations.\n\n    Declarations must never satisfy a check about the document body: strip them\n    first, or a document that declares 50 citations and writes nothing passes.\n    \"\"\"\n    return META_BLOCK_RE.sub(\"\", doc)\n\n\ndef raw_list(value):\n    if value is None:\n        return []\n    return [x.strip() for x in value.split(\",\") if x.strip()]\n\n\ndef resolve_ratio(urls):\n    \"\"\"Live-resolve links. Returns (good, total) or None when disabled.\"\"\"\n    if os.environ.get(\"METATASK_RESOLVE\", \"1\") == \"0\":\n        return None\n    good = 0\n    for url in urls:\n        try:\n            req = urllib.request.Request(url, headers={\"User-Agent\": \"Mozilla/5.0 (metatask-spec)\"})\n            with urllib.request.urlopen(req, timeout=20, context=ssl_context()) as resp:\n                if resp.status == 200:\n                    good += 1\n        except urllib.error.HTTPError as exc:\n            if exc.code >= 500:\n                die_invalid(\"server error %s while resolving %s\" % (exc.code, url))\n        except Exception as exc:  # noqa: BLE001\n            die_invalid(\"network unavailable while resolving links: %r\" % exc)\n    return good, len(urls)\n\n\ndef meta_block(doc, key):\n    \"\"\"Read <!-- meta: key=value --> from a document.\"\"\"\n    hit = re.search(r\"<!--\\\\s*meta:\\\\s*\" + key + r\"=([^>]*?)-->\", doc)\n    return hit.group(1).strip() if hit else None\n\n\ndef id_list(value):\n    if value is None:\n        return None\n    return sorted(x.strip() for x in value.split(\",\") if x.strip())\n\n# --- N5: 终检合并（join-report.json + 上游产物）----------------------------------\n\nREQUIRED = (\"join-report.json\",)\n\nARTIFACT = artifact_uri()\nmembers = load_members(fetch_bytes(ARTIFACT))\nmissing_members = [m for m in REQUIRED if m not in members]\ncheck(not missing_members, \"join bundle carries join-report.json (missing: %s)\" % (missing_members or \"none\"))\n\nreport = json_of(members, \"join-report.json\")\ncheck(isinstance(report, dict) and isinstance(report.get(\"unresolved\"), list),\n      \"join-report.json parses and declares unresolved[]\")\n\nrecomputed = report.get(\"recomputed\")\ncheck(isinstance(recomputed, dict) and bool(recomputed),\n      \"join-report carries a recomputation block (independent, not self-reported)\")\n\nclaimed = report.get(\"claimed\")\ncheck(isinstance(claimed, dict) and bool(claimed),\n      \"join-report carries the claimed figures to be compared against\")\n\nagreement = report.get(\"agreement\") or {}\ncheck(isinstance(agreement, dict) and \"all_match\" in agreement,\n      \"join-report states whether recomputation agreed with the claim (all_match present)\")\n\nsources = report.get(\"external_sources\")\ncheck(isinstance(sources, list) and len(sources) >= 10,\n      \"join-report records >= 10 external-source spot checks (got %s)\" % (len(sources) if isinstance(sources, list) else \"n/a\"))\n\nblank_src = [i for i, u in enumerate(sources or [])\n             if not (isinstance(u, dict) and any(str(v).strip() for v in u.values()))\n             and not (isinstance(u, str) and u.strip())]\ncheck(not blank_src,\n      \"every external-source entry carries a non-blank identifier (blank at: %s)\" % (blank_src[:5] or \"none\"))\n\nunresolved_empty = report.get(\"unresolved\") == []\ncheck(not unresolved_empty,\n      \"unresolved[] is NOT empty — known boundaries must be recorded, not silently cleared\")\n\nuris = report.get(\"artifacts\")\ncheck(isinstance(uris, list) and uris and all(str(u).startswith((\"pin://\", \"metafile://\", \"metaapp://\")) for u in uris),\n      \"every delivered artifact is cited by a chain URI (pin://, metafile://, metaapp://)\")\n\nclosure()\nemit(\"pass\", \"terminal join verified: %d external spot checks, %d artifacts cited\"\n     % (len(sources), len(uris)))\n","input":{"env":{"METATASK_ARTIFACT_URI":"metafile://<submission artifact>","METATASK_NODE":"<node id this spec judges>","METATASK_TASKID":"<task root pinId>","METATASK_DOWNLOAD_BASE":"<metafile resolution base; fetches <base>/file/<urlencoded URI>>","METATASK_RESOLVE":"1 (default) | 0 to disable live link resolution in offline runs","METATASK_INSECURE":"1 allows an unverified TLS context when no CA store exists"}},"output":{"stdout":"per-check evidence log + one JSON verdict line","exitCode":"0 pass | 1 fail | 2 invalid (null/empty/unresolvable input)"},"workspace":{"type":"metafile","notes":"non-git node: 提交件为 metafile 压缩包"},"validation":{"null_tolerance":true,"enumeration_closure":{"closure":"the check counter must equal EXPECTED_CHECKS on the pass path","selfcheck":{"expected_count":9}},"proposition_fidelity":{"correspondence":"pin://aa070e492601506cd02475f32f9dc302eb9039000bc57b02d4bb9c9484292c09i0","artifactPin":"pin://aa070e492601506cd02475f32f9dc302eb9039000bc57b02d4bb9c9484292c09i0"}}}