MVP Milestone 2

MVP_PROGRESS.md

@@ -30,3 +30,73 @@ Source-file status:
 - `data/gcmd_hierarchy.json` unchanged by this milestone.
 - `data/articles.json` unchanged by this milestone.
 - `prototype/app_hf_poc.py` unchanged by this milestone.
+
+## Milestone 2: Hierarchy Loader and Vocabulary Index
+
+Status: completed
+
+Scope completed:
+- Deterministic loader for `data/gcmd_hierarchy.json`
+- Recursive traversal of the GCMD hierarchy through `Category`, `Topic`, `Term`, `Variable_Level_1`, `Variable_Level_2`, and `Variable_Level_3`
+- Canonical UUID-bearing concept records
+- Vocabulary file SHA-256 version hash
+- Canonical paths excluding the root `Category`
+- UUID, canonical path, parent, direct-child, Topic, Term, Variable, ancestor, and descendant lookups
+- Typed vocabulary errors for malformed hierarchy content
+- Focused small hierarchy fixture and full-data smoke tests
+
+Implemented models and lookup structures:
+- `CanonicalConceptRecord` in `src/gcmd_classifier/models.py`
+- `VocabularyIndex` in `src/gcmd_classifier/vocabulary/index.py`
+- `load_vocabulary`, `build_vocabulary_index`, and `calculate_file_hash` in `src/gcmd_classifier/vocabulary/loader.py`
+
+Hierarchy assumptions discovered from the current real file:
+- The root node is the non-UUID `Category` named `EARTH SCIENCE`.
+- The current file contains only the `EARTH SCIENCE` Category represented by that root.
+- Every current Topic and Term has a UUID.
+- The only UUID-less node in the current file is the root Category.
+- Current UUID-bearing record counts are: 14 Topics, 144 Terms, 1,368 `Variable_Level_1`, 1,456 `Variable_Level_2`, 553 `Variable_Level_3`, and 3,535 total UUID-bearing records.
+- Current observed transitions are adjacent only: `Category -> Topic -> Term -> Variable_Level_1 -> Variable_Level_2 -> Variable_Level_3`.
+
+Error cases implemented:
+- Invalid JSON
+- Non-object root or child nodes
+- Missing or empty required `level` and `name`
+- Unsupported hierarchy levels
+- UUID-bearing root Category
+- UUID-less non-root nodes
+- Invalid parent-child level transitions
+- Children under `Variable_Level_3`
+- Non-list `children`
+- Non-string `definition`
+- Duplicate UUIDs
+- Same canonical path assigned to different UUIDs
+- Lookup requests for unknown UUIDs or invalid lookup scopes
+
+Verification:
+- `python -m ruff check .` passed: `All checks passed!`
+- `python -m ruff format --check .` passed: `8 files already formatted`
+- `python -m pytest` passed: `23 passed in 0.14s`
+
+Out of scope, not started:
+- Article loading
+- Model integration
+- Topic routing
+- Term routing
+- Variable-level classification
+- Deterministic final classification validation beyond vocabulary lookup integrity
+- Redundancy removal
+- Persistence, caching, and output generation
+- UI and root `app.py`
+
+Source-file status:
+- `data/gcmd_hierarchy.json` unchanged by this milestone.
+- `data/articles.json` unchanged by this milestone.
+- `prototype/app_hf_poc.py` unchanged by this milestone.
+
+Deviations from approved plan:
+- None.
+
+Remaining risks or assumptions:
+- The loader remains fully dynamic and does not hard-code vocabulary names, UUIDs, paths, branches, or record counts, but full-data smoke tests intentionally assert current-file counts.
+- Future approved vocabulary rules would be needed before accepting any UUID-less non-root hierarchy node.

src/gcmd_classifier/errors.py

@@ -0,0 +1,29 @@
+"""Typed exceptions for GCMD classifier deterministic components."""
+
+
+class GCMDClassifierError(Exception):
+    """Base exception for GCMD classifier errors."""
+
+
+class VocabularyError(GCMDClassifierError):
+    """Base exception for vocabulary loading and indexing errors."""
+
+
+class MalformedHierarchyError(VocabularyError):
+    """Raised when a hierarchy node is missing required structure or fields."""
+
+
+class DuplicateUUIDError(VocabularyError):
+    """Raised when the hierarchy contains the same UUID more than once."""
+
+
+class DuplicateCanonicalPathError(VocabularyError):
+    """Raised when one canonical path is assigned to multiple UUIDs."""
+
+
+class InvalidHierarchyTransitionError(VocabularyError):
+    """Raised when a node appears under an invalid hierarchy parent level."""
+
+
+class VocabularyLookupError(VocabularyError):
+    """Raised when a requested vocabulary concept or relationship is missing."""

