init

.gitattributes

@@ -0,0 +1,35 @@
+*.7z filter=lfs diff=lfs merge=lfs -text
+*.arrow filter=lfs diff=lfs merge=lfs -text
+*.bin filter=lfs diff=lfs merge=lfs -text
+*.bz2 filter=lfs diff=lfs merge=lfs -text
+*.ckpt filter=lfs diff=lfs merge=lfs -text
+*.ftz filter=lfs diff=lfs merge=lfs -text
+*.gz filter=lfs diff=lfs merge=lfs -text
+*.h5 filter=lfs diff=lfs merge=lfs -text
+*.joblib filter=lfs diff=lfs merge=lfs -text
+*.lfs.* filter=lfs diff=lfs merge=lfs -text
+*.mlmodel filter=lfs diff=lfs merge=lfs -text
+*.model filter=lfs diff=lfs merge=lfs -text
+*.msgpack filter=lfs diff=lfs merge=lfs -text
+*.npy filter=lfs diff=lfs merge=lfs -text
+*.npz filter=lfs diff=lfs merge=lfs -text
+*.onnx filter=lfs diff=lfs merge=lfs -text
+*.ot filter=lfs diff=lfs merge=lfs -text
+*.parquet filter=lfs diff=lfs merge=lfs -text
+*.pb filter=lfs diff=lfs merge=lfs -text
+*.pickle filter=lfs diff=lfs merge=lfs -text
+*.pkl filter=lfs diff=lfs merge=lfs -text
+*.pt filter=lfs diff=lfs merge=lfs -text
+*.pth filter=lfs diff=lfs merge=lfs -text
+*.rar filter=lfs diff=lfs merge=lfs -text
+*.safetensors filter=lfs diff=lfs merge=lfs -text
+saved_model/**/* filter=lfs diff=lfs merge=lfs -text
+*.tar.* filter=lfs diff=lfs merge=lfs -text
+*.tar filter=lfs diff=lfs merge=lfs -text
+*.tflite filter=lfs diff=lfs merge=lfs -text
+*.tgz filter=lfs diff=lfs merge=lfs -text
+*.wasm filter=lfs diff=lfs merge=lfs -text
+*.xz filter=lfs diff=lfs merge=lfs -text
+*.zip filter=lfs diff=lfs merge=lfs -text
+*.zst filter=lfs diff=lfs merge=lfs -text
+*tfevents* filter=lfs diff=lfs merge=lfs -text

README.md

@@ -0,0 +1,53 @@
+---
+title: Web Graph kNN Spot-check
+emoji: 🔎
+colorFrom: indigo
+colorTo: gray
+sdk: gradio
+sdk_version: 6.26.0
+python_version: 3.12.12
+app_file: app.py
+pinned: false
+short_description: Nearest neighbours of a host in the CC host-graph embedding
+---
+# Web graph kNN spot-check
+
+Type a hostname, get its nearest neighbours in the learned 128-d Common Crawl host-graph
+embedding. Companion to the 2D viewer: that one shows *where* hosts land after a UMAP
+projection, this one answers *who is actually nearest* in the real space.
+
+## Running locally
+
+```bash
+uv sync --extra faiss --extra viewer
+wgl knn build --hf-repo commoncrawl/web-graph-embeddings --shards 8 --row-groups 1 \
+    --out data/knn/cc_deg8_poc
+WGL_KNN_INDEX=data/knn/cc_deg8_poc uv run python knn-viewer/app.py
+```
+
+## Running as a Space
+
+The app reads its index from `$WGL_KNN_INDEX`, defaulting to `/data` — the mount path for a
+HuggingFace volume. Point a **private** bucket at it:
+
+```bash
+hf buckets create commoncrawl/web-graph-knn-index --private
+hf sync data/knn/cc_deg8_full hf://buckets/commoncrawl/web-graph-knn-index/cc_deg8 --delete
+hf spaces volumes set commoncrawl/web-graph-knn \
+    -v hf://buckets/commoncrawl/web-graph-knn-index:/data
+```
+
+`app.py` needs `wgl.knn.index`, which pulls only numpy, faiss and tldextract, so `wgl` is
+**vendored** next to `app.py` on the Space rather than installed. Only the *import closure* of
+`app.py` ships — 13 modules of 100, so the Space holds the code it runs and not the training
+tree. `KnnIndex` stays a `wgl` module rather than being copied into the app on purpose: it reads
+the artifact format `wgl.knn.build` writes, `wgl knn query` reads the same one, and
+`tests/test_knn_index.py` round-trips build against load to keep them honest. Deploying does the
+vendoring for you:
+
+```bash
+make knn-diff     # which files the Space is missing or has stale (changes nothing)
+make knn-deploy   # upload the app + vendored src/wgl, skipping what the Space already has
+make knn-serve    # run it locally against $KNN_INDEX first
+```
+

app.py

