{"name":"s4-report-structure","lang":"python3","entry":"spec-s4-report.py","script":"#!/usr/bin/env python3\n\"\"\"\nspec-s4-report.py — S4 (divergence root-cause report) verifier.\n\nMetaTask v1.3 pilot spec script (protocol draft §4.3 CI-style verifier contract).\n\nArtifact contract (acceptance-sheet.md §S4): the submission attachment is a\nmetafile pointing at a .tar.gz with members:\n  matrix.md   the S3 results matrix (same format as §S3): a table\n              `| vector | python | go | ts |` whose cells are PASS | RED |\n              AMBER, plus the per-engine digest lines\n  report.md   the divergence report. Every RED/AMBER matrix cell is attributed\n              by a section heading `#### ATTR: <vector-id> / <engine>` whose\n              body carries a `severity: low|medium|high|critical` line and\n              cites >= 1 spec clause `C-NN`.\n\nThese are STRUCTURAL checks: whether an attribution is correct (engine bug vs\nspec ambiguity, fix implementability) is the reviewer rubric's job.\n\nEnvironment (§4.3, non-git node): METATASK_ARTIFACT_URI / METATASK_NODE /\nMETATASK_TASKID (+ METATASK_DOWNLOAD_BASE for metafile resolution).\n\nExit codes: 0 pass · 1 fail · 2 invalid. Check counter asserted against\nEXPECTED_CHECKS on the pass path (enumeration closure self-check).\n\"\"\"\nimport hashlib\nimport io\nimport json\nimport os\nimport re\nimport sys\nimport tarfile\nimport zipfile\n\nEXPECTED_CHECKS = 7\nCELL_VALUES = {\"pass\", \"red\", \"amber\"}\nATTR_RE = re.compile(r\"^#{3,5}\\s+ATTR:\\s*(\\S+)\\s*/\\s*(python|go|ts)\\s*$\", re.IGNORECASE)\nSEVERITY_RE = re.compile(r\"^severity:\\s*(low|medium|high|critical)\\s*$\", re.IGNORECASE | re.MULTILINE)\nCLAUSE_REF_RE = re.compile(r\"\\bC-\\d{2,}\\b\")\n\nchecks_done = 0\n\n\ndef note(line):\n    print(line)\n\n\ndef emit(verdict, detail):\n    print(json.dumps({\"verdict\": verdict, \"detail\": detail, \"checks\": checks_done}, ensure_ascii=False))\n\n\ndef invalid(detail):\n    emit(\"invalid\", detail)\n    sys.exit(2)\n\n\ndef fail(detail):\n    emit(\"fail\", detail)\n    print(\"FAIL: %s\" % detail, file=sys.stderr)\n    sys.exit(1)\n\n\ndef check(name, ok, detail=\"\"):\n    global checks_done\n    checks_done += 1\n    note(\"[check %d] %s: %s%s\" % (checks_done, name, \"ok\" if ok else \"FAIL\", (\" — \" + detail) if detail else \"\"))\n    if not ok:\n        fail(\"%s%s\" % (name, (\" — \" + detail) if detail else \"\"))\n\n\ndef gate(name, ok, detail=\"\"):\n    global checks_done\n    checks_done += 1\n    note(\"[check %d] %s: %s%s\" % (checks_done, name, \"ok\" if ok else \"INVALID\", (\" — \" + detail) if detail else \"\"))\n    if not ok:\n        invalid(\"%s%s\" % (name, (\" — \" + detail) if detail else \"\"))\n\n\ndef resolve_artifact(uri):\n    if uri.startswith(\"file://\"):\n        return uri[len(\"file://\"):]\n    if os.path.isfile(uri):\n        return uri\n    if uri.startswith(\"metafile://\") or uri.startswith(\"pin://\"):\n        base = os.environ.get(\"METATASK_DOWNLOAD_BASE\", \"\").strip()\n        if not base:\n            note(\"[fetch] %s requires METATASK_DOWNLOAD_BASE (not set)\" % uri)\n            return None\n        from urllib.parse import quote\n        from urllib.request import urlopen\n\n        url = \"%s/file/%s\" % (base.rstrip(\"/\"), quote(uri, safe=\"\"))\n        target = os.path.join(os.environ.get(\"TMPDIR\", \"/tmp\"), \"metatask-fetch-%s\" % hashlib.sha256(uri.encode()).hexdigest()[:16])\n        try:\n            with urlopen(url, timeout=120) as response:\n                data = response.read()\n            with open(target, \"wb\") as handle:\n                handle.write(data)\n            note(\"[fetch] %s → %s (%d bytes)\" % (uri, url, len(data)))\n            return target\n        except Exception as err:\n            note(\"[fetch] %s via %s failed: %s\" % (uri, url, err))\n            return None\n    note(\"[fetch] unrecognized artifact URI scheme: %s\" % uri)\n    return None\n\n\ndef open_archive(path):\n    raw_members = {}\n    with open(path, \"rb\") as handle:\n        blob = handle.read()\n    if tarfile.is_tarfile(path):\n        with tarfile.open(fileobj=io.BytesIO(blob)) as tar:\n            for member in tar.getmembers():\n                if member.isfile():\n                    raw_members[member.name] = tar.extractfile(member).read()\n    elif zipfile.is_zipfile(io.BytesIO(blob)):\n        with zipfile.ZipFile(io.BytesIO(blob)) as zf:\n            for name in zf.namelist():\n                if not name.endswith(\"/\"):\n                    raw_members[name] = zf.read(name)\n    else:\n        fail(\"artifact is neither a tar archive nor a zip file\")\n    prefixes = set(name.split(\"/\")[0] for name in raw_members if \"/\" in name)\n    bare = [name for name in raw_members if \"/\" not in name]\n    if not bare and len(prefixes) == 1:\n        prefix = list(prefixes)[0] + \"/\"\n        raw_members = {name[len(prefix):]: data for name, data in raw_members.items()}\n    return raw_members\n\n\ndef parse_matrix_cells(matrix_text):\n    \"\"\"Return (cells, divergent) where cells = {(vector, engine): value}.\"\"\"\n    cells = {}\n    in_table = False\n    engines = []\n    for line in matrix_text.splitlines():\n        stripped = line.strip()\n        if not (stripped.startswith(\"|\") and stripped.endswith(\"|\")):\n            continue\n        parts = [cell.strip() for cell in stripped.strip(\"|\").split(\"|\")]\n        if parts and parts[0].lower() == \"vector\":\n            engines = [part.lower() for part in parts[1:]]\n            in_table = True\n            continue\n        if in_table and parts and not set(parts[0]) <= set(\"-: \"):\n            if len(parts) != len(engines) + 1:\n                return None, \"row %r has %d cells, expected %d\" % (parts[0], len(parts) - 1, len(engines))\n            for engine, value in zip(engines, parts[1:]):\n                if value.lower() not in CELL_VALUES:\n                    return None, \"cell %s/%s has value %r (want PASS|RED|AMBER)\" % (parts[0], engine, value)\n                cells[(parts[0], engine)] = value.lower()\n    if not cells:\n        return None, \"no matrix table found\"\n    divergent = sorted([key for key, value in cells.items() if value in (\"red\", \"amber\")])\n    return (cells, divergent)\n\n\ndef parse_attributions(report_text):\n    \"\"\"Return { (vector, engine): section_body }.\"\"\"\n    attributions = {}\n    current_key = None\n    current_body = []\n    for line in report_text.splitlines():\n        match = ATTR_RE.match(line.strip())\n        if match:\n            if current_key is not None:\n                attributions[current_key] = \"\\n\".join(current_body)\n            current_key = (match.group(1), match.group(2).lower())\n            current_body = []\n        elif current_key is not None:\n            if re.match(r\"^#{1,5}\\s\", line.strip()) and not ATTR_RE.match(line.strip()):\n                attributions[current_key] = \"\\n\".join(current_body)\n                current_key = None\n                current_body = []\n            else:\n                current_body.append(line)\n    if current_key is not None:\n        attributions[current_key] = \"\\n\".join(current_body)\n    return attributions\n\n\ndef main():\n    artifact = os.environ.get(\"METATASK_ARTIFACT_URI\", \"\").strip()\n    node = os.environ.get(\"METATASK_NODE\", \"\").strip()\n    taskid = os.environ.get(\"METATASK_TASKID\", \"\").strip()\n    note(\"[env] node=%s taskid=%s\" % (node or \"<missing>\", taskid or \"<missing>\"))\n    note(\"[env] artifact=%s\" % (artifact or \"<missing>\"))\n    note(\"[tool] python3: %s\" % sys.version.split()[0])\n\n    gate(\"env contract present\", bool(artifact and node and taskid),\n         \"METATASK_ARTIFACT_URI / METATASK_NODE / METATASK_TASKID must all be non-empty\")\n\n    local = resolve_artifact(artifact)\n    gate(\"artifact resolves\", local is not None, artifact)\n    if os.path.getsize(local) == 0:\n        invalid(\"artifact is an empty file (0 bytes): %s\" % artifact)\n\n    members = open_archive(local)\n    note(\"[archive] members: %s\" % \", \".join(sorted(members)))\n    missing = [name for name in (\"report.md\", \"matrix.md\") if name not in members]\n    check(\"report.md + matrix.md present\", not missing, \"missing: %s\" % \", \".join(missing) if missing else \"present\")\n    if not members[\"report.md\"].strip() or not members[\"matrix.md\"].strip():\n        invalid(\"report.md or matrix.md is empty\")\n\n    cells, divergent = parse_matrix_cells(members[\"matrix.md\"].decode(\"utf-8\"))\n    check(\"matrix table parses; every cell is PASS|RED|AMBER\", cells is not None,\n          \"%d cells, %d divergent\" % (len(cells), len(divergent)) if cells is not None else divergent)\n\n    attributions = parse_attributions(members[\"report.md\"].decode(\"utf-8\"))\n    missing_attr = [key for key in divergent if key not in attributions]\n    extra_attr = [key for key in attributions if key not in divergent]\n    check(\"every RED/AMBER cell has exactly one ATTR section (count matches)\",\n          not missing_attr and not extra_attr,\n          (\"missing: %s \" % \", \".join(\"%s/%s\" % k for k in missing_attr[:4]) if missing_attr else \"\")\n          + (\"extra: %s\" % \", \".join(\"%s/%s\" % k for k in extra_attr[:4]) if extra_attr else \"\")\n          or \"%d divergent cells, %d attributions\" % (len(divergent), len(attributions)))\n\n    no_severity = [key for key, body in attributions.items() if not SEVERITY_RE.search(body)]\n    check(\"every attribution declares severity (low|medium|high|critical)\",\n          not no_severity, \"missing severity: %s\" % \", \".join(\"%s/%s\" % k for k in no_severity[:4]) if no_severity else \"all severed\")\n\n    no_clause = [key for key, body in attributions.items() if not CLAUSE_REF_RE.search(body)]\n    check(\"every attribution cites >= 1 spec clause (C-NN)\",\n          not no_clause, \"no clause citation: %s\" % \", \".join(\"%s/%s\" % k for k in no_clause[:4]) if no_clause else \"all cited\")\n\n    if checks_done != EXPECTED_CHECKS:\n        invalid(\"check-counter mismatch: ran %d, spec declares %d (script bug)\" % (checks_done, EXPECTED_CHECKS))\n    emit(\"pass\", \"S4 structure green: %d matrix cells, %d divergent, all attributed with severity + clause citations\"\n         % (len(cells), len(divergent)))\n\n\nif __name__ == \"__main__\":\n    main()\n","input":{"env":{"METATASK_ARTIFACT_URI":"metafile://<S4 artifact tar.gz>","METATASK_NODE":"S4","METATASK_TASKID":"<task root pinId>","METATASK_DOWNLOAD_BASE":"optional; metafile resolution base"}},"output":{"stdout":"evidence log + final JSON line {\"verdict\",\"detail\",\"checks\"}","exitCode":"0 pass | 1 fail | 2 invalid"},"workspace":{"type":"metafile","notes":"S4 delivers one metafile bundle: report.md + matrix.md"},"validation":{"null_tolerance":true,"enumeration_closure":{"closure":"the attribution enumeration closes over the matrix's RED/AMBER cells: every divergent cell has exactly one ATTR section and vice versa; the pass path asserts the check counter equals the declared expected_count","selfcheck":{"expected_count":7}},"proposition_fidelity":{"correspondence":"metafile://d7f836544764bf4ca68bd7903b7a6c7a978ff16e34a828e9b31c928cbec05ccci0.md","artifactPin":"metafile://d7f836544764bf4ca68bd7903b7a6c7a978ff16e34a828e9b31c928cbec05ccci0.md","artifactKey":"acceptance-sheet","coverage":["statement","definitions","proof-direction"],"note":"statement = S4 deliverable (every divergent cell attributed); definitions = the ATTR section format; proof-direction = deliberately structural only - attribution substance is the reviewer rubric's (acceptance-sheet S4)."}}}