src/gcmd_classifier/models.py

@@ -0,0 +1,42 @@
+"""Shared typed models for the GCMD classifier MVP."""
+
+from typing import Literal
+
+from pydantic import BaseModel, ConfigDict, Field
+
+HierarchyLevel = Literal[
+    "Topic",
+    "Term",
+    "Variable_Level_1",
+    "Variable_Level_2",
+    "Variable_Level_3",
+]
+SourceHierarchyLevel = Literal[
+    "Category",
+    "Topic",
+    "Term",
+    "Variable_Level_1",
+    "Variable_Level_2",
+    "Variable_Level_3",
+]
+
+
+class CanonicalConceptRecord(BaseModel):
+    """Canonical UUID-bearing GCMD concept record derived from the hierarchy."""
+
+    model_config = ConfigDict(frozen=True)
+
+    UUID: str
+    name: str
+    level: HierarchyLevel
+    topic: str
+    term: str | None = None
+    path_components: tuple[str, ...]
+    canonical_path: str
+    parent_uuid: str | None = None
+    parent_name: str | None = None
+    child_uuids: tuple[str, ...] = Field(default_factory=tuple)
+    has_children: bool
+    assignable: bool
+    definition: str | None = None
+    vocabulary_version: str

src/gcmd_classifier/vocabulary/__init__.py

@@ -0,0 +1,15 @@
+"""Vocabulary loading and indexing for GCMD Science Keywords."""
+
+from gcmd_classifier.vocabulary.index import VocabularyIndex
+from gcmd_classifier.vocabulary.loader import (
+    build_vocabulary_index,
+    calculate_file_hash,
+    load_vocabulary,
+)
+
+__all__ = [
+    "VocabularyIndex",
+    "build_vocabulary_index",
+    "calculate_file_hash",
+    "load_vocabulary",
+]

src/gcmd_classifier/vocabulary/index.py