@@ -0,0 +1,337 @@
+"""Gradio UI for the kNN spot-check viewer.
+
+Type a hostname, get its nearest neighbours in the learned 128-d embedding space. Companion to the
+2D viewer (``embedding-2d-viewer/``): that one shows *where* hosts land after a UMAP projection,
+this one answers *who is actually nearest* in the real space.
+
+The point of the tool is **walking** the space, not one lookup: click any neighbour to make it the
+new query, with a back button and a breadcrumb trail. That turns "are these neighbours sensible?"
+into "does the neighbourhood stay coherent two or three hops out?", which is the question that
+actually catches a bad embedding.
+
+Runs two ways, unchanged:
+
+* **locally** — ``WGL_KNN_INDEX=data/knn/cc_deg8_poc python knn-viewer/app.py``;
+* **as a HF Space** — the index arrives on a mounted volume at ``/data``, which is the default.
+
+The index loads once at module scope, not per request: it is the whole cost of a query, and a
+Space re-imports this module only on cold start.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+import gradio as gr
+
+# Allow running straight from a repo checkout without installing the package.
+_REPO_SRC = Path(__file__).resolve().parent.parent / "src"
+if _REPO_SRC.is_dir() and str(_REPO_SRC) not in sys.path:
+    sys.path.insert(0, str(_REPO_SRC))
+
+from wgl.knn.index import KnnIndex  # noqa: E402
+from wgl.release.identity import domain_key as to_domain_key  # noqa: E402
+
+INDEX_PATH = Path(os.environ.get("WGL_KNN_INDEX", "/data"))
+EXAMPLES = ["www.bundestag.de", "www.bild.de", "https://www.bahn.de/service", "portal.gov.cz"]
+HEADERS = ["#", "host", "cosine", "registrable domain", "same site", "open"]
+DATATYPES = ["number", "str", "number", "str", "str", "markdown"]
+
+INDEX = KnnIndex.load(INDEX_PATH)
+_META = INDEX.meta
+_FOOTER = (
+    f"**{len(INDEX):,} hosts** · `{_META.get('release_id', 'unknown release')}` · "
+    f"model `{_META.get('model_space_version', 'unknown')}` · "
+    f"index `{_META.get('index_factory', '?')}` · dim {_META.get('dim', '?')} · "
+    f"built {str(_META.get('built_at', '?'))[:19]}"
+)
+_SUBSET_NOTE = ""
+if _META.get("source_row_groups_per_shard") is not None:
+    _SUBSET_NOTE = (
+        f"\n\n⚠️ **This is a proof-of-concept subset**, not the full corpus: source rows "
+        f"{_META.get('source_row_min'):,} to {_META.get('source_row_max'):,} of 52,913,544. "
+        "Hosts outside that slice will read as *not in the corpus*."
+    )
+
+
+# ── Common Crawl look ─────────────────────────────────────────────────────────────────────────
+# Same palette as the release figures (``src/wgl/viz/svg.py``) and the 2D viewer: brand blue
+# #0050bd, navy ink #152a47, cool neutral greys, Libre Franklin + IBM Plex Mono. Gradio derives a
+# lot of shades from its hue ramps, so both ramps are spelled out and the tokens that matter are
+# then pinned explicitly; every colour has a ``_dark`` twin because the Space follows the browser.
+_CC_BLUE = gr.themes.Color(
+    c50="#e8f0fb",
+    c100="#cfe0f6",
+    c200="#9dc0ee",
+    c300="#6ba1e5",
+    c400="#4a90e8",
+    c500="#0050bd",
+    c600="#0047a8",
+    c700="#003d92",
+    c800="#00306f",
+    c900="#152a47",
+    c950="#0d1929",
+)
+_CC_SLATE = gr.themes.Color(
+    c50="#f6f7fa",
+    c100="#eef1f6",
+    c200="#e8ecf2",
+    c300="#dde2e8",
+    c400="#8d99b0",
+    c500="#6e7191",
+    c600="#44536f",
+    c700="#2a456d",
+    c800="#1d3557",
+    c900="#152a47",
+    c950="#0d1929",
+)
+THEME = gr.themes.Base(
+    primary_hue=_CC_BLUE,
+    secondary_hue=_CC_BLUE,
+    neutral_hue=_CC_SLATE,
+    font=(
+        gr.themes.GoogleFont("Libre Franklin"),
+        "system-ui",
+        "-apple-system",
+        "Segoe UI",
+        "sans-serif",
+    ),
+    font_mono=(
+        gr.themes.GoogleFont("IBM Plex Mono"),
+        "ui-monospace",
+        "SFMono-Regular",
+        "Menlo",
+        "monospace",
+    ),
+).set(
+    # page + panels: surface over a recessed plane, inverted in dark
+    body_background_fill="#f6f7fa",
+    body_background_fill_dark="#0d1929",
+    body_text_color="#152a47",
+    body_text_color_dark="#ffffff",
+    body_text_color_subdued="#6e7191",
+    body_text_color_subdued_dark="#8d99b0",
+    background_fill_primary="#ffffff",
+    background_fill_primary_dark="#152a47",
+    background_fill_secondary="#f6f7fa",
+    background_fill_secondary_dark="#0d1929",
+    block_background_fill="#ffffff",
+    block_background_fill_dark="#152a47",
+    block_border_color="#dde2e8",
+    block_border_color_dark="#2a456d",
+    block_label_text_color="#44536f",
+    block_label_text_color_dark="#c3ccdb",
+    block_title_text_color="#152a47",
+    block_title_text_color_dark="#ffffff",
+    border_color_primary="#dde2e8",
+    border_color_primary_dark="#2a456d",
+    border_color_accent="#0050bd",
+    border_color_accent_dark="#4a90e8",
+    code_background_fill="#f6f7fa",
+    code_background_fill_dark="#0d1929",
+    # accents: brand blue in light, the lighter dark-theme blue on navy
+    color_accent="#0050bd",
+    link_text_color="#0050bd",
+    link_text_color_dark="#4a90e8",
+    link_text_color_hover="#0047a8",
+    link_text_color_hover_dark="#2895f7",
+    link_text_color_active="#0047a8",
+    link_text_color_active_dark="#2895f7",
+    link_text_color_visited="#6631ff",
+    link_text_color_visited_dark="#9353ff",
+    loader_color="#0050bd",
+    loader_color_dark="#4a90e8",
+    slider_color="#0050bd",
+    slider_color_dark="#4a90e8",
+    # inputs
+    input_background_fill="#ffffff",
+    input_background_fill_dark="#0d1929",
+    input_border_color="#dde2e8",
+    input_border_color_dark="#2a456d",
+    input_border_color_focus="#0050bd",
+    input_border_color_focus_dark="#4a90e8",
+    input_placeholder_color="#6e7191",
+    input_placeholder_color_dark="#8d99b0",
+    # buttons
+    button_primary_background_fill="#0050bd",
+    button_primary_background_fill_hover="#0047a8",
+    button_primary_background_fill_dark="#0050bd",
+    button_primary_background_fill_hover_dark="#4a90e8",
+    button_primary_border_color="#0050bd",
+    button_primary_border_color_dark="#0050bd",
+    button_primary_text_color="#ffffff",
+    button_primary_text_color_dark="#ffffff",
+    button_secondary_background_fill="#f6f7fa",
+    button_secondary_background_fill_hover="#e8ecf2",
+    button_secondary_background_fill_dark="#152a47",
+    button_secondary_background_fill_hover_dark="#1d3557",
+    button_secondary_border_color="#dde2e8",
+    button_secondary_border_color_dark="#2a456d",
+    button_secondary_text_color="#152a47",
+    button_secondary_text_color_dark="#ffffff",
+    # the neighbour table, the one dense surface in the app
+    table_border_color="#dde2e8",
+    table_border_color_dark="#2a456d",
+    table_even_background_fill="#f6f7fa",
+    table_even_background_fill_dark="#0d1929",
+    table_odd_background_fill="#ffffff",
+    table_odd_background_fill_dark="#152a47",
+    table_row_focus="#e8ecf2",
+    table_row_focus_dark="#1d3557",
+    table_text_color="#152a47",
+    table_text_color_dark="#ffffff",
+    # status colours, shared by both themes
+    error_text_color="#d03b3b",
+    error_text_color_dark="#d03b3b",
+)
+
+# The few things the theme tokens do not reach: the brand rule under the title, the mono trail and
+# the small print in the footer. Dark values ride on Gradio's own ``.dark`` class on the wrapper.
+CSS = """
+.gradio-container { max-width: 1180px !important; }
+#title h1 { margin-bottom: .35rem; padding-bottom: .35rem; letter-spacing: -.01em;
+  border-bottom: 3px solid #0050bd; }
+.dark #title h1 { border-bottom-color: #4a90e8; }
+#crumb, #footer { font-size: .82rem; color: #6e7191; }
+.dark #crumb, .dark #footer { color: #8d99b0; }
+#crumb code, #footer code { background: transparent; padding: 0; color: #44536f; }
+.dark #crumb code, .dark #footer code { color: #c3ccdb; }
+"""
+
+
+def _table(result) -> list[list[object]]:  # noqa: ANN001 - QueryResult, kept loose for gradio
+    """Render a query result as dataframe rows, with the free per-neighbour metadata."""
+    query_domain = to_domain_key(result.query, reversed_host=False)
+    rows: list[list[object]] = []
+    for neighbor in result.neighbors:
+        domain = to_domain_key(neighbor.host, reversed_host=False)
+        rows.append(
+            [
+                neighbor.rank,
+                neighbor.host,
+                round(neighbor.score, 4),
+                domain,
+                "✓" if domain and domain == query_domain else "",
+                f"[↗](https://{neighbor.host})",
+            ]
+        )
+    return rows
+
+
+def _breadcrumb(trail: list[str]) -> str:
+    """Render the navigation trail."""
+    if len(trail) < 2:
+        return ""
+    shown = trail[-6:]
+    prefix = "… → " if len(trail) > len(shown) else ""
+    return "**Trail:** " + prefix + " → ".join(f"`{h}`" for h in shown)
+
+
+def search(host: str, k: int, trail: list[str]) -> tuple:
+    """Run a lookup and return every output the UI needs.
+
+    Args:
+        host: Raw user input — a bare host, a domain, or a pasted URL.
+        k: Number of neighbours to return.
+        trail: The navigation history so far.
+
+    Returns:
+        ``(status, rows, neighbour_hosts, trail, breadcrumb, host_box)`` — the extra state lets a
+        row click resolve back to a hostname, and keeps the textbox in sync while walking.
+    """
+    trail = list(trail or [])
+    if not host or not host.strip():
+        return "Type a hostname to begin.", [], [], trail, _breadcrumb(trail), host
+
+    result = INDEX.query(host, k=int(k))
+    if not result.found:
+        return (
+            f"❌ **`{result.query or host.strip()}` is not in the indexed corpus.**\n\n"
+            "That is a normal answer, not an error — the index covers a degree-filtered "
+            "subset of the Common Crawl host graph. Tried the exact host, its `www.` variant "
+            "and its registrable domain.",
+            [],
+            [],
+            trail,
+            _breadcrumb(trail),
+            host,
+        )
+
+    if not trail or trail[-1] != result.query:
+        trail.append(result.query)
+    via = {
+        "exact": "exact match",
+        "www": "matched the `www.` variant",
+        "no-www": "matched without the `www.` prefix",
+        "domain": "fell back to the registrable domain",
+    }.get(result.resolved_via, result.resolved_via)
+    status = (
+        f"✅ **`{result.query}`** — {via}, row {result.row:,}. "
+        "*Click any row to explore that host's neighbours.*"
+    )
+    hosts = [n.host for n in result.neighbors]
+    return status, _table(result), hosts, trail, _breadcrumb(trail), result.query
+
+
+def follow(evt: gr.SelectData, hosts: list[str], k: int, trail: list[str]) -> tuple:
+    """Make the clicked neighbour the new query."""
+    row = evt.index[0] if isinstance(evt.index, (list, tuple)) else evt.index
+    if not hosts or row is None or row >= len(hosts):
+        return gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip()
+    return search(hosts[row], k, trail)
+
+
+def go_back(k: int, trail: list[str]) -> tuple:
+    """Step one hop back along the trail."""
+    trail = list(trail or [])
+    if len(trail) < 2:
+        return gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip(), gr.skip()
+    trail.pop()
+    previous = trail.pop()
+    return search(previous, k, trail)
+
+
+with gr.Blocks(title="Web graph kNN spot-check", analytics_enabled=False) as demo:
+    neighbor_state = gr.State([])
+    trail_state = gr.State([])
+
+    gr.Markdown("# Web graph kNN spot-check", elem_id="title")
+    gr.Markdown(
+        "Nearest neighbours of a host in the learned link-prediction embedding space "
+        "(cosine similarity on L2-normalized vectors). **Click a neighbour to walk to it.**"
+        + _SUBSET_NOTE
+    )
+    with gr.Row():
+        host_box = gr.Textbox(
+            label="Host or URL", placeholder="www.example.com", scale=4, autofocus=True
+        )
+        k_slider = gr.Slider(minimum=5, maximum=100, value=20, step=5, label="Neighbours", scale=1)
+    with gr.Row():
+        go = gr.Button("Search", variant="primary", scale=3)
+        back = gr.Button("← Back", scale=1)
+    status = gr.Markdown()
+    crumb = gr.Markdown(elem_id="crumb")
+    table = gr.Dataframe(
+        headers=HEADERS,
+        datatype=DATATYPES,
+        label="Nearest neighbours — click a row to explore it",
+        wrap=True,
+        interactive=False,
+    )
+    gr.Examples(examples=[[e] for e in EXAMPLES], inputs=[host_box])
+    gr.Markdown(_FOOTER, elem_id="footer")
+
+    OUTPUTS = [status, table, neighbor_state, trail_state, crumb, host_box]
+    go.click(search, [host_box, k_slider, trail_state], OUTPUTS)
+    host_box.submit(search, [host_box, k_slider, trail_state], OUTPUTS)
+    k_slider.release(search, [host_box, k_slider, trail_state], OUTPUTS)
+    table.select(follow, [neighbor_state, k_slider, trail_state], OUTPUTS)
+    back.click(go_back, [k_slider, trail_state], OUTPUTS)
+
+
+if __name__ == "__main__":
+    # Gradio 6 takes ``theme``/``css`` on launch(), not on the Blocks constructor (it warns and
+    # drops them there). The Space runs this file as its app_file, so this is the one entry point.
+    demo.launch(theme=THEME, css=CSS)

requirements.txt

@@ -0,0 +1,8 @@
+# The Space vendors src/wgl next to app.py, so only wgl.knn's own runtime deps are needed:
+# numpy + faiss for the index, tldextract for host_key/domain_key, rich for wgl.logging.
+# Versions are floors, not pins: the Space image resolves them for its own Python.
+gradio==6.26.0
+faiss-cpu>=1.14
+numpy>=2.0,<3
+tldextract>=5.1
+rich>=13.7

wgl/__init__.py

@@ -0,0 +1,3 @@
+"""web-graph-learning: directed Web graph representation learning at scale."""
+
+__version__ = "0.1.0"

wgl/data/__init__.py

@@ -0,0 +1 @@
+"""Graph data structures, loaders, and leakage-safe splitting."""

wgl/data/graph.py

@@ -0,0 +1,217 @@
+"""Directed graph stored as out-edge and in-edge CSR structures.
+
+Web graphs are fundamentally directed and the in/out asymmetry is the key signal for tasks like
+spam detection (see ``docs/brainstorming/edge-direction.md``). We therefore keep two CSR views and
+never symmetrize implicitly:
+
+* **out-CSR** (``out_indptr``, ``out_indices``): successors of each node (who it links to).
+* **in-CSR**  (``in_indptr``,  ``in_indices``):  predecessors of each node (who links to it).
+
+The COO ``edge_index`` of shape ``(2, num_edges)`` (rows: ``src``, ``dst``) is the canonical edge
+list. Arrays are plain NumPy so they save/load with ``np.savez`` and can be memory-mapped at scale.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+from numpy.typing import NDArray
+
+
+def _build_csr(
+    rows: NDArray[np.int64],
+    cols: NDArray[np.int64],
+    num_nodes: int,
+    index_dtype: np.dtype = np.int64,
+) -> tuple[NDArray[np.int64], NDArray[np.int64]]:
+    """Build a CSR ``(indptr, indices)`` grouping ``cols`` by ``rows``.
+
+    Args:
+        rows: Source endpoints (the CSR row of each entry).
+        cols: Destination endpoints (the value stored in each row).
+        num_nodes: Total number of nodes (CSR has ``num_nodes + 1`` indptr entries).
+        index_dtype: dtype of the returned ``indices`` (the node-id column). ``int32`` halves the
+            footprint when node ids fit (``num_nodes < 2**31``) — worth it at 10^10-edge scale.
+            ``indptr`` is always ``int64`` (it accumulates to ``num_edges``, which can exceed 2^31).
+
+    Returns:
+        A tuple ``(indptr, indices)``.
+    """
+    order = np.argsort(rows, kind="stable")
+    sorted_rows = rows[order]
+    sorted_cols = cols[order]
+    counts = np.bincount(sorted_rows, minlength=num_nodes)
+    indptr = np.zeros(num_nodes + 1, dtype=np.int64)
+    np.cumsum(counts, out=indptr[1:])
+    return indptr, sorted_cols.astype(index_dtype, copy=False)
+
+
+@dataclass
+class DirectedGraph:
+    """An immutable directed graph with cached out/in CSR adjacency.
+
+    Attributes:
+        num_nodes: Number of nodes.
+        edge_index: COO edges of shape ``(2, num_edges)`` (row 0 = src, row 1 = dst).
+        out_indptr: Out-CSR row pointers, shape ``(num_nodes + 1,)``.
+        out_indices: Out-CSR successors.
+        in_indptr: In-CSR row pointers, shape ``(num_nodes + 1,)``.
+        in_indices: In-CSR predecessors.
+        node_names: Optional per-node string identifiers (e.g. reversed hostnames).
+        features: Optional per-node feature matrix, shape ``(num_nodes, feature_dim)`` (e.g.
+            ogbn-arxiv's word2vec vectors). Persisted as ``node_features.npy``; ``None`` for
+            structure-only graphs (the CC web graph has no such features).
+    """
+
+    num_nodes: int
+    edge_index: NDArray[np.int64]
+    out_indptr: NDArray[np.int64]
+    out_indices: NDArray[np.int64]
+    in_indptr: NDArray[np.int64]
+    in_indices: NDArray[np.int64]
+    node_names: list[str] | None = None
+    features: NDArray[np.float32] | None = None
+
+    @property
+    def num_edges(self) -> int:
+        """Number of directed edges."""
+        return int(self.edge_index.shape[1])
+
+    @classmethod
+    def from_edges(
+        cls,
+        edge_index: NDArray[np.int64],
+        num_nodes: int | None = None,
+        node_names: list[str] | None = None,
+    ) -> DirectedGraph:
+        """Construct a graph from a COO edge list, building both CSR views.
+
+        Args:
+            edge_index: Array of shape ``(2, num_edges)`` with src in row 0 and dst in row 1.
+            num_nodes: Node count; inferred from the max id + 1 when omitted.
+            node_names: Optional per-node identifiers.
+
+        Returns:
+            A new :class:`DirectedGraph`.
+        """
+        edge_index = np.ascontiguousarray(edge_index, dtype=np.int64)
+        if edge_index.ndim != 2 or edge_index.shape[0] != 2:
+            msg = f"edge_index must have shape (2, E), got {edge_index.shape}"
+            raise ValueError(msg)
+        src, dst = edge_index[0], edge_index[1]
+        if num_nodes is None:
+            num_nodes = int(edge_index.max()) + 1 if edge_index.size else 0
+        out_indptr, out_indices = _build_csr(src, dst, num_nodes)
+        in_indptr, in_indices = _build_csr(dst, src, num_nodes)
+        return cls(
+            num_nodes=num_nodes,
+            edge_index=edge_index,
+            out_indptr=out_indptr,
+            out_indices=out_indices,
+            in_indptr=in_indptr,
+            in_indices=in_indices,
+            node_names=node_names,
+        )
+
+    def out_neighbors(self, node: int) -> NDArray[np.int64]:
+        """Return the successors of ``node`` (nodes it links to)."""
+        return self.out_indices[self.out_indptr[node] : self.out_indptr[node + 1]]
+
+    def in_neighbors(self, node: int) -> NDArray[np.int64]:
+        """Return the predecessors of ``node`` (nodes that link to it)."""
+        return self.in_indices[self.in_indptr[node] : self.in_indptr[node + 1]]
+
+    def out_degree(self) -> NDArray[np.int64]:
+        """Return the out-degree of every node."""
+        return np.diff(self.out_indptr)
+
+    def in_degree(self) -> NDArray[np.int64]:
+        """Return the in-degree of every node."""
+        return np.diff(self.in_indptr)
+
+    def to_undirected(self) -> DirectedGraph:
+        """Return a symmetrized copy with both ``(u, v)`` and ``(v, u)`` for every edge.
+
+        Used only to build the *undirected baseline* for direction ablations — it deliberately
+        discards the in/out asymmetry that the directed models exploit, so it should never be used
+        for the primary directed experiments.
+
+        Returns:
+            A new :class:`DirectedGraph` over the same nodes with reciprocated, de-duplicated edges.
+        """
+        both = np.concatenate([self.edge_index, self.edge_index[::-1]], axis=1)
+        undirected = np.unique(both, axis=1) if both.size else both
+        return DirectedGraph.from_edges(
+            undirected, num_nodes=self.num_nodes, node_names=self.node_names
+        )
+
+    def subgraph_from_edge_mask(self, mask: NDArray[np.bool_]) -> DirectedGraph:
+        """Return a graph over the same node set keeping only the masked edges.
+
+        Used to build the *message* graph for link prediction (a structural view that excludes
+        held-out evaluation edges).
+
+        Args:
+            mask: Boolean mask over edges of length ``num_edges``.
+
+        Returns:
+            A new :class:`DirectedGraph` with the same ``num_nodes`` and node names.
+        """
+        return DirectedGraph.from_edges(
+            self.edge_index[:, mask], num_nodes=self.num_nodes, node_names=self.node_names
+        )
+
+    def save(self, directory: str | Path) -> None:
+        """Persist the graph to ``directory`` (``graph.npz`` + ``meta.json``).
+
+        Args:
+            directory: Output directory (created if needed).
+        """
+        directory = Path(directory)
+        directory.mkdir(parents=True, exist_ok=True)
+        np.savez(
+            directory / "graph.npz",
+            edge_index=self.edge_index,
+            out_indptr=self.out_indptr,
+            out_indices=self.out_indices,
+            in_indptr=self.in_indptr,
+            in_indices=self.in_indices,
+        )
+        meta = {"num_nodes": self.num_nodes, "num_edges": self.num_edges}
+        (directory / "meta.json").write_text(json.dumps(meta, indent=2))
+        if self.node_names is not None:
+            (directory / "node_names.txt").write_text("\n".join(self.node_names))
+        if self.features is not None:
+            np.save(directory / "node_features.npy", self.features)
+
+    @classmethod
+    def load(cls, directory: str | Path, mmap: bool = False) -> DirectedGraph:
+        """Load a graph previously written by :meth:`save`.
+
+        Args:
+            directory: Directory containing ``graph.npz`` / ``meta.json``.
+            mmap: If True, memory-map the arrays (recommended for large graphs).
+
+        Returns:
+            The loaded :class:`DirectedGraph`.
+        """
+        directory = Path(directory)
+        data = np.load(directory / "graph.npz", mmap_mode="r" if mmap else None)
+        meta = json.loads((directory / "meta.json").read_text())
+        names_path = directory / "node_names.txt"
+        node_names = names_path.read_text().splitlines() if names_path.exists() else None
+        features_path = directory / "node_features.npy"
+        features = np.load(features_path) if features_path.exists() else None
+        return cls(
+            num_nodes=int(meta["num_nodes"]),
+            edge_index=data["edge_index"],
+            out_indptr=data["out_indptr"],
+            out_indices=data["out_indices"],
+            in_indptr=data["in_indptr"],
+            in_indices=data["in_indices"],
+            node_names=node_names,
+            features=features,
+        )

