feat(scholar): scholarly-graph federation (OpenAlex + Semantic Scholar + Crossref) with citation-edge provenance

CHANGELOG.md

@@ -5,6 +5,21 @@ built behind a feature branch and reviewed before merge.
 
 ## Unreleased
 
+- `gather.scholar`: a scholarly-graph federation adapter unifying OpenAlex, Semantic Scholar,
+  and Crossref into one intake. Pure per-provider parsers (`parse_openalex`,
+  `parse_semanticscholar`, `parse_crossref`) normalize each graph's shape onto one
+  provider-neutral `ScholarWork` (OpenAlex abstracts reconstructed from the inverted index);
+  `ScholarSource` is the isolated impure edge with an injectable fetcher, so the whole
+  federation is tested offline over recorded fixtures. Citation edges (references and citations)
+  are first-class provenance: `citation_edges` turns each into a re-checkable receipt (method
+  `citation-edge`, registered DIRECT on the method ladder) recording which provider asserted the
+  link, deduped by `(from, to, direction, provider)` and sealed into the digest alongside the
+  papers. `federate` joins works by normalized DOI (`normalize_doi` strips URL/`doi:` prefixes
+  and lowercases) into one unified `compiled` item whose `derived_from` points back at every
+  provider's contribution (no provenance dropped); a DOI is the only join key, never a fuzzy
+  title match, and a DOI-less work stays its own record. CLI: `gather scholar QUERY
+  [--providers ...] [--no-federate] [--edges] [--json]`; `--edges` folds the citation edges into
+  the same witnessed digest as the papers, so the seal covers the graph, not just the nodes.
 - `gather.federation`: the source-federation registry contract. A registry row is a closed
   nine-field shape (`id`, `system`, `family`, `domain`, `access`, `adapter`, `url`, `scope`,
   `priority`) with closed vocabularies for the access policy (`open`, `key_required`,

README.md

@@ -184,6 +184,7 @@ gather docs ./research-notes --scope "rubik,group theory"   # local files, offli
 gather web  "https://example.com/article" --store ./corpus  # static page, kept in a corpus
 gather feed "https://example.com/feed.xml" --json           # RSS or Atom
 gather arxiv "aperiodic monotile" --store ./corpus          # papers (abstracts + metadata)
+gather scholar "10.1234/monotile" --edges --json            # OpenAlex + Semantic Scholar + Crossref, deduped by DOI, with citation edges
 gather video "https://youtu.be/<id>" --comments --scope "rubik,group theory"
 gather corpus verify ./corpus                               # re-hash every stored body
 gather corpus availability ./corpus                         # witness a sealed availability rung per record
@@ -243,6 +244,7 @@ field of a witnessed rule or match breaks verification.
 - `gather.feed`: RSS and Atom feeds; pure parser handles both.
 - `gather.docs`: local text files or a directory of them; the impure edge is the filesystem.
 - `gather.arxiv`: papers from the arXiv API by id or query; pure parser, the Item carries the abstract and metadata.
+- `gather.scholar`: a scholarly-graph federation adapter unifying OpenAlex, Semantic Scholar, and Crossref into one intake. Citation edges (references and citations) are first-class provenance: each edge is its own re-checkable receipt (method `citation-edge`) recording which provider asserted the link, sealed into the digest alongside the papers. Records are deduped by normalized DOI into one unified `compiled` item whose `derived_from` points back at every provider's contribution (no provenance dropped); a DOI is the only join key, never a fuzzy title match. Pure per-provider parsers, one isolated impure edge, and an injectable fetcher so the whole federation is tested offline over recorded fixtures.
 - `gather.pdf`: text from a local PDF via `pdftotext` (an external tool, not a dependency); a best-effort reading, labelled as such.
 - `gather.store`: a durable, content-addressed `Corpus`. Bodies are deduped by hash while every distinct receipt is kept (no provenance dropped); the catalog streams; `verify` re-hashes every stored body (MATCH/MISSING/CORRUPT); the run history is kept too.
 - `gather.run`: the witnessed gather session. `gather_run` orchestrates fetch, scope, optional synthesis, digest, and store into one re-checkable `RunRecord` (its own seal plus the items' digest seal); the scope and synthesizer are composition seams that default to Null so the run stands alone.
@@ -282,6 +284,7 @@ Shipped:
 - The hard sources behind the same seam, as isolated external-tool edges: JavaScript pages (headless browser), scanned images (OCR), and audio (transcription).
 - A real model edge for the synthesizer seam, and a provenance-composition seam that folds an external origin verdict per item into the witnessed run.
 - A seal-covered availability rung per record (`gather corpus availability`): typed re-verification outcomes distinguish a source that no longer answers from one whose content changed, and a legacy record reports unwitnessed, never available.
+- A scholarly-graph federation adapter (`gather scholar`) unifying OpenAlex, Semantic Scholar, and Crossref: DOI-deduped unified records that keep every provider's provenance, and citation edges captured as first-class receipts sealed into the digest.
 
 Gather reached its organic completion at 1.5.0: every planned source and seam is shipped, and the
 accountability claims hold end to end across a final whole-system review. The item below is a scale

src/gather/browser_evidence.py

@@ -18,8 +18,10 @@ def parse_browser_evidence(packet: Mapping, *, fetched_at: float) -> Item:
     target_ref = str(packet.get("target_ref") or "")
     if not target_ref:
         raise ValueError("browser evidence packet missing target_ref")
-    after = packet.get("after") if isinstance(packet.get("after"), Mapping) else {}
-    verification = packet.get("verification") if isinstance(packet.get("verification"), Mapping) else {}
+    raw_after = packet.get("after")
+    after: Mapping = raw_after if isinstance(raw_after, Mapping) else {}
+    raw_verification = packet.get("verification")
+    verification: Mapping = raw_verification if isinstance(raw_verification, Mapping) else {}
     title = str(after.get("title") or after.get("url") or target_ref)
     text = _summary(packet, after, verification)
     meta = {

src/gather/cli.py

@@ -14,6 +14,7 @@
     cmd_parse,
     cmd_pdf,
     cmd_run,
+    cmd_scholar,
     cmd_transcribe,
     cmd_video,
     cmd_web,
@@ -118,6 +119,19 @@ def build_parser() -> argparse.ArgumentParser:
     _add_common(arxiv)
     arxiv.set_defaults(func=cmd_arxiv)
 
+    scholar = sub.add_parser(
+        "scholar",
+        help="federate OpenAlex + Semantic Scholar + Crossref, dedup by DOI, capture citation edges")
+    scholar.add_argument("query", help="a DOI (10.1234/xyz) or a free-text query")
+    scholar.add_argument("--providers", default="openalex,semanticscholar,crossref",
+                         help="comma-separated subset of the three scholarly graphs")
+    scholar.add_argument("--no-federate", action="store_true",
+                         help="emit each provider's paper directly, undeduped (default: dedup by DOI)")
+    scholar.add_argument("--edges", action="store_true",
+                         help="also fold citation edges into the digest as first-class receipts")
+    _add_common(scholar)
+    scholar.set_defaults(func=cmd_scholar)
+
     pdf = sub.add_parser("pdf", help="extract text from a local PDF (needs pdftotext on PATH)")
     pdf.add_argument("path")
     _add_common(pdf)

src/gather/commands.py

@@ -94,6 +94,62 @@ def cmd_pdf(args) -> int:
     return _fetch_and_emit(lambda: PdfSource().fetch(args.path), args, fail="read failed")
 
 
+def cmd_scholar(args) -> int:
+    from gather.scholar import PROVIDERS, ScholarError, ScholarSource
+
+    providers = tuple(_split(args.providers))
+    try:
+        src = ScholarSource(providers=providers or PROVIDERS, federated=not args.no_federate)
+    except ScholarError as exc:
+        print(f"scholar failed: {exc}", file=sys.stderr)
+        return 1
+    if not args.edges:
+        return _fetch_and_emit(lambda: src.fetch(args.query), args, fail="scholar failed")
+    try:
+        items, edges = src.graph(args.query)
+    except Exception as exc:
+        print(f"scholar failed: {exc}", file=sys.stderr)
+        return 1
+    return _emit_with_edges(items, edges, _scope(args), args.json, store=args.store)
+
+
+def _emit_with_edges(items, edges, scope, as_json, store=None) -> int:
+    """Emit scholar items plus citation-edge receipts, folding the edges into the same witnessed
+    digest as the papers, so the seal covers the graph (nodes and links), not just the nodes."""
+    from gather.digest import digest_of_receipts, verify_digest
+    from gather.scope import filter_scope
+    from gather.store import Corpus
+
+    kept, dropped = filter_scope(items, scope)
+    item_receipts = [
+        {"kind": i.kind, "id": i.id, "title": i.title, "source": i.provenance.source,
+         "ref": i.provenance.ref, "method": i.provenance.method, "sha256": i.provenance.sha256,
+         "derived_from": list(i.provenance.derived_from)}
+        for i in kept
+    ]
+    d = digest_of_receipts(item_receipts + edges)
+    stored = None
+    if store:
+        stored = Corpus(store).add(kept)
+    if as_json:
+        out = {
+            "papers": item_receipts, "edges": edges, "dropped": dropped,
+            "digest": {"seal": d.seal, "verified": verify_digest(d)},
+        }
+        if stored is not None:
+            out["stored"] = stored
+        print(json.dumps(out, indent=2, ensure_ascii=False))
+        return 0
+    print(f"gathered {len(kept)} paper(s), {len(edges)} citation edge(s)"
+          f"{f', dropped {dropped} out of scope' if scope else ''}")
+    for i in kept:
+        print(f"  {i.kind:<14} {i.id:<28} {i.title[:44]}")
+    print(f"digest seal: {d.seal[:16]}... | verified: {verify_digest(d)} | covers papers + edges")
+    if stored is not None:
+        print(f"stored to {store}: {stored['added']} added, {stored['deduped']} deduped")
+    return 0
+
+
 def cmd_api(args) -> int:
     from gather.api import ApiSource
     return _fetch_and_emit(

src/gather/method.py

@@ -14,6 +14,9 @@
     "yt-dlp", "auto-caption", "http-get", "file-read", "feed",
     "arxiv-api", "arxiv-api-id", "arxiv-api-search", "pdftotext",
     "api-get", "ocr", "transcribe", "browser-extract", "browser-evidence",
+    # scholarly-graph federation: each provider's paper fetch, and a citation edge (the provider
+    # asserted the link; gather records that it did). All DIRECT. See gather.scholar.
+    "openalex-api", "semanticscholar-api", "crossref-api", "citation-edge",
 })
 DERIVED_METHODS = frozenset({"compiled", "synthesized"})