@@ -0,0 +1,115 @@
+"""In-memory canonical vocabulary index for GCMD Science Keywords."""
+
+from __future__ import annotations
+
+from collections import Counter
+from collections.abc import Mapping
+from dataclasses import dataclass
+from types import MappingProxyType
+
+from gcmd_classifier.errors import VocabularyLookupError
+from gcmd_classifier.models import CanonicalConceptRecord, HierarchyLevel
+
+
+@dataclass(frozen=True)
+class VocabularyIndex:
+    """Lookup structures derived from the canonical GCMD hierarchy."""
+
+    records_by_uuid: Mapping[str, CanonicalConceptRecord]
+    records_by_path: Mapping[str, CanonicalConceptRecord]
+    parent_by_uuid: Mapping[str, str | None]
+    children_by_uuid: Mapping[str | None, tuple[str, ...]]
+    topic_uuids: tuple[str, ...]
+    terms_by_topic_uuid: Mapping[str, tuple[str, ...]]
+    variables_by_parent_uuid: Mapping[str, tuple[str, ...]]
+    vocabulary_version: str
+    root_level: str
+    root_name: str
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "records_by_uuid", MappingProxyType(dict(self.records_by_uuid)))
+        object.__setattr__(self, "records_by_path", MappingProxyType(dict(self.records_by_path)))
+        object.__setattr__(self, "parent_by_uuid", MappingProxyType(dict(self.parent_by_uuid)))
+        object.__setattr__(self, "children_by_uuid", MappingProxyType(dict(self.children_by_uuid)))
+        object.__setattr__(
+            self, "terms_by_topic_uuid", MappingProxyType(dict(self.terms_by_topic_uuid))
+        )
+        object.__setattr__(
+            self, "variables_by_parent_uuid", MappingProxyType(dict(self.variables_by_parent_uuid))
+        )
+
+    @property
+    def records(self) -> tuple[CanonicalConceptRecord, ...]:
+        """All UUID-bearing canonical records in deterministic path order."""
+        return tuple(self.records_by_path[path] for path in sorted(self.records_by_path))
+
+    def __len__(self) -> int:
+        return len(self.records_by_uuid)
+
+    def get(self, uuid: str) -> CanonicalConceptRecord:
+        """Return a concept by UUID or raise a typed lookup error."""
+        try:
+            return self.records_by_uuid[uuid]
+        except KeyError as exc:
+            raise VocabularyLookupError(f"Unknown GCMD UUID: {uuid}") from exc
+
+    def get_by_path(self, canonical_path: str) -> CanonicalConceptRecord:
+        """Return a concept by canonical path or raise a typed lookup error."""
+        try:
+            return self.records_by_path[canonical_path]
+        except KeyError as exc:
+            raise VocabularyLookupError(f"Unknown GCMD canonical path: {canonical_path}") from exc
+
+    def parent_of(self, uuid: str) -> str | None:
+        """Return the parent UUID for a concept, or None for Topic records."""
+        self.get(uuid)
+        return self.parent_by_uuid[uuid]
+
+    def children_of(self, parent_uuid: str | None) -> tuple[str, ...]:
+        """Return direct child UUIDs for a parent UUID; None returns Topic UUIDs."""
+        if parent_uuid is not None:
+            self.get(parent_uuid)
+        return self.children_by_uuid.get(parent_uuid, ())
+
+    def topics(self) -> tuple[CanonicalConceptRecord, ...]:
+        """Return all Topic records in source order."""
+        return tuple(self.get(uuid) for uuid in self.topic_uuids)
+
+    def terms_for_topic(self, topic_uuid: str) -> tuple[CanonicalConceptRecord, ...]:
+        """Return direct Term records under a Topic UUID."""
+        topic = self.get(topic_uuid)
+        if topic.level != "Topic":
+            raise VocabularyLookupError(f"UUID is not a Topic: {topic_uuid}")
+        return tuple(self.get(uuid) for uuid in self.terms_by_topic_uuid.get(topic_uuid, ()))
+
+    def variables_for_parent(self, parent_uuid: str) -> tuple[CanonicalConceptRecord, ...]:
+        """Return direct Variable records under a Term or Variable parent."""
+        parent = self.get(parent_uuid)
+        if parent.level == "Topic":
+            raise VocabularyLookupError(
+                f"Topic UUID does not have variable children: {parent_uuid}"
+            )
+        return tuple(self.get(uuid) for uuid in self.variables_by_parent_uuid.get(parent_uuid, ()))
+
+    def ancestors_of(self, uuid: str) -> tuple[str, ...]:
+        """Return ancestor UUIDs from nearest parent to highest UUID-bearing ancestor."""
+        self.get(uuid)
+        ancestors: list[str] = []
+        current = self.parent_by_uuid[uuid]
+        while current is not None:
+            ancestors.append(current)
+            current = self.parent_by_uuid[current]
+        return tuple(ancestors)
+
+    def is_ancestor(self, ancestor_uuid: str, descendant_uuid: str) -> bool:
+        """Return whether the first UUID is an ancestor of the second UUID."""
+        self.get(ancestor_uuid)
+        return ancestor_uuid in self.ancestors_of(descendant_uuid)
+
+    def is_descendant(self, descendant_uuid: str, ancestor_uuid: str) -> bool:
+        """Return whether the first UUID is a descendant of the second UUID."""
+        return self.is_ancestor(ancestor_uuid, descendant_uuid)
+
+    def count_by_level(self) -> Counter[HierarchyLevel]:
+        """Return UUID-bearing record counts by hierarchy level."""
+        return Counter(record.level for record in self.records_by_uuid.values())

src/gcmd_classifier/vocabulary/loader.py