wgl/knn/__init__.py

@@ -0,0 +1,27 @@
+"""kNN spot-check index over a released host-embedding table.
+
+A small debugging tool: given a hostname, return its nearest neighbours in the learned 128-d
+embedding space. Companion to the 2D viewer (``docs/embedding-2d-viewer.md``) — that one shows
+*where* hosts land after a UMAP projection, this one answers *who is actually nearest* in the real
+space. See [`docs/knn-viewer/plan.md`](../../../docs/knn-viewer/plan.md).
+
+The artifact is a flat FAISS ``IndexPQ`` plus a sorted hash → row map and an offset-indexed name
+file, so a query is a hash lookup followed by one exhaustive (but tiny) PQ scan. There is no IVF
+and no ``nprobe``: at PQ M=32 a full scan of 52.9 M rows costs well under a second, so the only
+approximation in the tool is the PQ codebook itself.
+"""
+
+from wgl.knn.build import BuildResult, build_index
+from wgl.knn.index import KnnIndex, Neighbor
+from wgl.knn.source import HostVectorSource, NpySource, ParquetSource, RowBatch
+
+__all__ = [
+    "BuildResult",
+    "HostVectorSource",
+    "KnnIndex",
+    "Neighbor",
+    "NpySource",
+    "ParquetSource",
+    "RowBatch",
+    "build_index",
+]

wgl/knn/build.py

@@ -0,0 +1,260 @@
+r"""Build the kNN spot-check index artifact from a host-vector source.
+
+The artifact is deliberately five small files plus a manifest, all in one directory:
+
+===============  ==========================================  ===========================
+file             dtype / format                              contents
+===============  ==========================================  ===========================
+``index.faiss``  FAISS ``OPQ<M>,PQ<M>`` (or ``Flat``)        one code per row + codebooks
+``keys.npy``     ``uint64[n]``, **sorted ascending**          hash of each host key
+``rows.npy``     ``uint32[n]``, permuted with ``keys``        index row for each key
+``names.txt``    UTF-8, ``\n``-joined                         forward host per index row
+``names.idx``    ``uint64[n + 1]``                            byte offset of each line
+``meta.json``    JSON                                         provenance + invariants
+===============  ==========================================  ===========================
+
+Names live on disk behind an offset index rather than in RAM: a query needs about twenty of them,
+so it seeks instead of loading ~1 GB of strings. Keys are hashed to ``uint64`` and sorted so the
+host lookup is a :func:`numpy.searchsorted` over a 0.4 GB array instead of a ~6 GB Python dict; the
+hash is verified by reading the name back, which makes a collision a clean miss rather than a wrong
+answer.
+
+The builder makes a **single pass** over the source. A quantizer has to be trained before anything
+can be encoded, so early batches are buffered until the training sample is full, trained on, then
+flushed — which keeps a remote source from being downloaded twice and keeps peak memory at the
+training sample rather than the whole table.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass, field
+from datetime import UTC, datetime
+from pathlib import Path
+
+import numpy as np
+from numpy.typing import NDArray
+
+from wgl.knn.source import HostVectorSource
+from wgl.logging import get_logger
+
+# Below this row count there is not enough data to train a product quantizer (PQ needs 256 training
+# points per sub-quantizer to fill its codebook), so the builder stores exact vectors instead.
+MIN_ROWS_FOR_PQ = 10_000
+TRAIN_SAMPLE = 200_000
+
+
+@dataclass
+class BuildResult:
+    """Summary of a completed build.
+
+    Attributes:
+        out: Directory the artifact was written to.
+        n: Number of rows indexed.
+        dim: Embedding dimensionality.
+        factory: The FAISS index-factory string actually used.
+        duplicates: Host keys dropped because they appeared more than once.
+        bytes_written: Total artifact size on disk.
+        meta: The ``meta.json`` contents.
+    """
+
+    out: Path
+    n: int
+    dim: int
+    factory: str
+    duplicates: int
+    bytes_written: int
+    meta: dict[str, object] = field(default_factory=dict)
+
+
+def hash_keys(keys: list[str]) -> NDArray[np.uint64]:
+    """Hash host keys to ``uint64``.
+
+    Uses BLAKE2b truncated to 8 bytes — a stdlib hash, so the index builder adds no dependency for
+    it. Collisions are not prevented (about 7.6e-5 probability across 53 M keys by the birthday
+    approximation) but they are *detected*: :meth:`wgl.knn.index.KnnIndex.lookup` reads the stored
+    name back and rejects a mismatch.
+
+    Args:
+        keys: Forward host keys.
+
+    Returns:
+        One ``uint64`` per key, in input order.
+    """
+    out = np.empty(len(keys), dtype=np.uint64)
+    for i, key in enumerate(keys):
+        digest = hashlib.blake2b(key.encode("utf-8"), digest_size=8).digest()
+        out[i] = int.from_bytes(digest, "big")
+    return out
+
+
+def choose_factory(n: int, dim: int, requested: str = "auto") -> str:
+    """Pick a FAISS index-factory string.
+
+    Args:
+        n: Row count.
+        dim: Embedding dimensionality.
+        requested: An explicit factory string, or ``"auto"``.
+
+    Returns:
+        The factory string to build with. ``"auto"`` resolves to ``OPQ32,PQ32`` when there are
+        enough rows to train a quantizer and the dimensionality divides evenly, else ``"Flat"``.
+    """
+    if requested != "auto":
+        return requested
+    if n < MIN_ROWS_FOR_PQ or dim % 32 != 0:
+        return "Flat"
+    return "OPQ32,PQ32"
+
+
+def _normalize(vectors: NDArray[np.float32]) -> NDArray[np.float32]:
+    """L2-normalize rows so that an inner product equals cosine similarity."""
+    array = np.ascontiguousarray(vectors, dtype=np.float32)
+    norms = np.linalg.norm(array, axis=1, keepdims=True)
+    return array / np.maximum(norms, 1e-12)
+
+
+def build_index(
+    source: HostVectorSource,
+    out: Path,
+    *,
+    factory: str = "auto",
+    train_sample: int = TRAIN_SAMPLE,
+    extra_meta: dict[str, object] | None = None,
+) -> BuildResult:
+    """Build and write the kNN index artifact.
+
+    Vectors are L2-normalized *before* training and encoding, so the index metric is inner product
+    and a returned score is a cosine similarity. The source's own ``row_id`` is recorded in
+    ``meta.json`` but the index uses its own dense 0-based numbering, so a subset build is a
+    first-class citizen rather than a sparse index full of holes.
+
+    Args:
+        source: Where rows come from.
+        out: Output directory; created if absent.
+        factory: FAISS factory string, or ``"auto"`` (see :func:`choose_factory`).
+        train_sample: Rows to buffer and train the quantizer on.
+        extra_meta: Extra provenance fields merged into ``meta.json``.
+
+    Returns:
+        A :class:`BuildResult` describing what was written.
+
+    Raises:
+        ValueError: If the source yields no rows.
+    """
+    import faiss
+
+    logger = get_logger()
+    out = Path(out)
+    out.mkdir(parents=True, exist_ok=True)
+    dim = source.dim
+
+    index: object | None = None
+    resolved_factory: str | None = None
+    pending: list[NDArray[np.float32]] = []
+    pending_rows = 0
+    key_chunks: list[NDArray[np.uint64]] = []
+    length_chunks: list[NDArray[np.uint64]] = []
+    row_id_chunks: list[NDArray[np.int64]] = []
+    n = 0
+
+    def _flush(buffered: list[NDArray[np.float32]]) -> None:
+        """Train on the buffer if needed, then add every buffered batch to the index."""
+        nonlocal index, resolved_factory
+        stacked = np.concatenate(buffered, axis=0)
+        if index is None:
+            resolved_factory = choose_factory(max(stacked.shape[0], n), dim, factory)
+            logger.info("building %s index (dim=%d)", resolved_factory, dim)
+            index = faiss.index_factory(dim, resolved_factory, faiss.METRIC_INNER_PRODUCT)
+            if not index.is_trained:
+                # Cap the training set: a source that yields million-row batches would otherwise
+                # train on the whole first batch, which is far slower for no gain in codebook
+                # quality (FAISS wants ~256 points per centroid, not millions).
+                train_on = stacked[:train_sample]
+                logger.info("training quantizer on %d rows", train_on.shape[0])
+                index.train(train_on)
+        index.add(stacked)
+
+    names_path = out / "names.txt"
+    with names_path.open("wb") as handle:
+        for batch in source.batches():
+            encoded = [key.encode("utf-8") for key in batch.host_keys]
+            handle.write(b"\n".join(encoded) + b"\n")
+            length_chunks.append(np.fromiter((len(b) + 1 for b in encoded), dtype=np.uint64))
+            key_chunks.append(hash_keys(batch.host_keys))
+            row_id_chunks.append(batch.row_ids)
+            n += len(encoded)
+
+            pending.append(_normalize(batch.vectors))
+            pending_rows += batch.vectors.shape[0]
+            if pending_rows >= train_sample:
+                _flush(pending)
+                pending, pending_rows = [], 0
+            logger.info("scanned %d rows", n)
+
+    if n == 0:
+        msg = "source yielded no rows"
+        raise ValueError(msg)
+    if pending:
+        _flush(pending)
+    del pending
+
+    if index is None or index.ntotal != n:
+        held = 0 if index is None else index.ntotal
+        msg = f"index holds {held} vectors but {n} names were written"
+        raise ValueError(msg)
+
+    # Identity map: hash, sort, and drop any duplicate host key (the contract's uniqueness rule).
+    keys = np.concatenate(key_chunks)
+    order = np.argsort(keys, kind="stable")
+    keys_sorted = keys[order]
+    rows_sorted = order.astype(np.uint32)
+    unique = np.ones(keys_sorted.shape[0], dtype=bool)
+    if keys_sorted.shape[0] > 1:
+        unique[1:] = keys_sorted[1:] != keys_sorted[:-1]
+    duplicates = int((~unique).sum())
+    if duplicates:
+        logger.warning("dropped %d duplicate host key(s) from the lookup map", duplicates)
+        keys_sorted = keys_sorted[unique]
+        rows_sorted = rows_sorted[unique]
+
+    offsets = np.zeros(n + 1, dtype=np.uint64)
+    np.cumsum(np.concatenate(length_chunks), out=offsets[1:])
+
+    faiss.write_index(index, str(out / "index.faiss"))
+    np.save(out / "keys.npy", keys_sorted)
+    np.save(out / "rows.npy", rows_sorted)
+    with (out / "names.idx").open("wb") as handle:
+        np.save(handle, offsets)
+
+    source_rows = np.concatenate(row_id_chunks)
+    meta: dict[str, object] = {
+        "n": n,
+        "dim": dim,
+        "index_factory": resolved_factory,
+        "metric": "inner_product",
+        "normalized": True,
+        "hash": "blake2b-64",
+        "host_key_fn": "wgl.release.identity.host_key",
+        "duplicates_dropped": duplicates,
+        "source_row_min": int(source_rows.min()),
+        "source_row_max": int(source_rows.max()),
+        "built_at": datetime.now(UTC).isoformat(),
+        "faiss_version": getattr(faiss, "__version__", "unknown"),
+        **source.describe(),
+        **(extra_meta or {}),
+    }
+    (out / "meta.json").write_text(json.dumps(meta, indent=2) + "\n", encoding="utf-8")
+
+    written = sum(p.stat().st_size for p in out.iterdir() if p.is_file())
+    logger.info("wrote %s (%.1f MB)", out, written / 1e6)
+    return BuildResult(
+        out=out,
+        n=n,
+        dim=dim,
+        factory=str(meta["index_factory"]),
+        duplicates=duplicates,
+        bytes_written=written,
+        meta=meta,
+    )

wgl/knn/index.py

@@ -0,0 +1,260 @@
+"""Load a built kNN artifact and answer nearest-neighbour queries.
+
+A query is three cheap steps and one scan:
+
+1. normalize the typed host to a **host key** (:func:`wgl.release.identity.host_key`) and hash it;
+2. :func:`numpy.searchsorted` the sorted key array to get the index row, then **read the stored
+   name back** to confirm the hash actually belongs to that host — a collision becomes a clean miss
+   rather than a confidently wrong answer;
+3. reconstruct that row's vector from the index and scan for the top ``k + 1``, dropping the self
+   match.
+
+Everything is in-process: after :meth:`KnnIndex.load` returns there is no network access, so the
+same object serves a CLI query, a test, and a Gradio callback identically.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+
+import numpy as np
+from numpy.typing import NDArray
+
+from wgl.knn.build import hash_keys
+from wgl.release.identity import domain_key as to_domain_key
+from wgl.release.identity import host_key as to_host_key
+
+
+@dataclass(frozen=True)
+class Neighbor:
+    """One nearest-neighbour result.
+
+    Attributes:
+        rank: 1-based position in the result list.
+        host: The neighbour's forward host key.
+        score: Cosine similarity to the query (vectors are L2-normalized, so a dot product).
+        row: The neighbour's row in this index.
+    """
+
+    rank: int
+    host: str
+    score: float
+    row: int
+
+
+@dataclass(frozen=True)
+class QueryResult:
+    """The answer to one lookup.
+
+    Attributes:
+        query: The host key the query resolved to.
+        resolved_via: How it resolved — ``"exact"``, ``"www"``, ``"no-www"``, ``"domain"``, or
+            ``"miss"`` when the host is not in the corpus at all.
+        row: The matched index row, or ``None`` on a miss.
+        neighbors: The ranked neighbours; empty on a miss.
+    """
+
+    query: str
+    resolved_via: str
+    row: int | None
+    neighbors: list[Neighbor]
+
+    @property
+    def found(self) -> bool:
+        """Whether the query resolved to a row in the corpus."""
+        return self.row is not None
+
+
+def normalize_query(text: str) -> str:
+    """Turn user input into a host key.
+
+    Accepts a bare host, a host with a scheme and path, or a pasted URL. The result goes through
+    :func:`wgl.release.identity.host_key`, so it matches exactly what the builder stored.
+
+    Args:
+        text: Raw user input.
+
+    Returns:
+        The canonical forward host key, or an empty string when nothing usable was typed.
+    """
+    candidate = text.strip()
+    if not candidate:
+        return ""
+    if "//" in candidate:
+        candidate = candidate.split("//", 1)[1]
+    candidate = candidate.split("/", 1)[0].split("?", 1)[0]
+    candidate = candidate.rsplit("@", 1)[-1]
+    if candidate.startswith("[") and "]" in candidate:  # bracketed IPv6 literal
+        candidate = candidate[1 : candidate.index("]")]
+    elif candidate.count(":") == 1:
+        candidate = candidate.split(":", 1)[0]
+    return to_host_key(candidate, reversed_host=False)
+
+
+class KnnIndex:
+    """A loaded kNN artifact: the FAISS index, the key map, and the on-disk name file."""
+
+    def __init__(
+        self,
+        index: object,
+        keys: NDArray[np.uint64],
+        rows: NDArray[np.uint32],
+        offsets: NDArray[np.uint64],
+        names_path: Path,
+        meta: dict[str, object],
+    ) -> None:
+        """Wrap already-loaded artifact components. Use :meth:`load` instead."""
+        self._index = index
+        self._keys = keys
+        self._rows = rows
+        self._offsets = offsets
+        self._names_path = names_path
+        self.meta = meta
+
+    @classmethod
+    def load(cls, path: Path) -> KnnIndex:
+        """Load an artifact directory written by :func:`wgl.knn.build.build_index`.
+
+        Args:
+            path: The artifact directory.
+
+        Returns:
+            A ready-to-query index.
+
+        Raises:
+            FileNotFoundError: If the directory is missing an expected file.
+        """
+        import faiss
+
+        path = Path(path)
+        required = ["index.faiss", "keys.npy", "rows.npy", "names.txt", "names.idx", "meta.json"]
+        missing = [name for name in required if not (path / name).is_file()]
+        if missing:
+            msg = f"{path} is not a kNN artifact; missing {', '.join(missing)}"
+            raise FileNotFoundError(msg)
+        with (path / "names.idx").open("rb") as handle:
+            offsets = np.load(handle)
+        return cls(
+            index=faiss.read_index(str(path / "index.faiss")),
+            keys=np.load(path / "keys.npy"),
+            rows=np.load(path / "rows.npy"),
+            offsets=offsets,
+            names_path=path / "names.txt",
+            meta=json.loads((path / "meta.json").read_text(encoding="utf-8")),
+        )
+
+    def __len__(self) -> int:
+        """Return the number of indexed rows."""
+        return int(self._index.ntotal)
+
+    def name_of(self, row: int) -> str:
+        """Return the host key stored at ``row`` by seeking into ``names.txt``."""
+        start = int(self._offsets[row])
+        end = int(self._offsets[row + 1])
+        with self._names_path.open("rb") as handle:
+            handle.seek(start)
+            return handle.read(end - start).decode("utf-8", "replace").rstrip("\n")
+
+    def names_of(self, rows: list[int]) -> list[str]:
+        """Return host keys for several rows, reusing one open file handle."""
+        with self._names_path.open("rb") as handle:
+            out = []
+            for row in rows:
+                start = int(self._offsets[row])
+                handle.seek(start)
+                blob = handle.read(int(self._offsets[row + 1]) - start)
+                out.append(blob.decode("utf-8", "replace").rstrip("\n"))
+        return out
+
+    def lookup(self, host: str) -> int | None:
+        """Return the index row for an exact host key, or ``None``.
+
+        The stored name is read back and compared, so a hash collision returns ``None`` rather than
+        a neighbouring host's row.
+
+        Args:
+            host: A canonical forward host key.
+
+        Returns:
+            The row id, or ``None`` when the host is not indexed.
+        """
+        if not host:
+            return None
+        needle = hash_keys([host])[0]
+        position = int(np.searchsorted(self._keys, needle))
+        if position >= self._keys.shape[0] or self._keys[position] != needle:
+            return None
+        row = int(self._rows[position])
+        return row if self.name_of(row) == host else None
+
+    def resolve(self, text: str, *, domain_fallback: bool = True) -> tuple[str, str, int | None]:
+        """Resolve user input to an index row, trying sensible variants in order.
+
+        ``cc_deg8`` covers 52.9 M of a 279.4 M-host graph, so a miss is a normal answer rather than
+        an error. Before giving up this tries the ``www.`` variant in both directions and then the
+        registrable domain, which is very often the host that was actually crawled.
+
+        Args:
+            text: Raw user input.
+            domain_fallback: Whether to fall back to the registrable domain.
+
+        Returns:
+            A ``(host_key, resolved_via, row)`` triple; ``row`` is ``None`` on a miss.
+        """
+        host = normalize_query(text)
+        if not host:
+            return "", "miss", None
+
+        candidates: list[tuple[str, str]] = [(host, "exact")]
+        if host.startswith("www."):
+            candidates.append((host[4:], "no-www"))
+        else:
+            candidates.append((f"www.{host}", "www"))
+        if domain_fallback:
+            domain = to_domain_key(host, reversed_host=False)
+            if domain and domain != host:
+                candidates.append((domain, "domain"))
+                candidates.append((f"www.{domain}", "domain"))
+
+        seen: set[str] = set()
+        for candidate, via in candidates:
+            if candidate in seen:
+                continue
+            seen.add(candidate)
+            row = self.lookup(candidate)
+            if row is not None:
+                return candidate, via, row
+        return host, "miss", None
+
+    def query(self, text: str, k: int = 20, *, domain_fallback: bool = True) -> QueryResult:
+        """Return the ``k`` nearest neighbours of a typed host.
+
+        Args:
+            text: Raw user input — a host, a URL, or a host with a port.
+            k: Number of neighbours to return (the query itself is excluded).
+            domain_fallback: Whether to retry the registrable domain when the host is absent.
+
+        Returns:
+            A :class:`QueryResult`; ``found`` is False when the host is not in the corpus.
+        """
+        host, via, row = self.resolve(text, domain_fallback=domain_fallback)
+        if row is None:
+            return QueryResult(query=host, resolved_via=via, row=None, neighbors=[])
+        vector = self._index.reconstruct(row).reshape(1, -1).astype(np.float32)
+        wanted = min(k + 1, len(self))
+        scores, ids = self._index.search(vector, wanted)
+        neighbors: list[Neighbor] = []
+        hits = [(int(i), float(s)) for i, s in zip(ids[0], scores[0], strict=True) if i != row]
+        hits = hits[:k]
+        for rank, (neighbor_row, score) in enumerate(hits, start=1):
+            neighbors.append(
+                Neighbor(rank=rank, host="", score=score, row=neighbor_row),
+            )
+        names = self.names_of([hit.row for hit in neighbors])
+        neighbors = [
+            Neighbor(rank=hit.rank, host=name, score=hit.score, row=hit.row)
+            for hit, name in zip(neighbors, names, strict=True)
+        ]
+        return QueryResult(query=host, resolved_via=via, row=row, neighbors=neighbors)

wgl/knn/source.py