@@ -0,0 +1,304 @@
+"""Deterministic loader for the GCMD Science Keyword hierarchy."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+from gcmd_classifier.errors import (
+    DuplicateCanonicalPathError,
+    DuplicateUUIDError,
+    InvalidHierarchyTransitionError,
+    MalformedHierarchyError,
+)
+from gcmd_classifier.models import CanonicalConceptRecord
+from gcmd_classifier.vocabulary.index import VocabularyIndex
+
+ROOT_LEVEL = "Category"
+ROOT_NAME = "EARTH SCIENCE"
+CANONICAL_PATH_SEPARATOR = " > "
+
+_ALLOWED_LEVELS = {
+    ROOT_LEVEL,
+    "Topic",
+    "Term",
+    "Variable_Level_1",
+    "Variable_Level_2",
+    "Variable_Level_3",
+}
+_VALID_CHILD_LEVELS = {
+    ROOT_LEVEL: "Topic",
+    "Topic": "Term",
+    "Term": "Variable_Level_1",
+    "Variable_Level_1": "Variable_Level_2",
+    "Variable_Level_2": "Variable_Level_3",
+    "Variable_Level_3": None,
+}
+_ASSIGNABLE_LEVELS = {
+    "Topic",
+    "Term",
+    "Variable_Level_1",
+    "Variable_Level_2",
+    "Variable_Level_3",
+}
+
+
+def calculate_file_hash(path: str | Path) -> str:
+    """Return a reproducible SHA-256 hash of a vocabulary file's raw bytes."""
+    return hashlib.sha256(Path(path).read_bytes()).hexdigest()
+
+
+def load_vocabulary(path: str | Path) -> VocabularyIndex:
+    """Load a GCMD hierarchy JSON file and return a canonical vocabulary index."""
+    hierarchy_path = Path(path)
+    raw_bytes = hierarchy_path.read_bytes()
+    try:
+        data = json.loads(raw_bytes)
+    except json.JSONDecodeError as exc:
+        raise MalformedHierarchyError(f"Invalid JSON in vocabulary file: {hierarchy_path}") from exc
+    vocabulary_version = hashlib.sha256(raw_bytes).hexdigest()
+    return build_vocabulary_index(data, vocabulary_version=vocabulary_version)
+
+
+def build_vocabulary_index(data: Any, vocabulary_version: str) -> VocabularyIndex:
+    """Build a canonical vocabulary index from an already-loaded hierarchy object."""
+    builder = _VocabularyIndexBuilder(vocabulary_version=vocabulary_version)
+    return builder.build(data)
+
+
+class _VocabularyIndexBuilder:
+    def __init__(self, vocabulary_version: str) -> None:
+        self.vocabulary_version = vocabulary_version
+        self.records_by_uuid: dict[str, CanonicalConceptRecord] = {}
+        self.records_by_path: dict[str, CanonicalConceptRecord] = {}
+        self.parent_by_uuid: dict[str, str | None] = {}
+        self.children_by_uuid: dict[str | None, tuple[str, ...]] = {}
+        self.topic_uuids: list[str] = []
+        self.terms_by_topic_uuid: dict[str, tuple[str, ...]] = {}
+        self.variables_by_parent_uuid: dict[str, tuple[str, ...]] = {}
+
+    def build(self, data: Any) -> VocabularyIndex:
+        if not isinstance(data, dict):
+            raise MalformedHierarchyError("Hierarchy root must be a JSON object.")
+        root_level = self._required_string(data, "level", path="<root>")
+        root_name = self._required_string(data, "name", path="<root>")
+        if root_level != ROOT_LEVEL or root_name != ROOT_NAME:
+            raise MalformedHierarchyError(
+                f"Hierarchy root must be {ROOT_LEVEL!r} named {ROOT_NAME!r}; "
+                f"found level={root_level!r}, name={root_name!r}."
+            )
+        if data.get("UUID"):
+            raise MalformedHierarchyError("The known root Category must not define a UUID.")
+
+        root_children = self._children(data, path=ROOT_NAME)
+        self._validate_children_transition(root_level, root_children, path=ROOT_NAME)
+        topic_uuids = self._walk_children(
+            root_children,
+            parent_level=root_level,
+            parent_uuid=None,
+            parent_name=root_name,
+            path_components=(),
+            topic=None,
+            term=None,
+            source_path=root_name,
+        )
+        self.topic_uuids = list(topic_uuids)
+        self.children_by_uuid[None] = topic_uuids
+
+        return VocabularyIndex(
+            records_by_uuid=self.records_by_uuid,
+            records_by_path=self.records_by_path,
+            parent_by_uuid=self.parent_by_uuid,
+            children_by_uuid=self.children_by_uuid,
+            topic_uuids=tuple(self.topic_uuids),
+            terms_by_topic_uuid=self.terms_by_topic_uuid,
+            variables_by_parent_uuid=self.variables_by_parent_uuid,
+            vocabulary_version=self.vocabulary_version,
+            root_level=root_level,
+            root_name=root_name,
+        )
+
+    def _walk_children(
+        self,
+        children: list[Any],
+        *,
+        parent_level: str,
+        parent_uuid: str | None,
+        parent_name: str,
+        path_components: tuple[str, ...],
+        topic: str | None,
+        term: str | None,
+        source_path: str,
+    ) -> tuple[str, ...]:
+        child_uuids: list[str] = []
+        for child in children:
+            child_uuid = self._walk_node(
+                child,
+                parent_level=parent_level,
+                parent_uuid=parent_uuid,
+                parent_name=parent_name,
+                path_components=path_components,
+                topic=topic,
+                term=term,
+                source_path=source_path,
+            )
+            child_uuids.append(child_uuid)
+        return tuple(child_uuids)
+
+    def _walk_node(
+        self,
+        node: Any,
+        *,
+        parent_level: str,
+        parent_uuid: str | None,
+        parent_name: str,
+        path_components: tuple[str, ...],
+        topic: str | None,
+        term: str | None,
+        source_path: str,
+    ) -> str:
+        if not isinstance(node, dict):
+            raise MalformedHierarchyError(
+                f"Hierarchy node under {source_path!r} must be an object."
+            )
+
+        level = self._required_string(node, "level", path=source_path)
+        name = self._required_string(node, "name", path=source_path)
+        node_path = f"{source_path} / {name}"
+        self._validate_level(level, path=node_path)
+        self._validate_transition(parent_level, level, path=node_path)
+
+        uuid = self._required_uuid(node, level=level, path=node_path)
+        definition = self._optional_string(node, "definition", path=node_path)
+        children = self._children(node, path=node_path)
+        self._validate_children_transition(level, children, path=node_path)
+
+        next_path_components = (*path_components, name)
+        canonical_path = CANONICAL_PATH_SEPARATOR.join(next_path_components)
+        next_topic = name if level == "Topic" else topic
+        next_term = name if level == "Term" else term
+
+        direct_child_uuids = self._walk_children(
+            children,
+            parent_level=level,
+            parent_uuid=uuid,
+            parent_name=name,
+            path_components=next_path_components,
+            topic=next_topic,
+            term=next_term,
+            source_path=node_path,
+        )
+
+        if uuid in self.records_by_uuid:
+            existing = self.records_by_uuid[uuid]
+            raise DuplicateUUIDError(
+                f"Duplicate UUID {uuid!r} at {canonical_path!r}; "
+                f"already used by {existing.canonical_path!r}."
+            )
+        existing_path = self.records_by_path.get(canonical_path)
+        if existing_path is not None and existing_path.UUID != uuid:
+            raise DuplicateCanonicalPathError(
+                f"Canonical path {canonical_path!r} is assigned to both "
+                f"{existing_path.UUID!r} and {uuid!r}."
+            )
+
+        record = CanonicalConceptRecord(
+            UUID=uuid,
+            name=name,
+            level=level,  # type: ignore[arg-type]
+            topic=next_topic or name,
+            term=next_term,
+            path_components=next_path_components,
+            canonical_path=canonical_path,
+            parent_uuid=parent_uuid,
+            parent_name=parent_name,
+            child_uuids=direct_child_uuids,
+            has_children=bool(children),
+            assignable=True,
+            definition=definition,
+            vocabulary_version=self.vocabulary_version,
+        )
+        self.records_by_uuid[uuid] = record
+        self.records_by_path[canonical_path] = record
+        self.parent_by_uuid[uuid] = parent_uuid
+        self.children_by_uuid[uuid] = direct_child_uuids
+        if level == "Topic":
+            self.terms_by_topic_uuid[uuid] = direct_child_uuids
+        elif level != "Variable_Level_3":
+            self.variables_by_parent_uuid[uuid] = direct_child_uuids
+        return uuid
+
+    def _required_string(self, node: dict[str, Any], field: str, *, path: str) -> str:
+        value = node.get(field)
+        if not isinstance(value, str) or not value.strip():
+            raise MalformedHierarchyError(f"Node at {path!r} must contain non-empty {field!r}.")
+        return value
+
+    def _optional_string(self, node: dict[str, Any], field: str, *, path: str) -> str | None:
+        if field not in node:
+            return None
+        value = node[field]
+        if value is None:
+            return None
+        if not isinstance(value, str):
+            raise MalformedHierarchyError(f"Optional field {field!r} at {path!r} must be a string.")
+        return value
+
+    def _required_uuid(self, node: dict[str, Any], *, level: str, path: str) -> str:
+        value = node.get("UUID")
+        if not isinstance(value, str) or not value.strip():
+            raise MalformedHierarchyError(
+                f"UUID-less hierarchy node at {path!r} is not allowed for level {level!r}."
+            )
+        if level in {"Topic", "Term"} and not value:
+            raise MalformedHierarchyError(f"{level} at {path!r} must contain a UUID.")
+        return value
+
+    def _children(self, node: dict[str, Any], *, path: str) -> list[Any]:
+        children = node.get("children", [])
+        if children is None:
+            return []
+        if not isinstance(children, list):
+            raise MalformedHierarchyError(f"Children for node at {path!r} must be a list.")
+        return children
+
+    def _validate_level(self, level: str, *, path: str) -> None:
+        if level not in _ALLOWED_LEVELS:
+            raise MalformedHierarchyError(f"Unsupported hierarchy level {level!r} at {path!r}.")
+        if level == ROOT_LEVEL:
+            raise MalformedHierarchyError("Category is only allowed as the known hierarchy root.")
+
+    def _validate_transition(self, parent_level: str, child_level: str, *, path: str) -> None:
+        expected = _VALID_CHILD_LEVELS.get(parent_level)
+        if child_level != expected:
+            raise InvalidHierarchyTransitionError(
+                f"Invalid hierarchy transition at {path!r}: {parent_level!r} -> {child_level!r}; "
+                f"expected child level {expected!r}."
+            )
+
+    def _validate_children_transition(self, level: str, children: list[Any], *, path: str) -> None:
+        expected = _VALID_CHILD_LEVELS[level]
+        if expected is None and children:
+            raise InvalidHierarchyTransitionError(
+                f"Node at {path!r} has children, but {level!r} cannot have child concepts."
+            )
+        for child in children:
+            if not isinstance(child, dict):
+                raise MalformedHierarchyError(f"Child under {path!r} must be an object.")
+            child_level = child.get("level")
+            if not isinstance(child_level, str) or not child_level.strip():
+                raise MalformedHierarchyError(
+                    f"Child under {path!r} must contain non-empty 'level'."
+                )
+            if child_level != expected:
+                raise InvalidHierarchyTransitionError(
+                    f"Invalid hierarchy transition under {path!r}: {level!r} -> {child_level!r}; "
+                    f"expected child level {expected!r}."
+                )
+
+
+def is_assignable_level(level: str) -> bool:
+    """Return whether a source hierarchy level is assignable when UUID-bearing."""
+    return level in _ASSIGNABLE_LEVELS