@@ -0,0 +1,403 @@
+"""Row sources for the kNN index builder.
+
+The builder is corpus-agnostic: it consumes :class:`RowBatch` objects and never learns where they
+came from. Two sources are provided.
+
+* :class:`ParquetSource` reads the **published release shards** (``row_id``/``host_key``/
+  ``domain_key``/``embedding``), locally or straight from a HuggingFace dataset repo over HTTP
+  range requests. ``host_key`` is already forward and normalized there, so nothing has to be
+  un-reversed. This is the quick path — a single row group is a usable proof-of-concept subset.
+* :class:`NpySource` reads the **cluster export** (a ``[n, dim]`` ``.npy`` plus the CC-reversed
+  ``node_names.txt``). This is the full-precision path used to scale up.
+
+Both stream in batches and never materialize the whole table.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import os
+import urllib.request
+from collections.abc import Iterator
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Protocol
+
+import numpy as np
+from numpy.typing import NDArray
+
+from wgl.release.identity import host_key as to_host_key
+
+_HF_RESOLVE = "https://huggingface.co/datasets/{repo}/resolve/{revision}/{path}"
+
+
+@dataclass(frozen=True)
+class RowBatch:
+    """A contiguous batch of rows from a host-embedding table.
+
+    Attributes:
+        row_ids: Global row ids, shape ``(m,)``. These index the *source* table, which for a
+            subset build is not the same as the index's own 0-based row numbering.
+        host_keys: Forward, normalized host keys (``www.example.com``), length ``m``.
+        vectors: Embedding rows, shape ``(m, dim)``. Not necessarily normalized.
+    """
+
+    row_ids: NDArray[np.int64]
+    host_keys: list[str]
+    vectors: NDArray[np.float32]
+
+    def __post_init__(self) -> None:
+        """Validate that the three parallel fields agree on length."""
+        n = len(self.host_keys)
+        if self.row_ids.shape[0] != n or self.vectors.shape[0] != n:
+            msg = (
+                f"RowBatch fields disagree: {self.row_ids.shape[0]} row_ids, {n} host_keys, "
+                f"{self.vectors.shape[0]} vectors"
+            )
+            raise ValueError(msg)
+
+
+class HostVectorSource(Protocol):
+    """A stream of :class:`RowBatch` objects plus the metadata the builder records."""
+
+    @property
+    def dim(self) -> int:
+        """Embedding dimensionality."""
+        ...
+
+    def describe(self) -> dict[str, object]:
+        """Return provenance fields to embed in the built index's ``meta.json``."""
+        ...
+
+    def batches(self) -> Iterator[RowBatch]:
+        """Yield row batches in ascending ``row_id`` order."""
+        ...
+
+
+class HttpRangeFile(io.RawIOBase):
+    """A seekable read-only file over HTTP range requests.
+
+    Parquet only needs the footer plus the column chunks actually requested, so wrapping a remote
+    shard in this class lets :mod:`pyarrow` read a schema (or one row group) without downloading
+    the whole file. HuggingFace serves ``206 Partial Content``, which is what makes this work.
+    """
+
+    def __init__(self, url: str, token: str | None = None) -> None:
+        """Open ``url`` and resolve its size and any redirect.
+
+        Args:
+            url: The URL to read.
+            token: Optional bearer token, required for private repositories.
+        """
+        self._headers = {"Authorization": f"Bearer {token}"} if token else {}
+        with self._open(url, {**self._headers, "Range": "bytes=0-0"}) as resp:
+            self.url = resp.geturl()
+            content_range = resp.headers.get("Content-Range")
+            if not content_range:
+                msg = f"server did not honour a Range request for {url}"
+                raise OSError(msg)
+            self.size = int(content_range.split("/")[1])
+        self._pos = 0
+
+    @staticmethod
+    def _open(url: str, headers: dict[str, str]) -> io.BufferedIOBase:
+        """Issue a GET with ``headers`` and return the open response."""
+        request = urllib.request.Request(url, headers=headers)  # https URLs only
+        return urllib.request.urlopen(request)
+
+    def readable(self) -> bool:
+        """Return True; the file is read-only but readable."""
+        return True
+
+    def seekable(self) -> bool:
+        """Return True; range requests make the stream seekable."""
+        return True
+
+    def seek(self, offset: int, whence: int = os.SEEK_SET) -> int:
+        """Move the read cursor and return the new absolute position."""
+        if whence == os.SEEK_SET:
+            self._pos = offset
+        elif whence == os.SEEK_CUR:
+            self._pos += offset
+        else:
+            self._pos = self.size + offset
+        return self._pos
+
+    def tell(self) -> int:
+        """Return the current absolute read position."""
+        return self._pos
+
+    def readinto(self, buffer: memoryview) -> int:  # type: ignore[override]
+        """Fill ``buffer`` from the current position and advance it."""
+        want = len(buffer)
+        if want == 0 or self._pos >= self.size:
+            return 0
+        end = min(self._pos + want, self.size) - 1
+        headers = {**self._headers, "Range": f"bytes={self._pos}-{end}"}
+        with self._open(self.url, headers) as resp:
+            data = resp.read()
+        buffer[: len(data)] = data
+        self._pos += len(data)
+        return len(data)
+
+
+def _open_shard(
+    location: str, token: str | None, buffer_bytes: int = 1 << 20
+) -> io.BufferedReader | io.IOBase:
+    """Open a shard that is either a local path or an ``https://`` URL."""
+    if location.startswith("https://"):
+        return io.BufferedReader(HttpRangeFile(location, token), buffer_size=buffer_bytes)  # type: ignore[arg-type]
+    return Path(location).open("rb")
+
+
+class ParquetSource:
+    """Read rows from published release Parquet shards (local paths or a HF dataset repo).
+
+    The release shards carry identity *co-located per row* (``row_id``, ``host_key``,
+    ``domain_key``) alongside the vector, so this source needs no separate join file and no
+    un-reversing — see ``docs/release-contract.md``.
+    """
+
+    def __init__(
+        self,
+        shards: list[str],
+        *,
+        token: str | None = None,
+        row_groups: int | None = None,
+        repo: str | None = None,
+        revision: str = "main",
+    ) -> None:
+        """Configure the source.
+
+        Args:
+            shards: Shard locations — local paths or ``https://`` URLs, in row order.
+            token: Bearer token for private repositories.
+            row_groups: Read at most this many row groups per shard (``None`` = all). This is the
+                subset knob: one row group of a release shard is ~1 M contiguous hosts.
+            repo: HuggingFace dataset repo id, recorded in ``meta.json`` when the shards came
+                from the Hub.
+            revision: Repo revision, recorded for provenance.
+        """
+        if not shards:
+            msg = "ParquetSource needs at least one shard"
+            raise ValueError(msg)
+        self.shards = shards
+        self.token = token
+        self.row_groups = row_groups
+        self.repo = repo
+        self.revision = revision
+        self._dim: int | None = None
+
+    @classmethod
+    def from_hf(
+        cls,
+        repo: str,
+        *,
+        token: str | None = None,
+        shards: list[int] | None = None,
+        row_groups: int | None = None,
+        revision: str = "main",
+        num_shards: int = 14,
+        pattern: str = "vectors/emb-{shard:05d}-of-{total:05d}.parquet",
+    ) -> ParquetSource:
+        """Build a source over shards of a HuggingFace dataset repo, read by range request.
+
+        Args:
+            repo: Dataset repo id, e.g. ``malteos/web-graph-embeddings``.
+            token: Bearer token; required while the dataset is private.
+            shards: Shard indices to read (``None`` = all ``num_shards``).
+            row_groups: Row groups per shard to read (``None`` = all).
+            revision: Git revision to resolve against.
+            num_shards: Total shard count, used to format the file names.
+            pattern: Shard path template.
+
+        Returns:
+            A configured :class:`ParquetSource`.
+        """
+        picked = list(range(num_shards)) if shards is None else shards
+        urls = [
+            _HF_RESOLVE.format(
+                repo=repo,
+                revision=revision,
+                path=pattern.format(shard=index, total=num_shards),
+            )
+            for index in picked
+        ]
+        return cls(urls, token=token, row_groups=row_groups, repo=repo, revision=revision)
+
+    @property
+    def dim(self) -> int:
+        """Embedding dimensionality, read from the first shard's schema."""
+        if self._dim is None:
+            import pyarrow.parquet as pq
+
+            with _open_shard(self.shards[0], self.token) as handle:
+                field = pq.ParquetFile(handle).schema_arrow.field("embedding")
+                self._dim = int(field.type.list_size)
+        return self._dim
+
+    def describe(self) -> dict[str, object]:
+        """Return provenance fields for ``meta.json``."""
+        return {
+            "source_kind": "parquet",
+            "source_repo": self.repo,
+            "source_revision": self.revision,
+            "source_shards": len(self.shards),
+            "source_row_groups_per_shard": self.row_groups,
+        }
+
+    def batches(self) -> Iterator[RowBatch]:
+        """Yield one :class:`RowBatch` per row group, in shard then row-group order."""
+        import pyarrow.parquet as pq
+
+        for location in self.shards:
+            with _open_shard(location, self.token) as handle:
+                reader = pq.ParquetFile(handle)
+                limit = reader.num_row_groups
+                if self.row_groups is not None:
+                    limit = min(limit, self.row_groups)
+                for group in range(limit):
+                    table = reader.read_row_group(
+                        group, columns=["row_id", "host_key", "embedding"]
+                    )
+                    vectors = np.asarray(
+                        table.column("embedding").combine_chunks().flatten(),
+                        dtype=np.float32,
+                    ).reshape(table.num_rows, -1)
+                    yield RowBatch(
+                        row_ids=table.column("row_id").to_numpy().astype(np.int64),
+                        host_keys=table.column("host_key").to_pylist(),
+                        vectors=vectors,
+                    )
+
+
+class NpySource:
+    """Read rows from a cluster export: an ``[n, dim]`` ``.npy`` plus ``node_names.txt``.
+
+    This is the full-precision path. The ``.npy`` is memory-mapped and read in batches, so a 27 GB
+    fp32 table never lands in RAM. Names are CC-**reversed** in that file and are converted to
+    forward host keys here, so the rest of the pipeline only ever sees forward hosts.
+    """
+
+    def __init__(
+        self,
+        emb: Path,
+        names: Path,
+        *,
+        batch_rows: int = 1_000_000,
+        limit: int | None = None,
+        reversed_names: bool = True,
+    ) -> None:
+        """Configure the source.
+
+        Args:
+            emb: Path to the ``[n, dim]`` float ``.npy`` export.
+            names: Path to the newline-delimited node-name file, one name per embedding row.
+            batch_rows: Rows per yielded batch.
+            limit: Read only the first ``limit`` rows (the subset knob for this source).
+            reversed_names: Whether ``names`` holds CC-reversed hosts (``com.example.www``).
+        """
+        self.emb = Path(emb)
+        self.names = Path(names)
+        self.batch_rows = batch_rows
+        self.limit = limit
+        self.reversed_names = reversed_names
+        self._shape: tuple[int, int] | None = None
+
+    @property
+    def shape(self) -> tuple[int, int]:
+        """Return the ``(rows, dim)`` shape of the export, honouring ``limit``."""
+        if self._shape is None:
+            table = np.load(self.emb, mmap_mode="r")
+            rows = table.shape[0] if self.limit is None else min(self.limit, table.shape[0])
+            self._shape = (int(rows), int(table.shape[1]))
+        return self._shape
+
+    @property
+    def dim(self) -> int:
+        """Embedding dimensionality."""
+        return self.shape[1]
+
+    def describe(self) -> dict[str, object]:
+        """Return provenance fields for ``meta.json``."""
+        return {
+            "source_kind": "npy",
+            "emb_source": str(self.emb),
+            "names_source": str(self.names),
+            "source_rows": self.shape[0],
+            "limit": self.limit,
+        }
+
+    def batches(self) -> Iterator[RowBatch]:
+        """Yield row batches, converting reversed CC names to forward host keys."""
+        table = np.load(self.emb, mmap_mode="r")
+        total = self.shape[0]
+        with self.names.open("r", encoding="utf-8", errors="replace") as handle:
+            start = 0
+            while start < total:
+                stop = min(start + self.batch_rows, total)
+                keys = []
+                for _ in range(stop - start):
+                    line = handle.readline()
+                    if not line:
+                        break
+                    keys.append(to_host_key(line.strip(), reversed_host=self.reversed_names))
+                if not keys:
+                    break
+                stop = start + len(keys)
+                yield RowBatch(
+                    row_ids=np.arange(start, stop, dtype=np.int64),
+                    host_keys=keys,
+                    vectors=np.asarray(table[start:stop], dtype=np.float32),
+                )
+                start = stop
+
+
+def load_hf_token(explicit: str | None = None) -> str | None:
+    """Resolve a HuggingFace token from an explicit value, the environment, or the CLI cache.
+
+    Args:
+        explicit: A token passed directly; wins when set.
+
+    Returns:
+        The token, or ``None`` when no token is configured.
+    """
+    if explicit:
+        return explicit
+    for var in ("HF_TOKEN", "HUGGING_FACE_HUB_TOKEN"):
+        value = os.environ.get(var)
+        if value:
+            return value
+    cached = Path.home() / ".cache" / "huggingface" / "token"
+    if cached.is_file():
+        text = cached.read_text(encoding="utf-8").strip()
+        if text:
+            return text
+    return None
+
+
+def read_release_manifest(
+    repo: str, token: str | None, revision: str = "main"
+) -> dict[str, object]:
+    """Fetch a release ``manifest.json`` from a HuggingFace dataset repo.
+
+    The manifest carries the ``release_id`` and ``model_space_version`` that make a built index
+    self-describing; a stale index is only detectable if these are copied across.
+
+    Args:
+        repo: Dataset repo id.
+        token: Bearer token for private repositories.
+        revision: Git revision.
+
+    Returns:
+        The parsed manifest, or an empty dict when it cannot be read.
+    """
+    url = _HF_RESOLVE.format(repo=repo, revision=revision, path="manifest.json")
+    headers = {"Authorization": f"Bearer {token}"} if token else {}
+    try:
+        request = urllib.request.Request(url, headers=headers)  # https URL
+        with urllib.request.urlopen(request) as resp:
+            return json.loads(resp.read().decode("utf-8"))
+    except (OSError, ValueError):
+        return {}

wgl/logging.py

@@ -0,0 +1,121 @@
+"""Structured logging and run-directory management.
+
+Every training/eval invocation creates a timestamped run directory under ``runs/<name>/<ts>/``
+containing the resolved ``config.yaml``, a human-readable ``log.txt``, and a machine-readable
+``metrics.jsonl`` (one JSON object per logged step). Logging goes to both stdout (rich) and the
+log file, so the workflow is friendly to humans and agents alike.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import os
+import time
+from pathlib import Path
+from typing import Any
+
+from rich.logging import RichHandler
+
+_LOGGER_NAME = "wgl"
+
+
+def get_logger() -> logging.Logger:
+    """Return the package logger (configured by :func:`setup_logging`)."""
+    return logging.getLogger(_LOGGER_NAME)
+
+
+def setup_logging(log_file: Path | None = None, level: int = logging.INFO) -> logging.Logger:
+    """Configure the package logger with a rich stdout handler and optional file handler.
+
+    Args:
+        log_file: Optional path to also write plain-text logs to.
+        level: Logging level.
+
+    Returns:
+        The configured logger.
+    """
+    logger = logging.getLogger(_LOGGER_NAME)
+    logger.setLevel(level)
+    logger.handlers.clear()
+    logger.propagate = False
+
+    rich_handler = RichHandler(rich_tracebacks=True, show_path=False, omit_repeated_times=False)
+    rich_handler.setFormatter(logging.Formatter("%(message)s", datefmt="[%X]"))
+    logger.addHandler(rich_handler)
+
+    if log_file is not None:
+        log_file.parent.mkdir(parents=True, exist_ok=True)
+        file_handler = logging.FileHandler(log_file)
+        file_handler.setFormatter(
+            logging.Formatter("%(asctime)s %(levelname)s %(message)s", datefmt="%Y-%m-%d %H:%M:%S")
+        )
+        logger.addHandler(file_handler)
+    return logger
+
+
+def timestamp() -> str:
+    """Return a filesystem-safe timestamp string."""
+    return time.strftime("%Y%m%d-%H%M%S", time.localtime())
+
+
+class RunDirectory:
+    """A timestamped output directory for a single experiment run.
+
+    Attributes:
+        path: The run directory path.
+    """
+
+    def __init__(self, runs_dir: Path, name: str) -> None:
+        """Create (or address) a run directory under ``runs_dir/name/<ts>``.
+
+        Args:
+            runs_dir: Root directory holding all runs.
+            name: Experiment name (a subdirectory).
+        """
+        self.path = Path(runs_dir) / name / timestamp()
+        self.path.mkdir(parents=True, exist_ok=True)
+        self._metrics_file = self.path / "metrics.jsonl"
+
+    @property
+    def config_path(self) -> Path:
+        """Path to the serialized resolved config."""
+        return self.path / "config.yaml"
+
+    @property
+    def log_path(self) -> Path:
+        """Path to the plain-text log file."""
+        return self.path / "log.txt"
+
+    @property
+    def checkpoint_path(self) -> Path:
+        """Path to the model checkpoint."""
+        return self.path / "model.pt"
+
+    def log_metrics(self, step: int, **metrics: Any) -> None:
+        """Append a metrics record to ``metrics.jsonl``.
+
+        Args:
+            step: Step or epoch index.
+            **metrics: Scalar metric values to record.
+        """
+        record = {"step": step, "wall_time": time.time(), **metrics}
+        with self._metrics_file.open("a") as fh:
+            fh.write(json.dumps(record) + "\n")
+
+    def read_metrics(self) -> list[dict[str, Any]]:
+        """Read all recorded metric rows back."""
+        if not self._metrics_file.exists():
+            return []
+        return [json.loads(line) for line in self._metrics_file.read_text().splitlines() if line]
+
+
+def mark_last_run(reference_dir: Path, run_path: Path) -> None:
+    """Write a ``.last_run`` pointer file so the CLI/Makefile can find the latest run.
+
+    Args:
+        reference_dir: Directory to write the pointer into (typically the data dir).
+        run_path: The run directory to record.
+    """
+    reference_dir.mkdir(parents=True, exist_ok=True)
+    (reference_dir / ".last_run").write_text(str(run_path) + os.linesep)

wgl/release/__init__.py

@@ -0,0 +1,81 @@
+"""V1 release contract: canonical identity keys and cross-release manifests.
+
+This package implements the *artifact contract* frozen in
+[`docs/release-contract.md`](../../../docs/release-contract.md) (runbook-v2 Phase 0):
+
+* :mod:`wgl.release.identity` — canonical, normalized host and domain/PLD keys (the stable
+  identifiers that survive across monthly releases, independent of Common Crawl's per-release
+  integer node ids);
+* :mod:`wgl.release.manifest` — a per-release manifest (stable key -> per-release local node id +
+  graph evidence) and the cross-release index that joins consecutive releases into a stable identity
+  table with ``active``/``new``/``missing``/``reappeared`` lifecycle states.
+
+These are deliberately model-agnostic: they describe *what is released and how it is joined* before
+any embedding is trained.
+"""
+
+from __future__ import annotations
+
+from wgl.release.identity import (
+    ConfidenceBucket,
+    EvidenceCohort,
+    confidence_bucket,
+    domain_key,
+    evidence_cohort,
+    forward_host,
+    host_key,
+    is_dangling,
+)
+from wgl.release.manifest import (
+    SCHEMA_VERSION,
+    CrossReleaseEntry,
+    CrossReleaseIndex,
+    HostState,
+    ReleaseManifest,
+    ReleaseRecord,
+)
+from wgl.release.schema import (
+    SCHEMA,
+    Field,
+    FieldLocation,
+    default_row_fields,
+    fixed_metadata_bytes_per_row,
+    sidecar_row_fields,
+)
+from wgl.release.sizing import (
+    CC_DOMAIN_NODES_ESTIMATE,
+    CC_HOST_NODES,
+    SizeEstimate,
+    estimate_size,
+    format_table,
+    size_table,
+)
+
+__all__ = [
+    "CC_DOMAIN_NODES_ESTIMATE",
+    "CC_HOST_NODES",
+    "SCHEMA",
+    "SCHEMA_VERSION",
+    "ConfidenceBucket",
+    "CrossReleaseEntry",
+    "CrossReleaseIndex",
+    "EvidenceCohort",
+    "Field",
+    "FieldLocation",
+    "HostState",
+    "ReleaseManifest",
+    "ReleaseRecord",
+    "SizeEstimate",
+    "confidence_bucket",
+    "default_row_fields",
+    "domain_key",
+    "estimate_size",
+    "evidence_cohort",
+    "fixed_metadata_bytes_per_row",
+    "format_table",
+    "forward_host",
+    "host_key",
+    "is_dangling",
+    "sidecar_row_fields",
+    "size_table",
+]

wgl/release/identity.py

@@ -0,0 +1,157 @@
+"""Canonical identity keys for the v1 release artifact.
+
+Common Crawl distributes the host graph with **reversed** host names (``com.example.www``) and
+**per-release, 0-based contiguous integer node ids**. The integer ids are *not* stable across
+monthly releases — a host gets a fresh id every crawl — so they cannot be the artifact's identity.
+The stable identifiers published in the artifact are *string keys* derived deterministically from
+the host name:
+
+* :func:`host_key`   — the canonical normalized host (``www.example.com``);
+* :func:`domain_key` — the registrable domain / pay-level domain, a.k.a. PLD (``example.com``).
+
+Both are pure functions of the input string (no network, no per-release state) so the same host maps
+to the same key in every release. The registrable domain is computed with the offline ``tldextract``
+snapshot (no network fetch), matching :mod:`wgl.tasks.tld`.
+
+This module also provides two small, *evidence-based* (not model-based) classifiers used in the
+release metadata: :func:`evidence_cohort` (the structural cohort from the runbook's coverage table)
+and :func:`confidence_bucket` (a reproducible confidence label derived only from observable graph
+evidence). See [`docs/release-contract.md`](../../../docs/release-contract.md).
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+
+import tldextract
+
+# Offline extractor: rely on the snapshot bundled with tldextract (no network fetch), exactly as
+# wgl.tasks.tld does, so host_to_tld and domain_key agree on the public-suffix boundary.
+_EXTRACT = tldextract.TLDExtract(suffix_list_urls=())
+
+
+def forward_host(name: str, reversed_host: bool = True) -> str:
+    """Return the forward (human) host name for a node name.
+
+    Args:
+        name: The host name, possibly in Common Crawl reversed notation (``com.example.www``).
+        reversed_host: Whether ``name`` is reversed. When False, ``name`` is returned normalized
+            but with its label order unchanged.
+
+    Returns:
+        The forward host (``www.example.com``), lowercased with surrounding/trailing dots stripped.
+    """
+    text = name.strip().strip(".").lower()
+    if not text:
+        return ""
+    if reversed_host:
+        return ".".join(reversed(text.split(".")))
+    return text
+
+
+def host_key(name: str, reversed_host: bool = True) -> str:
+    """Return the canonical, normalized **host key** for a node name.
+
+    The host key is the stable cross-release identifier for a host vertex. It is the forward host
+    name, lowercased, with whitespace and surrounding dots stripped. We deliberately keep the name
+    as published (ASCII/punycode labels are left as-is rather than IDNA-decoded) so the key is a
+    pure, reversible function of the source string — see the release contract for the rationale.
+
+    Args:
+        name: The (possibly reversed) Common Crawl host name.
+        reversed_host: Whether ``name`` is in reversed notation.
+
+    Returns:
+        The canonical host key (``www.example.com``); empty string for an empty input.
+    """
+    return forward_host(name, reversed_host=reversed_host)
+
+
+def domain_key(name: str, reversed_host: bool = True) -> str:
+    """Return the **domain/PLD key** (registrable domain) for a node name.
+
+    The registrable domain (eTLD+1, Common Crawl's "pay-level domain") groups every host under its
+    owning domain — both an independently useful artifact and the fallback representation for sparse
+    hosts. Computed from the public-suffix list via the offline ``tldextract`` snapshot.
+
+    Args:
+        name: The (possibly reversed) Common Crawl host name.
+        reversed_host: Whether ``name`` is in reversed notation.
+
+    Returns:
+        The registrable domain (``example.com``). When the host has no registrable domain above
+        its public suffix (e.g. the name *is* a public suffix, or an IP/onion address), falls back
+        to the canonical host key so every vertex still maps to a non-empty domain key.
+    """
+    host = forward_host(name, reversed_host=reversed_host)
+    if not host:
+        return ""
+    ext = _EXTRACT(host)
+    if ext.domain and ext.suffix:
+        return f"{ext.domain}.{ext.suffix}"
+    # No registrable domain above the public suffix (bare suffix, IP, etc.): host is its own domain.
+    return host
+
+
+class EvidenceCohort(StrEnum):
+    """Structural cohort of a host from its directed graph evidence (runbook-v2 coverage table)."""
+
+    BIDIRECTIONAL = "bidirectional"  # has both in-links and out-links
+    IN_ONLY = "in_only"  # has in-links, no out-links (dangling sink)
+    OUT_ONLY = "out_only"  # has out-links, no in-links (dangling source)
+    ISOLATED = "isolated"  # graph-isolated after preprocessing (no edges)
+
+
+class ConfidenceBucket(StrEnum):
+    """Reproducible confidence label from graph evidence only (not model-certainty claims)."""
+
+    HIGH = "high"  # well-observed: links in both directions
+    MEDIUM = "medium"  # one-directional evidence
+    LOW = "low"  # graph-isolated; relies on metadata / domain fallback
+
+
+def evidence_cohort(in_degree: int, out_degree: int) -> EvidenceCohort:
+    """Classify a host into its structural evidence cohort.
+
+    Args:
+        in_degree: Number of in-links (predecessors) in the release graph.
+        out_degree: Number of out-links (successors) in the release graph.
+
+    Returns:
+        The :class:`EvidenceCohort`.
+    """
+    has_in = in_degree > 0
+    has_out = out_degree > 0
+    if has_in and has_out:
+        return EvidenceCohort.BIDIRECTIONAL
+    if has_in:
+        return EvidenceCohort.IN_ONLY
+    if has_out:
+        return EvidenceCohort.OUT_ONLY
+    return EvidenceCohort.ISOLATED
+
+
+def confidence_bucket(in_degree: int, out_degree: int) -> ConfidenceBucket:
+    """Map graph evidence to a reproducible confidence bucket.
+
+    The bucket reflects *how much directed evidence* supports the host's representation, not any
+    model-certainty claim — so it is deterministic and reproducible from the published metadata.
+
+    Args:
+        in_degree: Number of in-links in the release graph.
+        out_degree: Number of out-links in the release graph.
+
+    Returns:
+        The :class:`ConfidenceBucket`.
+    """
+    cohort = evidence_cohort(in_degree, out_degree)
+    if cohort is EvidenceCohort.BIDIRECTIONAL:
+        return ConfidenceBucket.HIGH
+    if cohort is EvidenceCohort.ISOLATED:
+        return ConfidenceBucket.LOW
+    return ConfidenceBucket.MEDIUM
+
+
+def is_dangling(out_degree: int) -> bool:
+    """Whether a host is *dangling* (no out-links) — the dominant CC host-graph cohort (~75%)."""
+    return out_degree == 0

wgl/release/manifest.py

@@ -0,0 +1,327 @@
+"""Per-release manifest and the cross-release identity index.
+
+A **release manifest** records, for one monthly artifact, every host vertex's stable key together
+with the *ephemeral* per-release local node id and the graph evidence used for confidence/cohort
+metadata. The **cross-release index** joins consecutive manifests on the stable host key to produce
+a durable identity table with lifecycle states (``active``/``new``/``missing``/``reappeared``) — the
+mechanism that lets users compare consecutive releases even though Common Crawl reassigns
+integer node ids every crawl.
+
+This is the Phase 0.1 contract object. At full web scale the per-release host table is sharded
+(Phase 3.3); here it serializes to a small JSON header + a TSV of host rows, which is enough for the
+synthetic-release gate and for downsampled CC slices.
+"""
+
+from __future__ import annotations
+
+import csv
+import json
+from collections import Counter
+from collections.abc import Iterable, Sequence
+from dataclasses import dataclass, field
+from enum import StrEnum
+from pathlib import Path
+
+from wgl.data.graph import DirectedGraph
+from wgl.release.identity import (
+    confidence_bucket,
+    domain_key,
+    evidence_cohort,
+    host_key,
+    is_dangling,
+)
+
+# Bumped when the manifest/host-row schema changes in a backward-incompatible way.
+SCHEMA_VERSION = "0.1.0"
+
+_HOST_COLUMNS = (
+    "host_key",
+    "domain_key",
+    "local_node_id",
+    "in_degree",
+    "out_degree",
+    "is_dangling",
+    "cohort",
+    "confidence",
+)
+
+
+class HostState(StrEnum):
+    """Lifecycle state of a host key *as of* a given release, relative to prior releases."""
+
+    NEW = "new"  # present now, never seen in any earlier release
+    ACTIVE = "active"  # present now and in the immediately-preceding release (persistent)
+    REAPPEARED = "reappeared"  # present now, absent in the previous release, seen in an earlier one
+    MISSING = "missing"  # absent now, but present in some earlier release
+
+
+@dataclass(frozen=True)
+class ReleaseRecord:
+    """One host's row within a single release manifest.
+
+    Attributes:
+        host_key: Canonical stable host key (:func:`wgl.release.identity.host_key`).
+        domain_key: Registrable domain / PLD key.
+        local_node_id: The host's integer id *in this release's graph* (ephemeral across releases).
+        in_degree: In-links in this release.
+        out_degree: Out-links in this release.
+        is_dangling: Whether the host has no out-links.
+        cohort: Structural evidence cohort value.
+        confidence: Reproducible confidence-bucket value.
+    """
+
+    host_key: str
+    domain_key: str
+    local_node_id: int
+    in_degree: int
+    out_degree: int
+    is_dangling: bool
+    cohort: str
+    confidence: str
+
+    @classmethod
+    def from_evidence(
+        cls, key: str, dom: str, local_node_id: int, in_degree: int, out_degree: int
+    ) -> ReleaseRecord:
+        """Build a record, deriving the cohort/confidence/dangling fields from degrees."""
+        return cls(
+            host_key=key,
+            domain_key=dom,
+            local_node_id=int(local_node_id),
+            in_degree=int(in_degree),
+            out_degree=int(out_degree),
+            is_dangling=is_dangling(out_degree),
+            cohort=evidence_cohort(in_degree, out_degree).value,
+            confidence=confidence_bucket(in_degree, out_degree).value,
+        )
+
+
+@dataclass
+class ReleaseManifest:
+    """A single release's host manifest: stable keys -> per-release evidence.
+
+    Attributes:
+        release_id: Stable release identifier (e.g. ``cc-main-2026-mar-apr-may``).
+        source_crawls: The constituent monthly crawl ids (1-3 under the rolling-window policy).
+        records: Host rows keyed by ``host_key`` (insertion order = local node id order).
+        schema_version: Manifest schema version.
+        model_space_version: Embedding model-space tag; ``None`` until embeddings are trained.
+    """
+
+    release_id: str
+    source_crawls: list[str]
+    records: dict[str, ReleaseRecord] = field(default_factory=dict)
+    schema_version: str = SCHEMA_VERSION
+    model_space_version: str | None = None
+
+    @property
+    def host_keys(self) -> set[str]:
+        """The set of host keys present in this release."""
+        return set(self.records)
+
+    def __len__(self) -> int:
+        """Number of host records in the manifest."""
+        return len(self.records)
+
+    @classmethod
+    def from_graph(
+        cls,
+        graph: DirectedGraph,
+        release_id: str,
+        source_crawls: Sequence[str],
+        reversed_host: bool = True,
+        model_space_version: str | None = None,
+    ) -> ReleaseManifest:
+        """Build a manifest from a prepared :class:`DirectedGraph`.
+
+        Args:
+            graph: The release graph (its ``node_names`` provide the host strings).
+            release_id: Stable release identifier.
+            source_crawls: Constituent crawl ids for this release.
+            reversed_host: Whether ``node_names`` use Common Crawl reversed notation.
+            model_space_version: Optional model-space tag.
+
+        Returns:
+            The populated :class:`ReleaseManifest`.
+
+        Raises:
+            ValueError: If the graph has no node names, or two nodes collide on the same host key.
+        """
+        if graph.node_names is None:
+            msg = "ReleaseManifest.from_graph requires graph.node_names"
+            raise ValueError(msg)
+        in_deg = graph.in_degree()
+        out_deg = graph.out_degree()
+        records: dict[str, ReleaseRecord] = {}
+        for nid, name in enumerate(graph.node_names):
+            key = host_key(name, reversed_host=reversed_host)
+            if key in records:
+                msg = f"duplicate host key {key!r} (nodes {records[key].local_node_id} and {nid})"
+                raise ValueError(msg)
+            records[key] = ReleaseRecord.from_evidence(
+                key, domain_key(name, reversed_host=reversed_host), nid, in_deg[nid], out_deg[nid]
+            )
+        return cls(
+            release_id=release_id,
+            source_crawls=list(source_crawls),
+            records=records,
+            model_space_version=model_space_version,
+        )
+
+    def save(self, directory: str | Path) -> None:
+        """Persist the manifest to ``directory`` (``manifest.json`` header + ``hosts.tsv`` rows)."""
+        directory = Path(directory)
+        directory.mkdir(parents=True, exist_ok=True)
+        header = {
+            "release_id": self.release_id,
+            "source_crawls": self.source_crawls,
+            "schema_version": self.schema_version,
+            "model_space_version": self.model_space_version,
+            "num_hosts": len(self.records),
+        }
+        (directory / "manifest.json").write_text(json.dumps(header, indent=2))
+        with (directory / "hosts.tsv").open("w", newline="") as fh:
+            writer = csv.writer(fh, delimiter="\t")
+            writer.writerow(_HOST_COLUMNS)
+            for rec in self.records.values():
+                writer.writerow(
+                    [
+                        rec.host_key,
+                        rec.domain_key,
+                        rec.local_node_id,
+                        rec.in_degree,
+                        rec.out_degree,
+                        int(rec.is_dangling),
+                        rec.cohort,
+                        rec.confidence,
+                    ]
+                )
+
+    @classmethod
+    def load(cls, directory: str | Path) -> ReleaseManifest:
+        """Load a manifest previously written by :meth:`save`."""
+        directory = Path(directory)
+        header = json.loads((directory / "manifest.json").read_text())
+        records: dict[str, ReleaseRecord] = {}
+        with (directory / "hosts.tsv").open(newline="") as fh:
+            for row in csv.DictReader(fh, delimiter="\t"):
+                records[row["host_key"]] = ReleaseRecord(
+                    host_key=row["host_key"],
+                    domain_key=row["domain_key"],
+                    local_node_id=int(row["local_node_id"]),
+                    in_degree=int(row["in_degree"]),
+                    out_degree=int(row["out_degree"]),
+                    is_dangling=bool(int(row["is_dangling"])),
+                    cohort=row["cohort"],
+                    confidence=row["confidence"],
+                )
+        return cls(
+            release_id=header["release_id"],
+            source_crawls=list(header["source_crawls"]),
+            records=records,
+            schema_version=header.get("schema_version", SCHEMA_VERSION),
+            model_space_version=header.get("model_space_version"),
+        )
+
+
+@dataclass(frozen=True)
+class CrossReleaseEntry:
+    """A host key's durable cross-release identity row.
+
+    Attributes:
+        host_key: The stable host key.
+        first_release: Release id where the key was first observed.
+        last_seen_release: Most recent release where the key was present.
+        state: Lifecycle state as of the latest release in the index.
+        local_node_ids: Map ``release_id -> local node id`` for releases where the key was present.
+    """
+
+    host_key: str
+    first_release: str
+    last_seen_release: str
+    state: HostState
+    local_node_ids: dict[str, int]
+
+
+@dataclass
+class CrossReleaseIndex:
+    """Stable identity table joining an ordered sequence of release manifests on the host key.
+
+    The lifecycle ``state`` of each key is computed *relative to the latest release* in the
+    sequence: a key present in the latest release is ``new`` (never seen before), ``active`` (also
+    in the immediately-preceding release), or ``reappeared`` (absent in the previous release but
+    seen earlier); a key absent from the latest release but seen in any earlier one is ``missing``.
+    """
+
+    release_ids: list[str]
+    entries: dict[str, CrossReleaseEntry]
+
+    @classmethod
+    def build(cls, manifests: Iterable[ReleaseManifest]) -> CrossReleaseIndex:
+        """Join an *ordered* (oldest-to-newest) sequence of manifests into a cross-release index.
+
+        Args:
+            manifests: Release manifests in chronological order. At least one is required.
+
+        Returns:
+            The :class:`CrossReleaseIndex`.
+
+        Raises:
+            ValueError: If no manifests are given, or two share a release id.
+        """
+        mans = list(manifests)
+        if not mans:
+            msg = "CrossReleaseIndex.build requires at least one manifest"
+            raise ValueError(msg)
+        release_ids = [m.release_id for m in mans]
+        if len(set(release_ids)) != len(release_ids):
+            msg = f"duplicate release ids: {release_ids}"
+            raise ValueError(msg)
+
+        latest_id = release_ids[-1]
+        prev_id = release_ids[-2] if len(mans) >= 2 else None
+
+        entries: dict[str, CrossReleaseEntry] = {}
+        all_keys = set().union(*(m.host_keys for m in mans))
+        for key in all_keys:
+            present = [m for m in mans if key in m.records]
+            local_ids = {m.release_id: m.records[key].local_node_id for m in present}
+            entries[key] = CrossReleaseEntry(
+                host_key=key,
+                first_release=present[0].release_id,
+                last_seen_release=present[-1].release_id,
+                state=_state_for(set(local_ids), latest_id, prev_id),
+                local_node_ids=local_ids,
+            )
+        return cls(release_ids=release_ids, entries=entries)
+
+    def state_counts(self) -> dict[str, int]:
+        """Return a ``{state_value: count}`` summary over all entries."""
+        return dict(Counter(e.state.value for e in self.entries.values()))
+
+    def keys_in_state(self, state: HostState) -> set[str]:
+        """Return the host keys currently in the given lifecycle ``state``."""
+        return {k for k, e in self.entries.items() if e.state is state}
+
+
+def _state_for(present_releases: set[str], latest_id: str, prev_id: str | None) -> HostState:
+    """Lifecycle state of a key from the set of release ids it appears in.
+
+    Args:
+        present_releases: Release ids in which the key is present (non-empty — every key reaches
+            this function from the union of all manifests, so ``MISSING`` is the residual case).
+        latest_id: The newest release id in the index.
+        prev_id: The immediately-preceding release id, or ``None`` when only one release exists.
+
+    Returns:
+        ``MISSING`` if absent from the latest release; otherwise ``ACTIVE`` (also in the previous
+        release), ``REAPPEARED`` (absent in the previous release but seen in an earlier one), or
+        ``NEW`` (present only in the latest release).
+    """
+    if latest_id not in present_releases:
+        return HostState.MISSING
+    if prev_id is not None and prev_id in present_releases:
+        return HostState.ACTIVE
+    # Present now but not in the immediately-previous release: seen earlier => reappeared, else new.
+    seen_before_latest = present_releases - {latest_id}
+    return HostState.REAPPEARED if seen_before_latest else HostState.NEW

wgl/release/schema.py

@@ -0,0 +1,237 @@
+"""Declarative schema for the v1 release artifact (runbook-v2 Phase 0.2).
+
+The default download stays **compact**: one row per host (or per domain) with the stable key, the
+default embedding, and lightweight evidence/graph-context metadata. Larger diagnostics and any
+component/role vectors live in optional **sidecars** so they do not inflate the default artifact.
+
+Fields are split by *location*:
+
+* ``HEADER``      — one value per release (lives in the manifest header, not repeated per row);
+* ``ROW_DEFAULT`` — per-host/-domain, shipped in the default artifact;
+* ``ROW_SIDECAR`` — per-host/-domain, optional, shipped only on demand.
+
+This module is the single source of truth for which fields exist, their on-disk width, and where
+they live; :mod:`wgl.release.sizing` consumes the fixed-width metadata cost to estimate artifact
+sizes. See [`docs/release-contract.md`](../../../docs/release-contract.md) §0.2.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+
+class FieldLocation(StrEnum):
+    """Where a schema field lives in the released artifact."""
+
+    HEADER = "header"  # one value per release (manifest header)
+    ROW_DEFAULT = "row_default"  # per node, in the default artifact
+    ROW_SIDECAR = "row_sidecar"  # per node, optional sidecar
+
+
+class FieldKind(StrEnum):
+    """Coarse value kind, used by the size model."""
+
+    KEY = "key"  # variable-length string identifier
+    VECTOR = "vector"  # the embedding (size = dim x precision, handled by sizing)
+    SCALAR = "scalar"  # fixed-width numeric / categorical
+    META = "meta"  # release-level metadata (header)
+
+
+@dataclass(frozen=True)
+class Field:
+    """One artifact field.
+
+    Attributes:
+        name: Column name.
+        dtype: On-disk dtype label (e.g. ``uint32``, ``float32``, ``uint8``, ``str``).
+        location: Where the field lives (:class:`FieldLocation`).
+        kind: Value kind (:class:`FieldKind`).
+        nbytes: Fixed on-disk width in bytes for a single value, or ``None`` for variable-width
+            (the key) and the embedding (whose width is ``dim x precision``, computed by sizing).
+        doc: One-line description.
+    """
+
+    name: str
+    dtype: str
+    location: FieldLocation
+    kind: FieldKind
+    nbytes: int | None
+    doc: str
+
+
+# --- The v1 schema (compact default + optional sidecar) -----------------------------------------
+#
+# Degree maxima from the cc-main-2026-mar-apr-may host stats (maxindegree ~19.6M, maxoutdegree
+# ~7.0M, nodes ~262.35M) all fit uint32, so no field needs 64-bit width.
+
+SCHEMA: tuple[Field, ...] = (
+    # Release-level (header): one value per release, not repeated per row.
+    Field("release_id", "str", FieldLocation.HEADER, FieldKind.META, None, "Stable release id."),
+    Field(
+        "source_crawls",
+        "list[str]",
+        FieldLocation.HEADER,
+        FieldKind.META,
+        None,
+        "Constituent crawl ids (<=3, rolling-window policy).",
+    ),
+    Field(
+        "model_space_version",
+        "str",
+        FieldLocation.HEADER,
+        FieldKind.META,
+        None,
+        "Embedding space tag; vectors comparable only within the same value.",
+    ),
+    # Per-node default artifact.
+    Field(
+        "host_or_domain_key",
+        "str",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.KEY,
+        None,
+        "Stable host or domain/PLD key (the join identity).",
+    ),
+    Field(
+        "release_local_node_id",
+        "uint32",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        4,
+        "Ephemeral per-release integer id (for joining to this release's graph).",
+    ),
+    Field(
+        "embedding",
+        "float32",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.VECTOR,
+        None,
+        "Default L2-normalized embedding (width = dim x precision).",
+    ),
+    Field(
+        "confidence_bucket",
+        "uint8",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        1,
+        "Reproducible evidence-based confidence (high/medium/low).",
+    ),
+    Field(
+        "feature_availability",
+        "uint8",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        1,
+        "Bitmask of which feature groups were available for this node.",
+    ),
+    Field(
+        "in_degree", "uint32", FieldLocation.ROW_DEFAULT, FieldKind.SCALAR, 4, "In-links in release"
+    ),
+    Field(
+        "out_degree",
+        "uint32",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        4,
+        "Out-links in release.",
+    ),
+    Field(
+        "is_dangling",
+        "uint8",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        1,
+        "1 if the node has no out-links.",
+    ),
+    Field(
+        "component_bucket",
+        "uint8",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        1,
+        "Giant-SCC membership / coarse SCC-size bucket.",
+    ),
+    Field(
+        "pagerank",
+        "float32",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        4,
+        "Directed PageRank (from wgl.features).",
+    ),
+    Field(
+        "harmonic_centrality",
+        "float32",
+        FieldLocation.ROW_DEFAULT,
+        FieldKind.SCALAR,
+        4,
+        "Harmonic centrality where available.",
+    ),
+    # Optional sidecars (kept out of the default download).
+    Field(
+        "reverse_pagerank",
+        "float32",
+        FieldLocation.ROW_SIDECAR,
+        FieldKind.SCALAR,
+        4,
+        "Reverse (in-link) PageRank.",
+    ),
+    Field(
+        "source_role_embedding",
+        "float32",
+        FieldLocation.ROW_SIDECAR,
+        FieldKind.VECTOR,
+        None,
+        "Out-role/source component vector (when published separately).",
+    ),
+    Field(
+        "target_role_embedding",
+        "float32",
+        FieldLocation.ROW_SIDECAR,
+        FieldKind.VECTOR,
+        None,
+        "In-role/target component vector (when published separately).",
+    ),
+    Field(
+        "scc_id",
+        "uint32",
+        FieldLocation.ROW_SIDECAR,
+        FieldKind.SCALAR,
+        4,
+        "Full strongly-connected-component id (diagnostic).",
+    ),
+)
+
+
+def fields_at(location: FieldLocation) -> tuple[Field, ...]:
+    """Return the schema fields at the given location."""
+    return tuple(f for f in SCHEMA if f.location is location)
+
+
+def default_row_fields() -> tuple[Field, ...]:
+    """Per-node fields shipped in the default artifact."""
+    return fields_at(FieldLocation.ROW_DEFAULT)
+
+
+def sidecar_row_fields() -> tuple[Field, ...]:
+    """Per-node fields kept in optional sidecars."""
+    return fields_at(FieldLocation.ROW_SIDECAR)
+
+
+def header_fields() -> tuple[Field, ...]:
+    """Release-level header fields."""
+    return fields_at(FieldLocation.HEADER)
+
+
+def fixed_metadata_bytes_per_row() -> int:
+    """Fixed-width on-disk bytes per default row, excluding the variable key and the embedding.
+
+    This is the metadata overhead the size model adds on top of the embedding bytes (the key is
+    sized separately because it is variable-length).
+    """
+    return sum(
+        f.nbytes
+        for f in default_row_fields()
+        if f.kind is FieldKind.SCALAR and f.nbytes is not None
+    )

wgl/release/sizing.py

@@ -0,0 +1,151 @@
+"""Artifact size model for selecting the v1 embedding dimension and precision (Phase 0.2).
+
+Estimates the uncompressed and (approximate) compressed footprint of the host and domain artifacts
+across candidate dimensions (64/128/256) and precisions (fp32/fp16/int8). The dominant term is the
+embedding matrix (``num_nodes x dim x bytes_per_value``); the schema's fixed-width metadata
+(:func:`wgl.release.schema.fixed_metadata_bytes_per_row`) plus the variable key add a smaller,
+dimension-independent overhead.
+
+Compression ratios are deliberate, documented estimates: L2-normalized float vectors are
+high-entropy and compress poorly, so the ratios are near 1.0 for floats and lower only for int8 and
+for the repetitive metadata column block. They are decision aids, not guarantees — the real ratio is
+measured at export time (Phase 3.3). Node counts come from the real cc-main-2026-mar-apr-may host
+stats; the domain count is an estimate until the domain graph is extracted.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from wgl.release.schema import fixed_metadata_bytes_per_row
+
+# Real host-graph node count (cc-main-2026-mar-apr-may host.stats: nodes=262351908).
+CC_HOST_NODES = 262_351_908
+# Domain/PLD node count is not yet extracted on the cluster (host graph only). Registrable domains
+# typically number ~40-55% of hosts for these crawls; use ~135M as a planning estimate, replaced by
+# the measured count when the domain graph is prepared.
+CC_DOMAIN_NODES_ESTIMATE = 135_000_000
+
+# On-disk bytes per embedding value by precision label.
+PRECISION_BYTES = {"fp32": 4, "fp16": 2, "int8": 1}
+
+# Approximate compression ratios (compressed / uncompressed). Floats barely compress; int8 quantized
+# vectors and the small-int metadata block compress more. Estimates — verified at export time.
+_VECTOR_COMPRESSION = {"fp32": 0.95, "fp16": 0.95, "int8": 0.70}
+_METADATA_COMPRESSION = 0.45
+
+# Average bytes for a stable key string (forward host / domain). Estimate; tune from real keys.
+DEFAULT_AVG_KEY_BYTES = 24
+
+_GIB = 1024**3
+
+
+@dataclass(frozen=True)
+class SizeEstimate:
+    """Estimated footprint of one artifact at a given dimension and precision.
+
+    Attributes:
+        num_nodes: Number of rows (hosts or domains).
+        dim: Embedding dimension.
+        precision: Precision label (``fp32``/``fp16``/``int8``).
+        vector_bytes: Uncompressed embedding-matrix bytes.
+        metadata_bytes: Uncompressed default-row metadata bytes (key + fixed fields).
+        total_uncompressed: ``vector_bytes + metadata_bytes``.
+        total_compressed: Approximate compressed total.
+    """
+
+    num_nodes: int
+    dim: int
+    precision: str
+    vector_bytes: int
+    metadata_bytes: int
+    total_uncompressed: int
+    total_compressed: int
+
+    @property
+    def vector_gib(self) -> float:
+        """Uncompressed embedding-matrix size in GiB."""
+        return self.vector_bytes / _GIB
+
+    @property
+    def total_uncompressed_gib(self) -> float:
+        """Uncompressed total in GiB."""
+        return self.total_uncompressed / _GIB
+
+    @property
+    def total_compressed_gib(self) -> float:
+        """Approximate compressed total in GiB."""
+        return self.total_compressed / _GIB
+
+
+def estimate_size(
+    num_nodes: int,
+    dim: int,
+    precision: str = "fp32",
+    avg_key_bytes: int = DEFAULT_AVG_KEY_BYTES,
+) -> SizeEstimate:
+    """Estimate the footprint of one artifact.
+
+    Args:
+        num_nodes: Number of rows (hosts or domains).
+        dim: Embedding dimension (the *published* default vector width, not any internal
+            ComplEx real+imag doubling).
+        precision: One of ``fp32``/``fp16``/``int8``.
+        avg_key_bytes: Average bytes for a stable key string.
+
+    Returns:
+        The :class:`SizeEstimate`.
+
+    Raises:
+        ValueError: On an unknown precision or non-positive dim/nodes.
+    """
+    if precision not in PRECISION_BYTES:
+        msg = f"unknown precision {precision!r}; expected one of {sorted(PRECISION_BYTES)}"
+        raise ValueError(msg)
+    if dim <= 0 or num_nodes <= 0:
+        msg = f"dim and num_nodes must be positive (got dim={dim}, num_nodes={num_nodes})"
+        raise ValueError(msg)
+
+    vector_bytes = num_nodes * dim * PRECISION_BYTES[precision]
+    metadata_bytes = num_nodes * (avg_key_bytes + fixed_metadata_bytes_per_row())
+    total_uncompressed = vector_bytes + metadata_bytes
+    total_compressed = round(
+        vector_bytes * _VECTOR_COMPRESSION[precision] + metadata_bytes * _METADATA_COMPRESSION
+    )
+    return SizeEstimate(
+        num_nodes=num_nodes,
+        dim=dim,
+        precision=precision,
+        vector_bytes=vector_bytes,
+        metadata_bytes=metadata_bytes,
+        total_uncompressed=total_uncompressed,
+        total_compressed=total_compressed,
+    )
+
+
+def size_table(
+    num_nodes: int,
+    dims: tuple[int, ...] = (64, 128, 256),
+    precisions: tuple[str, ...] = ("fp32", "fp16", "int8"),
+    avg_key_bytes: int = DEFAULT_AVG_KEY_BYTES,
+) -> list[SizeEstimate]:
+    """Estimate sizes across the candidate (dim, precision) grid for one artifact."""
+    return [
+        estimate_size(num_nodes, dim, precision, avg_key_bytes)
+        for dim in dims
+        for precision in precisions
+    ]
+
+
+def format_table(estimates: list[SizeEstimate], label: str) -> str:
+    """Render a human-readable size table (GiB) for a list of estimates."""
+    lines = [
+        f"{label} ({estimates[0].num_nodes:,} nodes)" if estimates else f"{label} (no rows)",
+        f"  {'dim':>4}  {'prec':>5}  {'vectors':>10}  {'uncompressed':>13}  {'compressed':>11}",
+    ]
+    lines.extend(
+        f"  {e.dim:>4}  {e.precision:>5}  {e.vector_gib:>8.1f}GiB  "
+        f"{e.total_uncompressed_gib:>11.1f}GiB  {e.total_compressed_gib:>9.1f}GiB"
+        for e in estimates
+    )
+    return "\n".join(lines)