tests/fixtures/gcmd_hierarchy_small.json

@@ -0,0 +1,72 @@
+{
+  "level": "Category",
+  "name": "EARTH SCIENCE",
+  "children": [
+    {
+      "level": "Topic",
+      "name": "ATMOSPHERE",
+      "UUID": "topic-atmosphere",
+      "definition": "Atmospheric science.",
+      "children": [
+        {
+          "level": "Term",
+          "name": "ATMOSPHERIC CHEMISTRY",
+          "UUID": "term-atmospheric-chemistry",
+          "definition": "Chemistry of the atmosphere.",
+          "children": [
+            {
+              "level": "Variable_Level_1",
+              "name": "CARBON",
+              "UUID": "vl1-atmosphere-carbon",
+              "definition": "Atmospheric carbon variables.",
+              "children": [
+                {
+                  "level": "Variable_Level_2",
+                  "name": "CARBON DIOXIDE",
+                  "UUID": "vl2-carbon-dioxide",
+                  "children": [
+                    {
+                      "level": "Variable_Level_3",
+                      "name": "CARBON DIOXIDE PROFILES",
+                      "UUID": "vl3-carbon-dioxide-profiles",
+                      "definition": "Vertical carbon dioxide profiles."
+                    }
+                  ]
+                },
+                {
+                  "level": "Variable_Level_2",
+                  "name": "METHANE",
+                  "UUID": "vl2-methane"
+                }
+              ]
+            }
+          ]
+        },
+        {
+          "level": "Term",
+          "name": "WEATHER EVENTS",
+          "UUID": "term-weather-events"
+        }
+      ]
+    },
+    {
+      "level": "Topic",
+      "name": "OCEANS",
+      "UUID": "topic-oceans",
+      "children": [
+        {
+          "level": "Term",
+          "name": "OCEAN CHEMISTRY",
+          "UUID": "term-ocean-chemistry",
+          "children": [
+            {
+              "level": "Variable_Level_1",
+              "name": "CARBON",
+              "UUID": "vl1-ocean-carbon"
+            }
+          ]
+        }
+      ]
+    }
+  ]
+}

tests/test_vocabulary.py

@@ -0,0 +1,259 @@
+from __future__ import annotations
+
+import copy
+import json
+from pathlib import Path
+
+import pytest
+
+from gcmd_classifier.errors import (
+    DuplicateCanonicalPathError,
+    DuplicateUUIDError,
+    InvalidHierarchyTransitionError,
+    MalformedHierarchyError,
+    VocabularyLookupError,
+)
+from gcmd_classifier.vocabulary import build_vocabulary_index, calculate_file_hash, load_vocabulary
+
+FIXTURE_PATH = Path("tests/fixtures/gcmd_hierarchy_small.json")
+FULL_HIERARCHY_PATH = Path("data/gcmd_hierarchy.json")
+
+
+def load_fixture_data() -> dict:
+    return json.loads(FIXTURE_PATH.read_text())
+
+
+def build_fixture_index():
+    return load_vocabulary(FIXTURE_PATH)
+
+
+def test_recursive_traversal_and_variable_depth_branches() -> None:
+    index = build_fixture_index()
+
+    assert len(index) == 10
+    assert index.root_level == "Category"
+    assert index.root_name == "EARTH SCIENCE"
+    assert index.children_of(None) == ("topic-atmosphere", "topic-oceans")
+    assert index.children_of("term-weather-events") == ()
+    assert index.get("vl3-carbon-dioxide-profiles").level == "Variable_Level_3"
+
+
+def test_uuid_preservation_and_record_fields() -> None:
+    index = build_fixture_index()
+    record = index.get("vl3-carbon-dioxide-profiles")
+
+    assert record.UUID == "vl3-carbon-dioxide-profiles"
+    assert record.name == "CARBON DIOXIDE PROFILES"
+    assert record.level == "Variable_Level_3"
+    assert record.topic == "ATMOSPHERE"
+    assert record.term == "ATMOSPHERIC CHEMISTRY"
+    assert record.path_components == (
+        "ATMOSPHERE",
+        "ATMOSPHERIC CHEMISTRY",
+        "CARBON",
+        "CARBON DIOXIDE",
+        "CARBON DIOXIDE PROFILES",
+    )
+    assert (
+        record.canonical_path == "ATMOSPHERE > ATMOSPHERIC CHEMISTRY > CARBON > CARBON DIOXIDE > "
+        "CARBON DIOXIDE PROFILES"
+    )
+    assert record.parent_uuid == "vl2-carbon-dioxide"
+    assert record.parent_name == "CARBON DIOXIDE"
+    assert record.child_uuids == ()
+    assert record.has_children is False
+    assert record.assignable is True
+    assert record.definition == "Vertical carbon dioxide profiles."
+    assert record.vocabulary_version == calculate_file_hash(FIXTURE_PATH)
+
+
+def test_topic_term_and_variable_records_are_assignable() -> None:
+    index = build_fixture_index()
+
+    topic = index.get("topic-atmosphere")
+    term = index.get("term-atmospheric-chemistry")
+    internal_variable = index.get("vl1-atmosphere-carbon")
+    leaf_variable = index.get("vl2-methane")
+
+    assert topic.level == "Topic"
+    assert term.level == "Term"
+    assert internal_variable.level == "Variable_Level_1"
+    assert leaf_variable.level == "Variable_Level_2"
+    assert topic.assignable is True
+    assert term.assignable is True
+    assert internal_variable.assignable is True
+    assert leaf_variable.assignable is True
+    assert internal_variable.has_children is True
+    assert leaf_variable.has_children is False
+
+
+def test_root_category_is_non_assignable_context() -> None:
+    index = build_fixture_index()
+
+    assert "EARTH SCIENCE" not in index.records_by_path
+    assert all(record.level != "Category" for record in index.records_by_uuid.values())
+    assert index.parent_of("topic-atmosphere") is None
+    assert index.get("topic-atmosphere").parent_name == "EARTH SCIENCE"
+
+
+def test_optional_and_missing_definitions() -> None:
+    index = build_fixture_index()
+
+    assert index.get("topic-atmosphere").definition == "Atmospheric science."
+    assert index.get("vl2-methane").definition is None
+
+
+def test_canonical_path_excludes_root_category() -> None:
+    index = build_fixture_index()
+
+    assert index.get("topic-atmosphere").canonical_path == "ATMOSPHERE"
+    assert index.get("term-weather-events").canonical_path == "ATMOSPHERE > WEATHER EVENTS"
+    assert not index.get("topic-atmosphere").canonical_path.startswith("EARTH SCIENCE")
+
+
+def test_identical_names_in_different_branches_are_allowed_when_paths_differ() -> None:
+    index = build_fixture_index()
+
+    atmosphere_carbon = index.get("vl1-atmosphere-carbon")
+    ocean_carbon = index.get("vl1-ocean-carbon")
+
+    assert atmosphere_carbon.name == ocean_carbon.name == "CARBON"
+    assert atmosphere_carbon.canonical_path != ocean_carbon.canonical_path
+
+
+def test_duplicate_uuid_detection() -> None:
+    data = load_fixture_data()
+    data["children"][1]["UUID"] = "topic-atmosphere"
+
+    with pytest.raises(DuplicateUUIDError):
+        build_vocabulary_index(data, vocabulary_version="test")
+
+
+def test_duplicate_canonical_path_detection() -> None:
+    data = load_fixture_data()
+    duplicate = copy.deepcopy(data["children"][0]["children"][1])
+    duplicate["UUID"] = "term-weather-events-duplicate"
+    data["children"][0]["children"].append(duplicate)
+
+    with pytest.raises(DuplicateCanonicalPathError):
+        build_vocabulary_index(data, vocabulary_version="test")
+
+
+@pytest.mark.parametrize("field", ["level", "name"])
+def test_missing_required_fields(field: str) -> None:
+    data = load_fixture_data()
+    del data["children"][0][field]
+
+    with pytest.raises(MalformedHierarchyError):
+        build_vocabulary_index(data, vocabulary_version="test")
+
+
+def test_unexpected_uuidless_non_root_node() -> None:
+    data = load_fixture_data()
+    del data["children"][0]["UUID"]
+
+    with pytest.raises(MalformedHierarchyError):
+        build_vocabulary_index(data, vocabulary_version="test")
+
+
+def test_valid_hierarchy_level_transitions() -> None:
+    index = build_fixture_index()
+
+    assert index.get("topic-atmosphere").child_uuids == (
+        "term-atmospheric-chemistry",
+        "term-weather-events",
+    )
+    assert index.get("term-atmospheric-chemistry").child_uuids == ("vl1-atmosphere-carbon",)
+    assert index.get("vl1-atmosphere-carbon").child_uuids == (
+        "vl2-carbon-dioxide",
+        "vl2-methane",
+    )
+    assert index.get("vl2-carbon-dioxide").child_uuids == ("vl3-carbon-dioxide-profiles",)
+
+
+def test_invalid_hierarchy_level_transition() -> None:
+    data = load_fixture_data()
+    data["children"][0]["children"][0]["level"] = "Variable_Level_1"
+
+    with pytest.raises(InvalidHierarchyTransitionError):
+        build_vocabulary_index(data, vocabulary_version="test")
+
+
+def test_parent_and_direct_child_lookup() -> None:
+    index = build_fixture_index()
+
+    assert index.parent_of("vl2-carbon-dioxide") == "vl1-atmosphere-carbon"
+    assert index.children_of("vl1-atmosphere-carbon") == (
+        "vl2-carbon-dioxide",
+        "vl2-methane",
+    )
+    with pytest.raises(VocabularyLookupError):
+        index.parent_of("missing-uuid")
+
+
+def test_topic_to_term_lookup() -> None:
+    index = build_fixture_index()
+
+    terms = index.terms_for_topic("topic-atmosphere")
+    assert tuple(term.UUID for term in terms) == (
+        "term-atmospheric-chemistry",
+        "term-weather-events",
+    )
+    with pytest.raises(VocabularyLookupError):
+        index.terms_for_topic("term-weather-events")
+
+
+def test_term_to_variable_lookup() -> None:
+    index = build_fixture_index()
+
+    variables = index.variables_for_parent("term-atmospheric-chemistry")
+    assert tuple(variable.UUID for variable in variables) == ("vl1-atmosphere-carbon",)
+    assert index.variables_for_parent("vl2-methane") == ()
+    with pytest.raises(VocabularyLookupError):
+        index.variables_for_parent("topic-atmosphere")
+
+
+def test_ancestor_and_descendant_checks() -> None:
+    index = build_fixture_index()
+
+    assert index.is_ancestor("topic-atmosphere", "vl3-carbon-dioxide-profiles") is True
+    assert index.is_descendant("vl3-carbon-dioxide-profiles", "topic-atmosphere") is True
+    assert index.is_ancestor("term-weather-events", "vl3-carbon-dioxide-profiles") is False
+    assert index.ancestors_of("vl3-carbon-dioxide-profiles") == (
+        "vl2-carbon-dioxide",
+        "vl1-atmosphere-carbon",
+        "term-atmospheric-chemistry",
+        "topic-atmosphere",
+    )
+
+
+def test_vocabulary_file_hash_stability() -> None:
+    assert calculate_file_hash(FIXTURE_PATH) == calculate_file_hash(FIXTURE_PATH)
+    assert build_fixture_index().vocabulary_version == calculate_file_hash(FIXTURE_PATH)
+
+
+def test_source_file_immutability() -> None:
+    before = FULL_HIERARCHY_PATH.read_bytes()
+    load_vocabulary(FULL_HIERARCHY_PATH)
+    after = FULL_HIERARCHY_PATH.read_bytes()
+
+    assert after == before
+
+
+def test_every_current_topic_and_term_has_uuid() -> None:
+    index = load_vocabulary(FULL_HIERARCHY_PATH)
+
+    assert all(topic.UUID for topic in index.topics())
+    assert all(term.UUID for topic in index.topics() for term in index.terms_for_topic(topic.UUID))
+
+
+def test_full_data_smoke_counts_for_current_vocabulary() -> None:
+    index = load_vocabulary(FULL_HIERARCHY_PATH)
+    counts = index.count_by_level()
+
+    assert counts["Topic"] == 14
+    assert counts["Term"] == 144
+    assert counts["Variable_Level_1"] == 1368
+    assert counts["Variable_Level_2"] == 1456
+    assert counts["Variable_Level_3"] == 553
+    assert len(index) == 3535