diff --git a/.github/codecov.yml b/.github/codecov.yml
new file mode 100644
index 0000000..260abd1
--- /dev/null
+++ b/.github/codecov.yml
@@ -0,0 +1,22 @@
+coverage:
+  status:
+    project:
+      default:
+        # Overall coverage may not drop by more than the threshold.
+        target: auto
+        threshold: 1%
+    patch:
+      default:
+        # New and modified lines in a PR must be 80% covered.
+        target: 80%
+
+comment:
+  layout: "reach, diff, flags, files"
+  behavior: default
+  require_changes: false
+
+ignore:
+  - "tests/**"
+  - "docs/**"
+  - "build/**"
+  - "site/**"
diff --git a/src/DataBUS/AeDNAAssay.py b/src/DataBUS/AeDNAAssay.py
new file mode 100644
index 0000000..21f0330
--- /dev/null
+++ b/src/DataBUS/AeDNAAssay.py
@@ -0,0 +1,38 @@
+from .AeDNAEntity import AeDNAEntity
+
+AEDNAASSAY_PARAMS = [
+    "datasetid",
+    "assaytypeid",
+    "assayname",
+    "targettaxonomicassay",
+    "targetgene",
+    "subfragment",
+    "ampliconsize",
+    "pcrprimerforward",
+    "pcrprimerreverse",
+    "pcrprimernameforward",
+    "pcrprimernamereverse",
+    "pcrprimerreferenceforward",
+    "pcrprimerreferencereverse",
+    "probeseq",
+    "proberef",
+    "probereporter",
+    "probequencher",
+    "probeconc",
+]
+
+
+class AeDNAAssay(AeDNAEntity):
+    """An aeDNA assay record (ndb.aednaassays).
+
+    Holds the PCR/assay configuration for a dataset: assay name and type,
+    target gene/taxon, primer sequences and probe details.
+    """
+
+    _TABLE = "ndb.aednaassays"
+    _PARAMS = AEDNAASSAY_PARAMS
+    _PK = "assayid"
+    _INT_FIELDS = ("datasetid", "assaytypeid", "ampliconsize")
+
+    def __str__(self):
+        return f"AeDNAAssay(assayid={self.assayid}, datasetid={self.datasetid}, assayname={self.assayname})"
diff --git a/src/DataBUS/AeDNAEntity.py b/src/DataBUS/AeDNAEntity.py
new file mode 100644
index 0000000..239a7c6
--- /dev/null
+++ b/src/DataBUS/AeDNAEntity.py
@@ -0,0 +1,45 @@
+from .neotomaHelpers.utils import validate_int_values, validate_str_values
+
+
+class AeDNAEntity:
+    """Base for the flat aeDNA tables (ndb.aednaassays, ndb.aednalibraries).
+
+    Subclasses declare four class attributes:
+
+    * ``_TABLE`` — fully-qualified table name.
+    * ``_PARAMS`` — ordered list of insert columns.
+    * ``_PK`` — primary-key column name (returned by the insert).
+    * ``_INT_FIELDS`` — the ``_PARAMS`` that are integer FKs/sizes; every other
+      param is validated as a string.
+
+    Construction validates each field and stores it as an attribute; ``_PK`` is
+    initialised to ``None`` and set after :meth:`insert_to_db`.
+    """
+
+    _TABLE: str = ""
+    _PARAMS: list[str] = []
+    _PK: str = ""
+    _INT_FIELDS: tuple[str, ...] = ()
+
+    def __init__(self, **fields):
+        setattr(self, self._PK, None)
+        for name in self._PARAMS:
+            value = fields.get(name)
+            if name in self._INT_FIELDS:
+                value = validate_int_values(value, name)
+            else:
+                value = validate_str_values(value, name)
+            setattr(self, name, value)
+
+    def insert_to_db(self, cur):
+        """Insert the record into ``_TABLE`` and return the new ``_PK`` value."""
+        cols = self._PARAMS
+        query = (
+            f"INSERT INTO {self._TABLE} ({', '.join(cols)}) "
+            f"VALUES ({', '.join('%(' + c + ')s' for c in cols)}) "
+            f"RETURNING {self._PK};"
+        )
+        cur.execute(query, {c: getattr(self, c) for c in cols})
+        pk = cur.fetchone()[0]
+        setattr(self, self._PK, pk)
+        return pk
diff --git a/src/DataBUS/AeDNALibrary.py b/src/DataBUS/AeDNALibrary.py
new file mode 100644
index 0000000..0bb273f
--- /dev/null
+++ b/src/DataBUS/AeDNALibrary.py
@@ -0,0 +1,30 @@
+from .AeDNAEntity import AeDNAEntity
+
+AEDNALIBRARY_PARAMS = [
+    "datasetid",
+    "assayid",
+    "libid",
+    "seqrunid",
+    "pcrplateid",
+    "barcodingpcrappr",
+    "platform",
+    "instrument",
+    "seqkit",
+    "libscreen",
+]
+
+
+class AeDNALibrary(AeDNAEntity):
+    """An aeDNA sequencing library record (ndb.aednalibraries).
+
+    Holds the library-preparation and sequencing metadata for a dataset's
+    assay: platform, instrument, kit, and library/run identifiers.
+    """
+
+    _TABLE = "ndb.aednalibraries"
+    _PARAMS = AEDNALIBRARY_PARAMS
+    _PK = "libraryid"
+    _INT_FIELDS = ("datasetid", "assayid")
+
+    def __str__(self):
+        return f"AeDNALibrary(libraryid={self.libraryid}, datasetid={self.datasetid}, assayid={self.assayid})"
diff --git a/src/DataBUS/AeDNAModel.py b/src/DataBUS/AeDNAModel.py
index ba99893..3d27e4d 100644
--- a/src/DataBUS/AeDNAModel.py
+++ b/src/DataBUS/AeDNAModel.py
@@ -1,6 +1,13 @@
 from .neotomaHelpers.utils import validate_int_values
 
-AEDNAMODEL_PARAMS = ["sequenceid", "taxonid", "model"]
+AEDNAMODEL_PARAMS = [
+    "sequenceid",
+    "taxonid",
+    "model",
+    "supersededbymodelid",
+    "publicationid",
+    "notes",
+]
 
 
 class AeDNAModel:
@@ -75,21 +82,21 @@ def insert_to_db(self, cur):
         self.modelid = cur.fetchone()[0]
         return self.modelid
 
-    def supersede_previous(self, cur, superseeds_list):
+    def supersede_previous(self, cur, supersedes_list):
         """Mark older aeDNA model entries as superseded by this model.
 
-        For each model name in superseeds_list, finds existing aednamodels rows
+        For each model name in supersedes_list, finds existing aednamodels rows
         that share this taxonid and that model name, then sets their
         supersededbymodelid to this model's modelid.
 
         Args:
             cur (psycopg2.cursor): Database cursor for executing queries.
-            superseeds_list (list[str]): Model names that this model supersedes.
+            supersedes_list (list[str]): Model names that this model supersedes.
 
         Returns:
             int: Total number of rows updated.
         """
-        if not superseeds_list or self.modelid is None:
+        if not supersedes_list or self.modelid is None:
             return 0
         update_q = """
             UPDATE ndb.aednamodels
@@ -100,7 +107,7 @@ def supersede_previous(self, cur, superseeds_list):
               AND modelid != %(new_modelid)s;
         """
         total_updated = 0
-        for old_model in superseeds_list:
+        for old_model in supersedes_list:
             cur.execute(
                 update_q,
                 {
diff --git a/src/DataBUS/Project.py b/src/DataBUS/Project.py
index 9aad362..da492e0 100644
--- a/src/DataBUS/Project.py
+++ b/src/DataBUS/Project.py
@@ -7,6 +7,10 @@
     "projectenddate",
 ]
 
+GRANT_PARAMS = ["grantname", "grantnumber", "dateawarded", "dateended"]
+
+INSTITUTION_PARAMS = ["institutionid", "institutionname", "institutionlocation"]
+
 
 class Project:
     """A research project record in the Neotoma database.
@@ -50,7 +54,7 @@ def insert_to_db(self, cur):
         Returns:
             int: The projectid assigned by the database.
         """
-        q = """
+        project_q = """
             INSERT INTO ndb.projects
                 (parentprojectid, projectname, projectdescription,
                  projectstartdate, projectenddate)
@@ -66,7 +70,7 @@ def insert_to_db(self, cur):
             "projectstartdate": self.projectstartdate,
             "projectenddate": self.projectenddate,
         }
-        cur.execute(q, inputs)
+        cur.execute(project_q, inputs)
         self.projectid = cur.fetchone()[0]
         return self.projectid
 
@@ -113,7 +117,7 @@ def insert_to_db(self, cur):
         Returns:
             int: The grantid assigned by the database.
         """
-        q = """
+        grant_q = """
             INSERT INTO ndb.grants (grantname, grantnumber, dateawarded, dateended)
             VALUES (%(grantname)s, %(grantnumber)s, %(dateawarded)s, %(dateended)s)
             RETURNING grantid;
@@ -124,7 +128,7 @@ def insert_to_db(self, cur):
             "dateawarded": self.dateawarded,
             "dateended": self.dateended,
         }
-        cur.execute(q, inputs)
+        cur.execute(grant_q, inputs)
         self.grantid = cur.fetchone()[0]
         return self.grantid
 
@@ -175,7 +179,7 @@ def insert_to_db(self, cur):
         Returns:
             str: The institutionid.
         """
-        q = """
+        institution_q = """
             INSERT INTO ndb.institutions (institutionid, institutionname, institutionlocation)
             VALUES (%(institutionid)s, %(institutionname)s, %(institutionlocation)s)
             ON CONFLICT (institutionid) DO UPDATE
@@ -188,7 +192,7 @@ def insert_to_db(self, cur):
             "institutionname": self.institutionname,
             "institutionlocation": self.institutionlocation,
         }
-        cur.execute(q, inputs)
+        cur.execute(institution_q, inputs)
         self.institutionid = cur.fetchone()[0]
         return self.institutionid
 
@@ -196,9 +200,6 @@ def __str__(self):
         return f"Institution(institutionid={self.institutionid}, name={self.institutionname})"
 
 
-# ── Junction table helpers ────────────────────────────────────────────────────
-
-
 def insert_project_dataset(cur, projectid, datasetid):
     """Link a project to a dataset via ndb.projectdatasets."""
     q = """
diff --git a/src/DataBUS/Response.py b/src/DataBUS/Response.py
index 7a648b3..871a063 100644
--- a/src/DataBUS/Response.py
+++ b/src/DataBUS/Response.py
@@ -26,6 +26,8 @@ class Response:
         id_dict (dict): Mapping of data identifiers.
         name (dict): Name mapping dictionary.
         indices (list): List of indices.
+        matched (dict): Name/location match flags from close-record searches.
+        doublematched (bool): True when both name and location matched.
     """
 
     def __init__(self):
@@ -40,6 +42,8 @@ def __init__(self):
         self.name = {}
         self.indices = []
         self.counter = 0
+        self.matched = {}
+        self.doublematched = False
 
     @property
     def validAll(self):
diff --git a/src/DataBUS/__init__.py b/src/DataBUS/__init__.py
index 0105028..47cae40 100644
--- a/src/DataBUS/__init__.py
+++ b/src/DataBUS/__init__.py
@@ -16,6 +16,10 @@
 import logging
 import os
 
+from .AeDNAAssay import AeDNAAssay
+from .AeDNAEntity import AeDNAEntity
+from .AeDNALibrary import AeDNALibrary
+from .AeDNAModel import AeDNAModel
 from .AnalysisUnit import AnalysisUnit
 from .ChronControl import ChronControl
 from .Chronology import Chronology
@@ -30,9 +34,12 @@
 from .Geog import Geog, WrongCoordinates
 from .Hiatus import Hiatus
 from .LeadModel import LeadModel
+from .Project import Grant, Institution, Project
 from .Response import Response
 from .Sample import Sample
 from .SampleAge import SampleAge
+from .Sequence import Sequence
+from .SequenceData import SequenceData
 from .Site import Site
 from .Speleothem import ExternalSpeleothem, Speleothem
 from .UThSeries import UThSeries, insert_uraniumseriesdata
diff --git a/src/DataBUS/neotomaHelpers/__init__.py b/src/DataBUS/neotomaHelpers/__init__.py
index 0948ada..70bac0a 100644
--- a/src/DataBUS/neotomaHelpers/__init__.py
+++ b/src/DataBUS/neotomaHelpers/__init__.py
@@ -18,5 +18,5 @@
 from .read_csv import read_csv, read_xlsx
 from .safe_step import safe_step
 from .template_to_dict import template_to_dict
-from .utils import convert_to_bp, retrieve_dict
+from .utils import convert_to_bool, convert_to_bp, retrieve_dict
 from .write_csv import write_csv
diff --git a/src/DataBUS/neotomaHelpers/pull_params.py b/src/DataBUS/neotomaHelpers/pull_params.py
index fad2b95..9ec8e5d 100644
--- a/src/DataBUS/neotomaHelpers/pull_params.py
+++ b/src/DataBUS/neotomaHelpers/pull_params.py
@@ -57,8 +57,16 @@ def _process_value_entry(param_name, val_entry, csv_template, table, add_unit_in
         clean_valor = ut.clean_column(
             val_entry.get("column"), template_rows, clean=not val_entry.get("rowwise")
         )
-    except KeyError:
-        return
+    except (KeyError, TypeError):
+        # The named column is not in the data at all. An entry carrying a
+        # `value:` constant supplies the datum directly instead — that is how
+        # sheets with a non-machine-readable header (projectMetadata and
+        # friends) are read. Without such a constant the parameter is genuinely
+        # absent, as before. Note this deliberately does not cover a column that
+        # exists but is blank: falling back there would mask missing data.
+        if val_entry.get("value") is None:
+            return
+        clean_valor = val_entry["value"]
     if not clean_valor:
         if "taxonname" not in val_entry:
             add_unit_inputs[param_name] = None
diff --git a/src/DataBUS/neotomaHelpers/read_csv.py b/src/DataBUS/neotomaHelpers/read_csv.py
index 23ab18d..614db89 100644
--- a/src/DataBUS/neotomaHelpers/read_csv.py
+++ b/src/DataBUS/neotomaHelpers/read_csv.py
@@ -8,12 +8,14 @@ def read_xlsx(filename, num_headers=1):
     """Read an Excel file and return a dict mapping sheet names to rows.
 
     Each sheet is parsed into a list of dictionaries using the header row(s)
-    as column names. When ``num_headers >= 2``, sheets whose second row
-    contains at least one ``None`` value are treated as having multi-row
-    headers: the first two rows are joined with ``_`` to form combined
-    column names (e.g. ``TaxonName`` + ``ASV1`` → ``TaxonName_ASV1``).
-    Sheets where the second row has no ``None`` values are assumed to have
-    a single header row (the second row is treated as data).
+    as column names. FAIRe sheets (3-row header: requirement level, section,
+    ``term_name``) are detected per sheet and read from the ``term_name`` row.
+    Otherwise, when ``num_headers >= 2``, sheets whose second row contains at
+    least one ``None`` value are treated as having multi-row headers: the first
+    two rows are joined with ``_`` to form combined column names (e.g.
+    ``TaxonName`` + ``ASV1`` → ``TaxonName_ASV1``). Sheets where the second row
+    has no ``None`` values are assumed to have a single header row (the second
+    row is treated as data).
 
     Examples:
         >>> read_xlsx('data.xlsx')  # doctest: +SKIP
@@ -44,11 +46,14 @@ def read_xlsx(filename, num_headers=1):
         if not rows:
             result[sheet_name] = []
             continue
-        if num_headers >= 2 and len(rows) >= 2 and any(v is None for v in rows[1]):
+        if _is_faire_header(rows):
+            headers = _single_header(rows[2])
+            data_start = 3
+        elif num_headers >= 2 and len(rows) >= 2 and any(v is None for v in rows[1]):
             headers = _combine_header_rows(rows[0], rows[1])
-            data_start = num_headers
+            data_start = _stacked_data_start(rows)
         else:
-            headers = [str(h) if h is not None else f"col_{i}" for i, h in enumerate(rows[0])]
+            headers = _single_header(rows[0])
             data_start = 1
         result[sheet_name] = [
             dict(zip(headers, row, strict=False))
@@ -83,6 +88,46 @@ def _combine_header_rows(row0, row1):
     return headers
 
 
+def _single_header(row):
+    """Turn one header row into column names, filling blanks with ``col_<i>``."""
+    return _combine_header_rows(row, [None] * len(row))
+
+
+def _stacked_data_start(rows):
+    """First data-row index for a stacked taxon header (name / ASV / sequence).
+
+    Sub-header rows only populate the taxon columns — the same positions that
+    are ``None`` in row 1. The first row that also fills a non-taxon column is
+    the first data row. Falls back to ``len(rows)`` when none qualifies.
+    """
+    sparse = [i for i, v in enumerate(rows[1]) if v is None]
+    for idx in range(1, len(rows)):
+        if not all(rows[idx][i] is None for i in sparse):
+            return idx
+    return len(rows)
+
+
+_FAIRE_LEVELS = {"m", "o", "r", "hr"}
+
+
+def _is_faire_header(rows):
+    """Detect a 3-row FAIRe header (requirement level / section / term_name).
+
+    True when the sheet has at least three rows and the first row is a row of
+    FAIRe requirement-level codes (``M``/``O``/``R``/``HR``), optionally led by
+    the ``requirement_level_code`` label cell. The real column names then live
+    on the third row (``term_name``).
+    """
+    if len(rows) < 3:
+        return False
+    values = [str(v).strip() for v in rows[0] if v is not None]
+    if not values:
+        return False
+    if values[0].lstrip("# ").lower() == "requirement_level_code":
+        values = values[1:]
+    return bool(values) and all(v.lstrip("# ").lower() in _FAIRE_LEVELS for v in values)
+
+
 def read_csv(filename):
     """Read CSV file and return a structured list of dictionaries.
 
diff --git a/src/DataBUS/neotomaHelpers/utils.py b/src/DataBUS/neotomaHelpers/utils.py
index b3a4426..ee2ab2f 100644
--- a/src/DataBUS/neotomaHelpers/utils.py
+++ b/src/DataBUS/neotomaHelpers/utils.py
@@ -60,8 +60,19 @@ def _convert_coordinates(value):
     return [float(num) for num in value[0].split(",")]
 
 
-def _convert_bool(value, is_rowwise):
-    """Convert strings to booleans, treating 'true'/'false'/'1'/'0' case-insensitively."""
+def convert_to_bool(value, is_rowwise=False):
+    """Convert strings to booleans, treating 'true'/'1'/'yes' case-insensitively.
+
+    ``None`` is passed through as ``None`` so callers can tell "not recorded"
+    apart from "recorded as false".
+
+    Args:
+        value: A single value, or a list of them when ``is_rowwise`` is True.
+        is_rowwise (bool): Whether ``value`` is a per-row list.
+
+    Returns:
+        bool | None | list: The converted value(s).
+    """
 
     def _to_bool(v):
         if v is None:
@@ -73,6 +84,10 @@ def _to_bool(v):
     return [_to_bool(v) for v in value] if is_rowwise else _to_bool(value)
 
 
+# Retained for the type-dispatch table in convert_value_by_type.
+_convert_bool = convert_to_bool
+
+
 def _convert_string(value, is_rowwise):
     """Convert value(s) to strings, handling NA and empty values."""
     converted = [str(v) if v is not None else None for v in value] if is_rowwise else str(value)
@@ -358,6 +373,13 @@ def validate_int_values(value, name: str) -> int | None:
     return int(value) if value is not None else None
 
 
+def validate_str_values(value, name: str) -> str | None:
+    """Validates that a value is a str or None, returning it unchanged."""
+    if value is not None and not isinstance(value, str):
+        raise TypeError(f"✗ {name} must be a string or None.")
+    return value
+
+
 def validate_date_values(value, name: str) -> datetime.date | None:
     """Validates that a value is an date or None, returning it as date or None."""
     if isinstance(value, list):
diff --git a/src/DataBUS/neotomaValidator/__init__.py b/src/DataBUS/neotomaValidator/__init__.py
index 21767f9..f20737a 100644
--- a/src/DataBUS/neotomaValidator/__init__.py
+++ b/src/DataBUS/neotomaValidator/__init__.py
@@ -18,7 +18,9 @@
 import DataBUS.neotomaHelpers
 
 from .insert_final import insert_final
+from .valid_aednalibraries import valid_aednalibraries
 from .valid_analysisunit import valid_analysisunit
+from .valid_assays import valid_assays
 from .valid_chroncontrols import valid_chroncontrols
 from .valid_chronologies import valid_chronologies
 from .valid_collunit import valid_collunit
diff --git a/src/DataBUS/neotomaValidator/_aedna_entity_validator.py b/src/DataBUS/neotomaValidator/_aedna_entity_validator.py
new file mode 100644
index 0000000..6c4b9fb
--- /dev/null
+++ b/src/DataBUS/neotomaValidator/_aedna_entity_validator.py
@@ -0,0 +1,59 @@
+import DataBUS.neotomaHelpers as nh
+from DataBUS import Response
+
+
+def validate_aedna_entity(cur, yml_dict, cls, databus, links, present_fields, resolve=None):
+    """Validate and insert one flat aeDNA entity (assay or library).
+
+    Reads ``cls``'s ``value:`` constants from the template (via ``retrieve_dict``,
+    as ``valid_dataset`` does). If none of *present_fields* is set, returns an
+    empty-but-valid Response (nothing to do). Otherwise runs the optional
+    *resolve* hook, fills the *links* foreign keys from ``databus``, builds
+    ``cls(**inputs)``, inserts it, and stores the new primary key in
+    ``response.id_int``.
+
+    Args:
+        cur (cursor): Database cursor.
+        yml_dict (dict): Parsed template.
+        cls (type): An :class:`AeDNAEntity` subclass (``AeDNAAssay``/``AeDNALibrary``).
+        databus (dict | None): Prior validation results supplying parent ids.
+        links (dict): ``{param: databus_key}`` — ``databus[key].id_int`` fills the FK.
+        present_fields (list): Params whose presence means the entity was described.
+        resolve (callable | None): Optional ``(cur, inputs, response)`` pre-insert hook.
+
+    Returns:
+        Response: Validation result with the inserted id in ``id_int``.
+    """
+    response = Response()
+    label = cls.__name__
+    inputs = {}
+    for param in cls._PARAMS:
+        matches = nh.retrieve_dict(yml_dict, f"{cls._TABLE}.{param}")
+        inputs[param] = next((m.get("value") for m in matches if m.get("value") is not None), None)
+
+    if not any(inputs.get(field) for field in present_fields):
+        response.valid.append(True)
+        response.message.append(f"? No {label} entries found in template.")
+        return response
+
+    if resolve is not None:
+        resolve(cur, inputs, response)
+
+    for param, key in links.items():
+        try:
+            inputs[param] = databus[key].id_int
+        except Exception as e:
+            inputs[param] = None
+            response.valid.append(False)
+            response.message.append(f"✗ {key} ID not available for {label}: {e}")
+
+    try:
+        entity = cls(**inputs)
+        response.valid.append(True)
+        response.message.append(f"✔ {label} can be created.")
+        response.id_int = entity.insert_to_db(cur)
+        response.message.append(f"✔ {label} inserted with ID {response.id_int}.")
+    except Exception as e:
+        response.valid.append(False)
+        response.message.append(f"✗ {label} cannot be created: {e}")
+    return response
diff --git a/src/DataBUS/neotomaValidator/valid_aednalibraries.py b/src/DataBUS/neotomaValidator/valid_aednalibraries.py
new file mode 100644
index 0000000..fa25b0c
--- /dev/null
+++ b/src/DataBUS/neotomaValidator/valid_aednalibraries.py
@@ -0,0 +1,24 @@
+from DataBUS import AeDNALibrary
+from DataBUS.AeDNALibrary import AEDNALIBRARY_PARAMS
+
+from ._aedna_entity_validator import validate_aedna_entity
+
+_LINKS = {"datasetid": "datasets", "assayid": "assays"}
+_PRESENT = [p for p in AEDNALIBRARY_PARAMS if p not in _LINKS]
+
+
+def valid_aednalibraries(cur, yml_dict, csv_file, databus=None):
+    """Validate and insert the dataset's aeDNA library (ndb.aednalibraries).
+
+    Reads the library ``value:`` constants (platform, instrument, kit,
+    identifiers), links the dataset and the freshly inserted assay, and inserts
+    one ``AeDNALibrary``.
+    """
+    return validate_aedna_entity(
+        cur,
+        yml_dict,
+        AeDNALibrary,
+        databus,
+        links=_LINKS,
+        present_fields=_PRESENT,
+    )
diff --git a/src/DataBUS/neotomaValidator/valid_assays.py b/src/DataBUS/neotomaValidator/valid_assays.py
new file mode 100644
index 0000000..2be9091
--- /dev/null
+++ b/src/DataBUS/neotomaValidator/valid_assays.py
@@ -0,0 +1,39 @@
+from DataBUS import AeDNAAssay
+
+from ._aedna_entity_validator import validate_aedna_entity
+
+
+def _resolve_assaytype(cur, inputs, response):
+    """Resolve the assay type name in ``assaytypeid`` to its ndb.assaytypes id."""
+    if isinstance(inputs.get("assaytypeid"), str):
+        cur.execute(
+            "SELECT assaytypeid FROM ndb.assaytypes WHERE LOWER(assaytype) = %(t)s",
+            {"t": inputs["assaytypeid"].lower().strip()},
+        )
+        row = cur.fetchone()
+        if row:
+            inputs["assaytypeid"] = row[0]
+            response.valid.append(True)
+            response.message.append(f"✔ Assay type resolved to assaytypeid {row[0]}.")
+        else:
+            inputs["assaytypeid"] = None
+            response.valid.append(False)
+            response.message.append("✗ Assay type is not known to Neotoma.")
+
+
+def valid_assays(cur, yml_dict, csv_file, databus=None):
+    """Validate and insert the dataset's aeDNA assay (ndb.aednaassays).
+
+    Reads the assay ``value:`` constants from the template, resolves the assay
+    type name to an id, links the dataset, and inserts one ``AeDNAAssay``. The
+    new assayid lands in ``response.id_int`` for ``valid_aednalibraries``.
+    """
+    return validate_aedna_entity(
+        cur,
+        yml_dict,
+        AeDNAAssay,
+        databus,
+        links={"datasetid": "datasets"},
+        present_fields=["assayname"],
+        resolve=_resolve_assaytype,
+    )
diff --git a/src/DataBUS/neotomaValidator/valid_data.py b/src/DataBUS/neotomaValidator/valid_data.py
index a01e5fa..004c4f6 100644
--- a/src/DataBUS/neotomaValidator/valid_data.py
+++ b/src/DataBUS/neotomaValidator/valid_data.py
@@ -68,6 +68,8 @@ def valid_data(cur, yml_dict, csv_file, databus=None):
     else:
         data = {k: v for k, v in inputs2.items() if v is not None}
         for key in inputs:
+            if inputs[key] is None:  # base-param placeholder from pull_params
+                continue
             # For compound keys (taxonname::asv), extract the real taxon name
             real_taxon = key.split("::")[0] if "::" in key else key
             data[key]["_entry_key"] = [key] * len(inputs[key]["value"])
@@ -102,6 +104,13 @@ def valid_data(cur, yml_dict, csv_file, databus=None):
         txname = entry_key if entry_key else datum.get("taxonid")
         if txname not in response.id_dict:
             response.id_dict[txname] = []
+        # A measurement that was not taken for this sample is not a datum. The
+        # all-None check above only drops a taxon measured nowhere; a taxon
+        # measured for some samples still yields empty rows for the rest, and
+        # those carry no units either, so the variable lookup would fail on a
+        # row that should never have existed.
+        if datum.get("value") is None:
+            continue
         for param, (query, key) in par.items():
             if isinstance(datum.get(param), str) and datum[param].strip().lower() == "none":
                 datum[param] = None
diff --git a/src/DataBUS/neotomaValidator/valid_project.py b/src/DataBUS/neotomaValidator/valid_project.py
index d5640e9..9fb6a51 100644
--- a/src/DataBUS/neotomaValidator/valid_project.py
+++ b/src/DataBUS/neotomaValidator/valid_project.py
@@ -132,6 +132,7 @@ def valid_project(cur, yml_dict, csv_file, databus=None):
             try:
                 insert_project_participant(cur, projectid, contactid)
                 response.valid.append(True)
+                response.message.append(f"✔ Participant '{name}' linked (contactid {contactid}).")
             except Exception as e:
                 response.message.append(f"? Could not add participant '{name}': {e}")
         else:
@@ -260,15 +261,22 @@ def _extract_contact_institutions(yml_dict):
 
 
 def _resolve_contact(cur, name):
-    """Look up a contactid by name. Returns int or None."""
+    """Look up a contactid via the same strategy as PI lookup (nh.get_contacts).
+
+    Falls back to reordering a "Given [M.] Surname" string into Neotoma's
+    "Surname, Given" convention when the first lookup misses.
+    """
     if not name:
         return None
-    cur.execute(
-        "SELECT contactid FROM ndb.contacts WHERE LOWER(contactname) = %(name)s;",
-        {"name": name.lower().strip()},
-    )
-    result = cur.fetchone()
-    return result[0] if result else None
+    from DataBUS.neotomaHelpers.get_contacts import get_contacts
+
+    name = name.strip()
+    contact = get_contacts(cur, name)
+    if contact.get("id") is None and "," not in name:
+        parts = name.split()
+        if len(parts) >= 2:
+            contact = get_contacts(cur, f"{parts[-1]}, {parts[0]}")
+    return contact.get("id")
 
 
 def _resolve_keyword(cur, keyword):
diff --git a/src/DataBUS/neotomaValidator/valid_sample.py b/src/DataBUS/neotomaValidator/valid_sample.py
index 3d20f3f..fd87ef4 100644
--- a/src/DataBUS/neotomaValidator/valid_sample.py
+++ b/src/DataBUS/neotomaValidator/valid_sample.py
@@ -55,6 +55,20 @@ def valid_sample(cur, yml_dict, csv_file, databus):
         response.message.append(f"✗ Error pulling sample parameters: {e}")
         response.valid.append(False)
         return response
+    # The zip below pairs each sample name with an analysis unit purely by
+    # position, and ``strict=False`` makes a length mismatch truncate in silence
+    # — samples go missing and the survivors keep names belonging to other rows.
+    # Refuse to guess when the row counts disagree.
+    row_counts = {k: len(v) for k, v in inputs.items() if isinstance(v, list)}
+    if len(set(row_counts.values())) > 1:
+        response.valid.append(False)
+        response.message.append(
+            "✗ Sample parameters have mismatched row counts: "
+            + ", ".join(f"{k}={n}" for k, n in sorted(row_counts.items()))
+            + ". Every rowwise column must cover the same samples, in the same order."
+        )
+        return response
+
     response.counter = 0
     get_taxonid = """SELECT taxonid FROM ndb.taxa
                      WHERE LOWER(taxonname) %% %(taxonname)s;"""
diff --git a/src/DataBUS/neotomaValidator/valid_sample_age.py b/src/DataBUS/neotomaValidator/valid_sample_age.py
index a026bfd..6718853 100644
--- a/src/DataBUS/neotomaValidator/valid_sample_age.py
+++ b/src/DataBUS/neotomaValidator/valid_sample_age.py
@@ -62,7 +62,12 @@ def valid_sample_age(cur, yml_dict, csv_file, databus=None):
     for chron in inputs:
         sa = inputs[chron]
         sa["sampleid"] = sample_ids
-        sa = {k: v if isinstance(v, list) else [v] for k, v in sa.items()}
+        # Chronology-wide values (agemodel, for one) arrive as scalars alongside
+        # the per-sample age lists. Wrapping them as one-element lists would make
+        # the zip below stop after a single row and silently write one sample age
+        # for the whole dataset, so broadcast them across every sample instead.
+        n_rows = max((len(v) for v in sa.values() if isinstance(v, list)), default=1)
+        sa = {k: v if isinstance(v, list) else [v] * n_rows for k, v in sa.items()}
         chronologyid = chron_id_map.get(chron)
         for _, row in enumerate(zip(*sa.values(), strict=False)):
             sa_age = dict(zip(sa.keys(), row, strict=False))
diff --git a/src/DataBUS/neotomaValidator/valid_speleothem.py b/src/DataBUS/neotomaValidator/valid_speleothem.py
index 53e2cbc..3fe8727 100644
--- a/src/DataBUS/neotomaValidator/valid_speleothem.py
+++ b/src/DataBUS/neotomaValidator/valid_speleothem.py
@@ -85,10 +85,10 @@ def valid_speleothem(cur, yml_dict, csv_file, databus=None):
         response.valid.append(True)
         response.message.append("✔ No speleothem parameters provided.")
         return response
-    if inputs.get("monitoring", "").lower() == "yes":
-        inputs["monitoring"] = True
-    else:
-        inputs["monitoring"] = False
+    # pull_params reports an absent column as an explicit None rather than by
+    # omitting the key, so a `.get(..., "")` default would never fire.  Reuse the
+    # shared coercion and collapse the "not recorded" case to False.
+    inputs["monitoring"] = bool(nh.convert_to_bool(inputs.get("monitoring"), is_rowwise=False))
     if isinstance(inputs.get("ref_id"), str):
         inputs["ref_id"] = list(map(int, inputs.get("ref_id", []).split(",")))
     for inp in inputs:
diff --git a/tests/test_helpers.py b/tests/test_helpers.py
index 848090d..db375bc 100644
--- a/tests/test_helpers.py
+++ b/tests/test_helpers.py
@@ -124,6 +124,66 @@ def test_read_xlsx_empty_sheet(self, tmp_path):
         result = nh.read_xlsx(path)
         assert result["Empty"] == []
 
+    def test_read_xlsx_faire_3row_header(self, tmp_path):
+        wb = openpyxl.Workbook()
+        ws = wb.active
+        ws.title = "sampleMetadata"
+        ws.append(["# requirement_level_code", "M", "M"])
+        ws.append(["# section", "Sample", "Location"])
+        ws.append(["samp_name", "SiteName", "decimalLatitude"])
+        ws.append(["WLO17", "West Okoboji Lake", "43.37"])
+        path = str(tmp_path / "faire.xlsx")
+        wb.save(path)
+
+        result = nh.read_xlsx(path, num_headers=2)
+        assert len(result["sampleMetadata"]) == 1
+        assert result["sampleMetadata"][0]["samp_name"] == "WLO17"
+        assert result["sampleMetadata"][0]["SiteName"] == "West Okoboji Lake"
+
+    def test_read_xlsx_faire_3row_header_no_label(self, tmp_path):
+        wb = openpyxl.Workbook()
+        ws = wb.active
+        ws.title = "taxaRaw"
+        ws.append(["M", "HR", "O"])
+        ws.append(["OTU", "ASV", "Taxonomy"])
+        ws.append(["dna_sequence", "scientificName", "taxonID"])
+        ws.append(["ACGT", "Choricystis limnetica", "123"])
+        path = str(tmp_path / "taxaraw.xlsx")
+        wb.save(path)
+
+        result = nh.read_xlsx(path, num_headers=2)
+        assert result["taxaRaw"][0]["scientificName"] == "Choricystis limnetica"
+
+    def test_read_xlsx_stacked_taxon_header_skips_sequence_row(self, tmp_path):
+        wb = openpyxl.Workbook()
+        ws = wb.active
+        ws.title = "data"
+        ws.append(["sitename", "SampleName", "Taxon_A"])
+        ws.append([None, None, "ASV1"])
+        ws.append([None, None, "ACGTACGT"])
+        ws.append(["Lake X", "S1", 3860])
+        ws.append(["Lake X", "S2", 122])
+        path = str(tmp_path / "stacked.xlsx")
+        wb.save(path)
+
+        result = nh.read_xlsx(path, num_headers=2)
+        assert len(result["data"]) == 2
+        assert result["data"][0]["Taxon_A_ASV1"] == 3860
+        assert result["data"][0]["sitename"] == "Lake X"
+
+    def test_read_xlsx_flat_header_unaffected(self, tmp_path):
+        wb = openpyxl.Workbook()
+        ws = wb.active
+        ws.title = "ageModels"
+        ws.append(["PlotDepth(cm)", "AgeModel", "AgeType"])
+        ws.append(["0.5", "CRS", "Calendar"])
+        path = str(tmp_path / "flat.xlsx")
+        wb.save(path)
+
+        result = nh.read_xlsx(path, num_headers=2)
+        assert len(result["ageModels"]) == 1
+        assert result["ageModels"][0]["PlotDepth(cm)"] == "0.5"
+
 
 class TestPullParamsWithSheet:
     def test_pull_params_sheet_selection(self, tmp_path):
@@ -194,6 +254,71 @@ def test_pull_params_csv_list_ignores_sheet_field(self):
         assert result.get("sitename") == "PlainCSVLake"
 
 
+class TestPullParamsValueConstants:
+    """`value:` constants stand in for columns that are not in the data.
+
+    Sheets whose header is not machine-readable (projectMetadata in the FAIRe
+    workbook, for one) cannot be pulled by column, so their entries carry the
+    datum inline as `value:`. That fallback applies only when the column is
+    absent outright — a column that exists but is blank must stay blank rather
+    than quietly picking up the constant.
+    """
+
+    def _yml(self, **extra):
+        entry = {
+            "column": "PrincipalInvestigator",
+            "neotoma": "ndb.datasetpis.contactname",
+            "rowwise": False,
+            "type": "string",
+        }
+        entry.update(extra)
+        return {"metadata": [entry]}
+
+    def test_constant_used_when_column_is_absent(self):
+        csv_file = [{"SomeOtherColumn": "x"}]
+        result = nh.pull_params(
+            ["contactname"], self._yml(value="Spanbauer, Trisha"), csv_file, "ndb.datasetpis"
+        )
+        assert result.get("contactname") == ["Spanbauer, Trisha"]
+
+    def test_constant_used_when_csv_is_a_sheet_dict(self):
+        """No `sheet:` key means template_rows is the whole sheet dict, not rows."""
+        csv_file = {"data": [{"SomeOtherColumn": "x"}]}
+        result = nh.pull_params(
+            ["contactname"], self._yml(value="Spanbauer, Trisha"), csv_file, "ndb.datasetpis"
+        )
+        assert result.get("contactname") == ["Spanbauer, Trisha"]
+
+    def test_piped_constant_is_split_into_several_contacts(self):
+        csv_file = [{"SomeOtherColumn": "x"}]
+        result = nh.pull_params(
+            ["contactname"],
+            self._yml(value="Spanbauer, Trisha | Goring, Simon"),
+            csv_file,
+            "ndb.datasetpis",
+        )
+        assert result.get("contactname") == ["Spanbauer, Trisha", "Goring, Simon"]
+
+    def test_absent_column_without_a_constant_stays_none(self):
+        csv_file = [{"SomeOtherColumn": "x"}]
+        result = nh.pull_params(["contactname"], self._yml(), csv_file, "ndb.datasetpis")
+        assert result.get("contactname") is None
+
+    def test_present_column_wins_over_the_constant(self):
+        csv_file = [{"PrincipalInvestigator": "Goring, Simon"}]
+        result = nh.pull_params(
+            ["contactname"], self._yml(value="Spanbauer, Trisha"), csv_file, "ndb.datasetpis"
+        )
+        assert result.get("contactname") == ["Goring, Simon"]
+
+    def test_blank_column_does_not_fall_back_to_the_constant(self):
+        csv_file = [{"PrincipalInvestigator": ""}]
+        result = nh.pull_params(
+            ["contactname"], self._yml(value="Spanbauer, Trisha"), csv_file, "ndb.datasetpis"
+        )
+        assert result.get("contactname") is None
+
+
 class TestToyData:
     """Quick sanity checks that the toy data files are readable."""
 
diff --git a/tests/test_insert_final.py b/tests/test_insert_final.py
new file mode 100644
index 0000000..8e3b779
--- /dev/null
+++ b/tests/test_insert_final.py
@@ -0,0 +1,63 @@
+"""Tests for insert_final, the last step of an upload.
+
+It writes one row to ``ndb.datasetsubmissions`` with a fixed submissiontypeid
+of 6 and today's date, pulling the three ids straight off the databus.
+"""
+
+from datetime import datetime
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+
+
+def _response(id_int):
+    response = Response()
+    response.id_int = id_int
+    return response
+
+
+@pytest.fixture
+def databus():
+    return {
+        "datasets": _response(74655),
+        "database": _response(17),
+        "contacts": _response(42),
+    }
+
+
+class TestInsertFinal:
+    def test_inserts_submission(self, mock_cur, databus):
+        response = nv.insert_final(mock_cur, databus)
+
+        assert response.validAll
+        assert response.message == ["✔ Dataset submission has been finalized"]
+        assert "ndb.datasetsubmissions" in mock_cur.last_query
+
+    def test_submission_row_contents(self, mock_cur, databus):
+        nv.insert_final(mock_cur, databus)
+
+        assert mock_cur.last_params == {
+            "datasetid": 74655,
+            "databaseid": 17,
+            "contactid": 42,
+            "submissiontypeid": 6,
+            "submissiondate": datetime.now().date(),
+        }
+
+    def test_execute_failure_is_reported(self, mock_cur, databus):
+        def boom(query, params=None):
+            raise RuntimeError("connection lost")
+
+        mock_cur.execute = boom
+        response = nv.insert_final(mock_cur, databus)
+
+        assert not response.validAll
+        assert any("cannot be finalized" in m for m in response.message)
+        assert any("connection lost" in m for m in response.message)
+
+    def test_missing_databus_key_raises(self, mock_cur):
+        """The ids are read before the try block, so a missing key propagates."""
+        with pytest.raises(KeyError):
+            nv.insert_final(mock_cur, {"datasets": _response(1)})
diff --git a/tests/test_valid_assays.py b/tests/test_valid_assays.py
new file mode 100644
index 0000000..5646eae
--- /dev/null
+++ b/tests/test_valid_assays.py
@@ -0,0 +1,104 @@
+"""Tests for the valid_assays / valid_aednalibraries validators.
+
+Both read their fields from ``value:`` constants in the template rather than
+from a sheet, so these tests build small yml_dicts by hand.  A file that
+describes no assay must still validate — that is what lets non-aeDNA uploads
+run through the same pipeline.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+
+
+def _constant(neotoma, value):
+    return {"column": neotoma.split(".")[-1], "neotoma": neotoma, "value": value,
+            "rowwise": False, "required": False, "type": "string"}
+
+
+def _assay_yml(**overrides):
+    """Template entries describing the Okoboji assay."""
+    fields = {
+        "ndb.aednaassays.assayname": "18SrRNAV7",
+        "ndb.aednaassays.assaytypeid": "metabarcoding",
+        "ndb.aednaassays.targetgene": "18S rRNA (SSU eukaryote)",
+        "ndb.aednaassays.pcrprimerforward": "5'-GGCTTAATTTGACTCAACRCG-3'",
+    }
+    fields.update(overrides)
+    return {"metadata": [_constant(k, v) for k, v in fields.items()]}
+
+
+def _library_yml():
+    return {"metadata": [
+        _constant("ndb.aednalibraries.platform", "ILLUMINA"),
+        _constant("ndb.aednalibraries.instrument", "Miseq"),
+    ]}
+
+
+@pytest.fixture
+def databus():
+    """Prior validation results supplying the parent ids."""
+    datasets, assays = Response(), Response()
+    datasets.id_int = 74655
+    assays.id_int = 1
+    return {"datasets": datasets, "assays": assays}
+
+
+class TestValidAssays:
+    def test_inserts_assay(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (7,)
+        response = nv.valid_assays(mock_cur, _assay_yml(), csv_file=None, databus=databus)
+        assert response.validAll
+        assert response.id_int == 7
+        assert "ndb.aednaassays" in mock_cur.last_query
+
+    def test_assay_type_resolved_to_id(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (3,)
+        nv.valid_assays(mock_cur, _assay_yml(), csv_file=None, databus=databus)
+        assert mock_cur.last_params["assaytypeid"] == 3
+
+    def test_dataset_id_taken_from_databus(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (7,)
+        nv.valid_assays(mock_cur, _assay_yml(), csv_file=None, databus=databus)
+        assert mock_cur.last_params["datasetid"] == 74655
+
+    def test_unknown_assay_type_is_invalid(self, mock_cur, databus):
+        mock_cur.mock_fetchone = None
+        response = nv.valid_assays(mock_cur, _assay_yml(), csv_file=None, databus=databus)
+        assert not response.validAll
+        assert any("not known to Neotoma" in m for m in response.message)
+
+    def test_no_assay_described_still_valid(self, mock_cur, databus):
+        """No assay in the template means nothing to insert, not a failure."""
+        response = nv.valid_assays(mock_cur, {"metadata": []}, csv_file=None, databus=databus)
+        assert response.validAll
+        assert mock_cur.last_query is None
+
+
+class TestValidAednaLibraries:
+    def test_inserts_library_linked_to_assay(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (2,)
+        response = nv.valid_aednalibraries(
+            mock_cur, _library_yml(), csv_file=None, databus=databus
+        )
+        assert response.validAll
+        assert response.id_int == 2
+        assert mock_cur.last_params["assayid"] == 1
+        assert mock_cur.last_params["datasetid"] == 74655
+        assert mock_cur.last_params["platform"] == "ILLUMINA"
+
+    def test_no_library_described_still_valid(self, mock_cur, databus):
+        response = nv.valid_aednalibraries(
+            mock_cur, {"metadata": []}, csv_file=None, databus=databus
+        )
+        assert response.validAll
+        assert mock_cur.last_query is None
+
+    def test_missing_assay_id_is_invalid(self, mock_cur):
+        datasets = Response()
+        datasets.id_int = 74655
+        response = nv.valid_aednalibraries(
+            mock_cur, _library_yml(), csv_file=None, databus={"datasets": datasets}
+        )
+        assert not response.validAll
diff --git a/tests/test_valid_dataset_database.py b/tests/test_valid_dataset_database.py
new file mode 100644
index 0000000..d9d9431
--- /dev/null
+++ b/tests/test_valid_dataset_database.py
@@ -0,0 +1,80 @@
+"""Tests for the valid_dataset_database validator.
+
+``valid_dataset_database`` is the only validator with no ``csv_file`` argument —
+it reads the database name from a ``value:`` constant in the template and
+resolves it against ``ndb.constituentdatabases``.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+
+
+def _db_yml(name="Neotoma"):
+    return {
+        "metadata": [
+            {
+                "neotoma": "ndb.datasetdatabases.databasename",
+                "column": "databasename",
+                "value": name,
+                "rowwise": False,
+                "required": False,
+                "type": "string",
+            }
+        ]
+    }
+
+
+@pytest.fixture
+def databus():
+    datasets = Response()
+    datasets.id_int = 74655
+    return {"datasets": datasets}
+
+
+class TestValidDatasetDatabase:
+    def test_resolves_and_inserts(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (17,)
+        response = nv.valid_dataset_database(mock_cur, _db_yml(), databus=databus)
+
+        assert response.validAll
+        assert response.id_int == 17
+        assert "ts.insertdatasetdatabase" in mock_cur.last_query
+        assert mock_cur.last_params == {"datasetid": 74655, "databaseid": 17}
+
+    def test_lookup_is_lowercased_and_stripped(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (17,)
+        nv.valid_dataset_database(mock_cur, _db_yml("  NEOTOMA  "), databus=databus)
+
+        first_query, first_params = mock_cur._execute_calls[0]
+        assert "ndb.constituentdatabases" in first_query
+        assert first_params == {"databasename": "neotoma"}
+
+    def test_unknown_database_returns_early(self, mock_cur, databus):
+        mock_cur.mock_fetchone = None
+        response = nv.valid_dataset_database(mock_cur, _db_yml("Nowhere"), databus=databus)
+
+        assert not response.validAll
+        assert any("not found in Neotoma" in m for m in response.message)
+        # Returned before any DatasetDatabase was built.
+        assert response.id_int is None
+        assert len(mock_cur._execute_calls) == 1
+
+    def test_missing_databus_uses_placeholder_dataset(self, mock_cur):
+        mock_cur.mock_fetchone = (17,)
+        response = nv.valid_dataset_database(mock_cur, _db_yml(), databus=None)
+
+        assert not response.validAll
+        assert any("Cannot retrieve Dataset ID" in m for m in response.message)
+        # Still builds the object against the placeholder datasetid of 1.
+        assert response.id_int == 17
+        assert mock_cur.last_params == {"datasetid": 1, "databaseid": 17}
+
+    def test_non_string_name_skips_lookup(self, mock_cur, databus):
+        """A numeric databasename means no lookup runs, so databaseid is absent."""
+        response = nv.valid_dataset_database(mock_cur, _db_yml(5), databus=databus)
+
+        assert isinstance(response, Response)
+        assert not response.validAll
+        assert any("Cannot create Database object" in m for m in response.message)
diff --git a/tests/test_valid_geopolitical_units.py b/tests/test_valid_geopolitical_units.py
new file mode 100644
index 0000000..ee8c6e6
--- /dev/null
+++ b/tests/test_valid_geopolitical_units.py
@@ -0,0 +1,132 @@
+"""Tests for the valid_geopolitical_units validator.
+
+Geopolitical units are resolved top-down: the national unit is looked up by
+name alone, then each subnational level is looked up by name *and* by the id of
+the level above it, so 'Washington' the county resolves under the right state.
+Each level found is linked to the site via ``ts.insertsitegeopol``.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+
+
+def _entry(field):
+    return {
+        "neotoma": f"ndb.sitegeopolitical.{field}",
+        "column": field,
+        "rowwise": False,
+        "required": False,
+        "type": "string",
+    }
+
+
+def _geopol(**levels):
+    """Build the (yml_dict, csv_file) pair for a set of geopolitical levels.
+
+    ``pull_params`` reads values out of the CSV rows by column name — a
+    ``value:`` constant in the template is only honoured by the
+    ``retrieve_dict``-based validators — so each level needs both a metadata
+    entry naming the column and a CSV row carrying it.
+    """
+    yml = {"metadata": [_entry(k) for k in levels]}
+    csv_file = [dict(levels)]
+    return yml, csv_file
+
+
+@pytest.fixture
+def databus():
+    sites = Response()
+    sites.id_int = 5001
+    return {"sites": sites}
+
+
+def _lookups(cur):
+    """Params of the geopoliticalunits SELECTs, in order (excludes the inserts)."""
+    return [p for q, p in cur._execute_calls if "ndb.geopoliticalunits" in q]
+
+
+def _inserts(cur):
+    return [p for q, p in cur._execute_calls if "insertsitegeopol" in q]
+
+
+class TestValidGeopoliticalUnits:
+    def test_no_units_given_is_valid(self, mock_cur, databus):
+        response = nv.valid_geopolitical_units(
+            mock_cur, {"metadata": []}, csv_file=[], databus=databus
+        )
+
+        assert response.validAll
+        assert response.message == ["? No Geopolitical Units given."]
+        assert mock_cur._execute_calls == []
+
+    def test_national_unit_not_found_returns_early(self, mock_cur, databus):
+        mock_cur.mock_fetchone = None
+        yml, csv_file = _geopol(national_unit="Atlantis")
+        response = nv.valid_geopolitical_units(mock_cur, yml, csv_file, databus=databus)
+
+        assert not response.validAll
+        assert any("National Unit Atlantis not found" in m for m in response.message)
+        assert response.id_list == []
+        assert _inserts(mock_cur) == []
+
+    def test_national_unit_found_and_linked(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (300,)
+        yml, csv_file = _geopol(national_unit="United States")
+        response = nv.valid_geopolitical_units(mock_cur, yml, csv_file, databus=databus)
+
+        assert response.validAll
+        assert response.id_list == [300]
+        assert _lookups(mock_cur)[0] == {"geopoliticalname": "united states"}
+        assert _inserts(mock_cur) == [{"siteid": 5001, "geopolid": 300}]
+
+    def test_subnational_lookup_uses_parent_id(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (300,)
+        yml, csv_file = _geopol(national_unit="United States", subnational_unit_lv1="Iowa")
+        nv.valid_geopolitical_units(mock_cur, yml, csv_file, databus=databus)
+
+        state_lookup = _lookups(mock_cur)[1]
+        assert state_lookup == {"geopoliticalname": "iowa", "highergeopoliticalid": 300}
+
+    def test_all_levels_collected_in_id_list(self, mock_cur, databus):
+        # Each successive fetchone stands in for the next level down.
+        ids = iter([(300,), (301,), (302,)])
+        mock_cur.fetchone = lambda: next(ids, None)
+        yml, csv_file = _geopol(
+            national_unit="United States",
+            subnational_unit_lv1="Iowa",
+            subnational_unit_lv2="Dickinson",
+        )
+        response = nv.valid_geopolitical_units(mock_cur, yml, csv_file, databus=databus)
+
+        assert response.validAll
+        assert response.id_list == [300, 301, 302]
+        assert _inserts(mock_cur) == [
+            {"siteid": 5001, "geopolid": 300},
+            {"siteid": 5001, "geopolid": 301},
+            {"siteid": 5001, "geopolid": 302},
+        ]
+
+    def test_unknown_subregion_stops_the_walk(self, mock_cur, databus):
+        ids = iter([(300,), None])
+        mock_cur.fetchone = lambda: next(ids, None)
+        yml, csv_file = _geopol(
+            national_unit="United States",
+            subnational_unit_lv1="Nowhere",
+            subnational_unit_lv2="Dickinson",
+        )
+        response = nv.valid_geopolitical_units(mock_cur, yml, csv_file, databus=databus)
+
+        # The national unit still validated, and only it was linked.
+        assert response.id_list == [300]
+        assert any("Subregional Unit Nowhere not found" in m for m in response.message)
+        assert _inserts(mock_cur) == [{"siteid": 5001, "geopolid": 300}]
+
+    def test_missing_site_id_reports_link_failure(self, mock_cur):
+        mock_cur.mock_fetchone = (300,)
+        yml, csv_file = _geopol(national_unit="United States")
+        response = nv.valid_geopolitical_units(mock_cur, yml, csv_file, databus={})
+
+        assert not response.validAll
+        assert any("Could not link national unit to site" in m for m in response.message)
diff --git a/tests/test_valid_hiatus.py b/tests/test_valid_hiatus.py
new file mode 100644
index 0000000..82b37b7
--- /dev/null
+++ b/tests/test_valid_hiatus.py
@@ -0,0 +1,154 @@
+"""Tests for the valid_hiatus validator and its _find_clusters helper.
+
+A hiatus is expressed in the CSV as a flag on the rows that bound the gap.
+``valid_hiatus`` collects the flagged row indices, groups the consecutive ones
+into clusters, then maps each cluster's first/last index onto the analysis unit
+ids produced earlier in the pipeline.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+from DataBUS.neotomaValidator.valid_hiatus import _find_clusters
+
+
+def _hiatus_yml():
+    return {
+        "metadata": [
+            {
+                "neotoma": "ndb.hiatuses.hiatus",
+                "column": "Hiatus",
+                "rowwise": True,
+                "required": False,
+            }
+        ]
+    }
+
+
+def _csv(flags):
+    """One row per flag; 'NA' becomes None once pull_params has cleaned it."""
+    return [{"Hiatus": f} for f in flags]
+
+
+def _spans(cur):
+    """The (start, end) AU pair of every insert_hiatus call the cursor saw.
+
+    ``Hiatus.insert_to_db`` issues two statements per hiatus — the function DDL
+    with no params, then the parameterised SELECT — so filter to the latter.
+    """
+    return [
+        (p["analysisunitstart"], p["analysisunitend"])
+        for _, p in cur._execute_calls
+        if p and "analysisunitstart" in p
+    ]
+
+
+@pytest.fixture
+def databus():
+    """Ten analysis units, ids 100..109, one per CSV row."""
+    aus = Response()
+    aus.id_list = list(range(100, 110))
+    return {"analysisunits": aus}
+
+
+class TestFindClusters:
+    def test_empty(self):
+        assert _find_clusters([]) == []
+
+    def test_single_run(self):
+        assert _find_clusters([2, 3, 4]) == [[2, 3, 4]]
+
+    def test_splits_on_gap(self):
+        assert _find_clusters([0, 1, 5, 6, 9]) == [[0, 1], [5, 6], [9]]
+
+    def test_sorts_before_grouping(self):
+        assert _find_clusters([6, 1, 5, 0]) == [[0, 1], [5, 6]]
+
+    def test_isolated_indices(self):
+        assert _find_clusters([0, 2, 4]) == [[0], [2], [4]]
+
+
+class TestValidHiatus:
+    def test_no_hiatus_column_returns_early(self, mock_cur, databus):
+        response = nv.valid_hiatus(mock_cur, {"metadata": []}, csv_file=[], databus=databus)
+
+        assert response.validAll
+        assert response.message == ["✔ No hiatuses found in the data."]
+        assert mock_cur._execute_calls == []
+
+    def test_all_rows_blank_produces_no_hiatus(self, mock_cur, databus):
+        """A present-but-empty hiatus column is not the same as an absent one.
+
+        The early return only fires when the column is missing entirely, so an
+        all-blank column falls through to an empty cluster list: no inserts and
+        no per-hiatus messages.
+        """
+        response = nv.valid_hiatus(
+            mock_cur, _hiatus_yml(), csv_file=_csv(["NA", "NA", "NA"]), databus=databus
+        )
+
+        assert response.validAll
+        assert response.message == []
+        assert mock_cur._execute_calls == []
+
+    def test_consecutive_flags_become_one_hiatus(self, mock_cur, databus):
+        # Rows 2 and 3 flagged -> one hiatus spanning AU 102..103.
+        mock_cur.mock_fetchone = (1,)
+        response = nv.valid_hiatus(
+            mock_cur,
+            _hiatus_yml(),
+            csv_file=_csv(["NA", "NA", "yes", "yes", "NA"]),
+            databus=databus,
+        )
+
+        assert response.validAll
+        assert "✔ Hiatus can be created." in response.message
+        assert "✔ Hiatus inserted." in response.message
+        assert mock_cur.last_params["analysisunitstart"] == 102
+        assert mock_cur.last_params["analysisunitend"] == 103
+        assert _spans(mock_cur) == [(102, 103)]
+
+    def test_two_separate_gaps_insert_twice(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (1,)
+        response = nv.valid_hiatus(
+            mock_cur,
+            _hiatus_yml(),
+            csv_file=_csv(["yes", "NA", "NA", "yes", "yes"]),
+            databus=databus,
+        )
+
+        assert response.validAll
+        assert _spans(mock_cur) == [(100, 100), (103, 104)]
+
+    def test_single_flagged_row_uses_same_au_both_ends(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (1,)
+        nv.valid_hiatus(
+            mock_cur, _hiatus_yml(), csv_file=_csv(["NA", "yes", "NA"]), databus=databus
+        )
+
+        assert mock_cur.last_params["analysisunitstart"] == 101
+        assert mock_cur.last_params["analysisunitend"] == 101
+
+    def test_missing_analysisunits_falls_back_to_indices(self, mock_cur):
+        response = nv.valid_hiatus(
+            mock_cur, _hiatus_yml(), csv_file=_csv(["NA", "yes", "yes"]), databus={}
+        )
+
+        assert not response.validAll
+        assert any("Could not resolve analysis unit IDs" in m for m in response.message)
+        # Fell back to the raw row indices 1 and 2.
+        assert mock_cur.last_params["analysisunitstart"] == 1
+        assert mock_cur.last_params["analysisunitend"] == 2
+
+    def test_insert_failure_is_reported(self, mock_cur, databus):
+        def boom(query, params=None):
+            raise RuntimeError("no such table")
+
+        mock_cur.execute = boom
+        response = nv.valid_hiatus(
+            mock_cur, _hiatus_yml(), csv_file=_csv(["yes", "yes"]), databus=databus
+        )
+
+        assert not response.validAll
+        assert any("Could not insert hiatus" in m for m in response.message)
diff --git a/tests/test_valid_pbmodel.py b/tests/test_valid_pbmodel.py
new file mode 100644
index 0000000..bbe7635
--- /dev/null
+++ b/tests/test_valid_pbmodel.py
@@ -0,0 +1,123 @@
+"""Tests for the valid_pbmodel validator.
+
+A 210Pb model applies one basis + cumulative inventory across every analysis
+unit in the core, so the validator emits one ``insert_lead_model`` call per
+analysis unit id on the databus.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+
+
+def _entry(field, column, type_spec="string"):
+    return {
+        "neotoma": f"ndb.leadmodels.{field}",
+        "column": column,
+        "rowwise": False,
+        "required": False,
+        "type": type_spec,
+    }
+
+
+def _pb(basis="constant rate of supply", inventory="145.3"):
+    """(yml_dict, csv_file) describing one lead model.
+
+    ``pull_params`` reads from the CSV by column name, so every field needs a
+    metadata entry naming its column plus a row carrying the value.
+    """
+    yml = {
+        "metadata": [
+            _entry("pbbasisid", "PbBasis"),
+            _entry("cumulativeinventory", "CumInventory", "float"),
+        ]
+    }
+    return yml, [{"PbBasis": basis, "CumInventory": inventory}]
+
+
+@pytest.fixture
+def databus():
+    aus = Response()
+    aus.id_list = [201, 202, 203]
+    return {"analysisunits": aus}
+
+
+def _model_inserts(cur):
+    """Params of every insert_lead_model SELECT.
+
+    ``LeadModel.insert_to_db`` first replays the CREATE FUNCTION DDL — whose
+    body also mentions insert_lead_model — with no params, so require params
+    to pick out the real inserts.
+    """
+    return [p for q, p in cur._execute_calls if p and "insert_lead_model" in q]
+
+
+class TestValidPbModel:
+    def test_no_parameters_is_valid(self, mock_cur, databus):
+        response = nv.valid_pbmodel(mock_cur, {"metadata": []}, csv_file=[], databus=databus)
+
+        assert response.validAll
+        assert any("No lead model parameters provided" in m for m in response.message)
+        assert mock_cur._execute_calls == []
+
+    def test_one_model_per_analysis_unit(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (4,)
+        yml, csv_file = _pb()
+        response = nv.valid_pbmodel(mock_cur, yml, csv_file, databus)
+
+        assert response.validAll
+        assert "✔  Lead model can be inserted." in response.message
+        inserts = _model_inserts(mock_cur)
+        assert [p["analysisunitid"] for p in inserts] == [201, 202, 203]
+
+    def test_basis_name_resolved_to_id(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (4,)
+        yml, csv_file = _pb(basis="constant rate of supply")
+        nv.valid_pbmodel(mock_cur, yml, csv_file, databus)
+
+        basis_query, basis_params = mock_cur._execute_calls[0]
+        assert "ndb.leadmodelbasis" in basis_query
+        assert basis_params == {"pbbasisid": "constant rate of supply"}
+        assert all(p["pbbasisid"] == 4 for p in _model_inserts(mock_cur))
+
+    def test_cumulative_inventory_is_carried_through(self, mock_cur, databus):
+        mock_cur.mock_fetchone = (4,)
+        yml, csv_file = _pb(inventory="145.3")
+        nv.valid_pbmodel(mock_cur, yml, csv_file, databus)
+
+        assert all(p["cumulativeinventory"] == 145.3 for p in _model_inserts(mock_cur))
+
+    def test_unknown_basis_leaves_id_none(self, mock_cur, databus):
+        mock_cur.mock_fetchone = None
+        yml, csv_file = _pb(basis="not a real basis")
+        nv.valid_pbmodel(mock_cur, yml, csv_file, databus)
+
+        assert all(p["pbbasisid"] is None for p in _model_inserts(mock_cur))
+
+    def test_missing_analysis_units_uses_placeholder_range(self, mock_cur):
+        mock_cur.mock_fetchone = (4,)
+        yml, csv_file = _pb()
+        response = nv.valid_pbmodel(mock_cur, yml, csv_file, databus={})
+
+        assert not response.validAll
+        assert any("using placeholder range" in m for m in response.message)
+        # The placeholder is range(1, 10) -> nine models.
+        assert len(_model_inserts(mock_cur)) == 9
+
+    def test_insert_failure_is_reported_once(self, mock_cur, databus):
+        calls = []
+
+        def execute(query, params=None):
+            calls.append((query, params))
+            if "insert_lead_model" in query:
+                raise RuntimeError("no such function")
+
+        mock_cur.execute = execute
+        mock_cur.mock_fetchone = (4,)
+        yml, csv_file = _pb()
+        response = nv.valid_pbmodel(mock_cur, yml, csv_file, databus)
+
+        assert not response.validAll
+        failures = [m for m in response.message if "Lead model cannot be inserted" in m]
+        assert len(failures) == 1
diff --git a/tests/test_valid_project.py b/tests/test_valid_project.py
new file mode 100644
index 0000000..b4c54ca
--- /dev/null
+++ b/tests/test_valid_project.py
@@ -0,0 +1,445 @@
+"""Tests for the valid_project validator and its extraction helpers.
+
+``valid_project`` walks the whole project hierarchy in one pass: it inserts (or
+finds) the project, links it to the dataset, then works through grants,
+funding institutions, awardees, participants, keywords and contact-institution
+pairings. Every one of those steps is a different SQL statement, so these tests
+use a routing cursor that answers each query by pattern.
+
+Unlike most validators, valid_project reads ``value:`` constants straight off
+the template via ``entry.get("value", entry.get("column"))`` — it never touches
+the CSV — so the yml_dicts here carry the data directly.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+from DataBUS.neotomaValidator.valid_project import (
+    _extract_contact_institutions,
+    _extract_grant_info,
+    _extract_keywords,
+    _extract_participants,
+    _extract_project_info,
+    _resolve_contact,
+    _resolve_keyword,
+)
+
+
+class RoutingCursor:
+    """A cursor whose fetchone() answer depends on the query it just saw.
+
+    ``routes`` maps a distinctive substring of a query to the tuple that
+    fetchone() should return for it. Anything unmatched returns None, which is
+    how "not found" reads to the validator.
+    """
+
+    def __init__(self, routes=None):
+        self.routes = routes or {}
+        self.calls = []
+        self._next = None
+
+    def execute(self, query, params=None):
+        self.calls.append((query, params))
+        self._next = None
+        for marker, value in self.routes.items():
+            if marker in query:
+                self._next = value
+                break
+
+    def fetchone(self):
+        return self._next
+
+    def fetchall(self):
+        return []
+
+    def queries_matching(self, marker):
+        return [p for q, p in self.calls if marker in q]
+
+
+def _entry(neotoma, value, **extra):
+    entry = {"neotoma": neotoma, "value": value}
+    entry.update(extra)
+    return entry
+
+
+def _project_yml(name="Okoboji aeDNA", **extra_entries):
+    metadata = [
+        _entry("ndb.projects.projectname", name),
+        _entry("ndb.projects.projectdescription", "Lake sediment eDNA survey"),
+    ]
+    metadata.extend(extra_entries.get("metadata", []))
+    return {"metadata": metadata}
+
+
+@pytest.fixture
+def databus():
+    datasets = Response()
+    datasets.id_int = 74655
+    contacts = Response()
+    contacts.id_int = 42
+    return {"datasets": datasets, "contacts": contacts}
+
+
+# Default happy-path routing: the project is new, everything else resolves.
+BASE_ROUTES = {
+    "SELECT projectid FROM ndb.projects": None,
+    "INSERT INTO ndb.projects": (900,),
+}
+
+
+class TestExtractProjectInfo:
+    def test_returns_none_without_project_entries(self):
+        assert _extract_project_info({"metadata": []}) is None
+        assert _extract_project_info({"metadata": [_entry("ndb.sites.sitename", "X")]}) is None
+
+    def test_collects_fields_by_last_path_segment(self):
+        info = _extract_project_info(_project_yml())
+        assert info == {
+            "projectname": "Okoboji aeDNA",
+            "projectdescription": "Lake sediment eDNA survey",
+        }
+
+    def test_falls_back_to_column_when_no_value(self):
+        yml = {"metadata": [{"neotoma": "ndb.projects.projectname", "column": "ProjName"}]}
+        assert _extract_project_info(yml) == {"projectname": "ProjName"}
+
+
+class TestExtractGrantInfo:
+    def test_no_grants(self):
+        assert _extract_grant_info({"metadata": []}) == []
+
+    def test_single_grant(self):
+        yml = {
+            "metadata": [
+                _entry("ndb.grants.grantname", "NSF BIO"),
+                _entry("ndb.grants.grantnumber", "2055632"),
+            ]
+        }
+        assert _extract_grant_info(yml) == [{"grantname": "NSF BIO", "grantnumber": "2055632"}]
+
+    def test_nested_institution_and_awardee(self):
+        yml = {
+            "metadata": [
+                _entry(
+                    "ndb.grants.grantname",
+                    "NSF BIO",
+                    institutionid="https://ror.org/01y2jtd41",
+                    institutionname="UW-Madison",
+                    institutionlocation="Madison, WI",
+                    awardee="Goring, Simon",
+                ),
+            ]
+        }
+        (grant,) = _extract_grant_info(yml)
+        assert grant["institution"]["institutionid"] == "https://ror.org/01y2jtd41"
+        assert grant["awardee"] == "Goring, Simon"
+
+    def test_non_grant_entry_closes_the_current_group(self):
+        yml = {
+            "metadata": [
+                _entry("ndb.grants.grantnumber", "111"),
+                _entry("ndb.projects.projectname", "P"),
+                _entry("ndb.grants.grantnumber", "222"),
+            ]
+        }
+        assert _extract_grant_info(yml) == [{"grantnumber": "111"}, {"grantnumber": "222"}]
+
+
+class TestExtractListFields:
+    def test_participants_split_on_pipe(self):
+        yml = {
+            "metadata": [
+                _entry("ndb.projectparticipants.contactname", "Goring, Simon | Williams, John")
+            ]
+        }
+        assert _extract_participants(yml) == ["Goring, Simon", "Williams, John"]
+
+    def test_participants_accept_a_list(self):
+        yml = {"metadata": [_entry("ndb.projectparticipants.contactname", ["A", "B"])]}
+        assert _extract_participants(yml) == ["A", "B"]
+
+    def test_participants_absent(self):
+        assert _extract_participants({"metadata": []}) == []
+
+    def test_keywords_split_on_pipe_and_strip(self):
+        yml = {"metadata": [_entry("ndb.projectkeywords.keyword", " eDNA | lake  |  ")]}
+        assert _extract_keywords(yml) == ["eDNA", "lake"]
+
+    def test_contact_institutions(self):
+        yml = {
+            "metadata": [
+                _entry(
+                    "ndb.contactinstitutions.contactname",
+                    "Goring, Simon",
+                    institutionid="https://ror.org/01y2jtd41",
+                    institutionname="UW-Madison",
+                )
+            ]
+        }
+        (ci,) = _extract_contact_institutions(yml)
+        assert ci["contactname"] == "Goring, Simon"
+        assert ci["institutionname"] == "UW-Madison"
+
+
+class TestResolveKeyword:
+    def test_none_keyword(self):
+        assert _resolve_keyword(RoutingCursor(), None) is None
+
+    def test_found(self):
+        cur = RoutingCursor({"ndb.keywords": (12,)})
+        assert _resolve_keyword(cur, "  eDNA ") == 12
+        assert cur.queries_matching("ndb.keywords") == [{"kw": "edna"}]
+
+    def test_not_found(self):
+        assert _resolve_keyword(RoutingCursor(), "nope") is None
+
+
+class ContactCursor:
+    """Answers get_contacts' exact-name lookup for a fixed set of known contacts.
+
+    get_contacts issues the same SQL text no matter which name it is given, so
+    routing here keys off the bound parameter instead of the query. Only the
+    exact-contactname lookup is answered; the surname/givennames fallback query
+    carries different params and so returns None, which reads as "not found".
+    """
+
+    def __init__(self, known=None):
+        self.known = known or {}
+        self.seen = []
+        self._next = None
+
+    def execute(self, query, params=None):
+        self.seen.append(params)
+        self._next = None
+        if params and "contactname" in params:
+            self._next = self.known.get(params["contactname"])
+
+    def fetchone(self):
+        return self._next
+
+    @property
+    def names_looked_up(self):
+        return [p["contactname"] for p in self.seen if p and "contactname" in p]
+
+
+# Neotoma stores contacts as "Surname, Given"; FAIRe templates carry recordedBy
+# the other way round, so _resolve_contact has to bridge the two.
+SPANBAUER = {"spanbauer, trisha": (10234, "Spanbauer, Trisha")}
+
+
+class TestResolveContact:
+    def test_empty_name_is_none(self):
+        assert _resolve_contact(ContactCursor(), None) is None
+        assert _resolve_contact(ContactCursor(), "") is None
+
+    def test_canonical_surname_first_matches_directly(self):
+        cur = ContactCursor(SPANBAUER)
+        assert _resolve_contact(cur, "Spanbauer, Trisha") == 10234
+        assert cur.names_looked_up == ["spanbauer, trisha"]
+
+    def test_given_name_first_is_reordered_on_retry(self):
+        cur = ContactCursor(SPANBAUER)
+        assert _resolve_contact(cur, "Trisha L. Spanbauer") == 10234
+        # First attempt uses the template's order and misses; the retry reorders
+        # it into Neotoma's convention, dropping the middle initial.
+        assert cur.names_looked_up == ["trisha l. spanbauer", "spanbauer, trisha"]
+
+    def test_surrounding_whitespace_is_stripped(self):
+        cur = ContactCursor(SPANBAUER)
+        assert _resolve_contact(cur, "  Trisha L. Spanbauer  ") == 10234
+
+    def test_name_with_a_comma_is_not_reordered(self):
+        cur = ContactCursor()
+        assert _resolve_contact(cur, "Nobody, Here") is None
+        # A comma means the caller already used Neotoma's order, so there is
+        # nothing to retry — exactly one lookup, not two.
+        assert cur.names_looked_up == ["nobody, here"]
+
+    def test_unknown_name_is_none(self):
+        cur = ContactCursor()
+        assert _resolve_contact(cur, "Trisha L. Spanbauer") is None
+
+
+class TestValidProject:
+    def test_no_project_entry_is_valid(self, databus):
+        response = nv.valid_project(
+            RoutingCursor(), {"metadata": []}, csv_file=None, databus=databus
+        )
+
+        assert response.validAll
+        assert response.message == ["? No project entry found in template."]
+
+    def test_missing_project_name_is_invalid(self, databus):
+        yml = {"metadata": [_entry("ndb.projects.projectdescription", "no name here")]}
+        response = nv.valid_project(RoutingCursor(), yml, csv_file=None, databus=databus)
+
+        assert not response.validAll
+        assert any("Project name is required" in m for m in response.message)
+
+    def test_new_project_is_inserted_and_linked(self, databus):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        response = nv.valid_project(cur, _project_yml(), csv_file=None, databus=databus)
+
+        assert response.validAll
+        assert response.id_int == 900
+        assert any("inserted (ID: 900)" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.projectdatasets") == [
+            {"projectid": 900, "datasetid": 74655}
+        ]
+
+    def test_existing_project_is_reused(self, databus):
+        cur = RoutingCursor({"SELECT projectid FROM ndb.projects": (777,)})
+        response = nv.valid_project(cur, _project_yml(), csv_file=None, databus=databus)
+
+        assert response.id_int == 777
+        assert any("already exists (ID: 777)" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.projects") == []
+
+    def test_project_lookup_is_lowercased(self, databus):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        nv.valid_project(cur, _project_yml("  Okoboji aeDNA "), csv_file=None, databus=databus)
+
+        assert cur.queries_matching("SELECT projectid FROM ndb.projects") == [
+            {"name": "okoboji aedna"}
+        ]
+
+    def test_missing_dataset_does_not_abort(self, databus):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        response = nv.valid_project(cur, _project_yml(), csv_file=None, databus={})
+
+        # Project still inserted; only the link is reported as a soft failure.
+        assert response.id_int == 900
+        assert any("Could not link project to dataset" in m for m in response.message)
+
+    def test_grant_inserted_and_linked(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(_entry("ndb.grants.grantnumber", "2055632"))
+        routes = dict(BASE_ROUTES)
+        routes["SELECT grantid FROM ndb.grants"] = None
+        routes["INSERT INTO ndb.grants"] = (55,)
+        cur = RoutingCursor(routes)
+
+        response = nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert any("Grant inserted (ID: 55)" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.projectgrants") == [
+            {"projectid": 900, "grantid": 55}
+        ]
+
+    def test_existing_grant_is_reused(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(_entry("ndb.grants.grantnumber", "2055632"))
+        routes = dict(BASE_ROUTES)
+        routes["SELECT grantid FROM ndb.grants"] = (55,)
+        cur = RoutingCursor(routes)
+
+        response = nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert any("already exists (ID: 55)" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.grants") == []
+
+    def test_funding_institution_linked_to_grant(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(
+            _entry(
+                "ndb.grants.grantnumber",
+                "2055632",
+                institutionid="https://ror.org/01y2jtd41",
+                institutionname="UW-Madison",
+            )
+        )
+        routes = dict(BASE_ROUTES)
+        routes["SELECT grantid FROM ndb.grants"] = None
+        routes["INSERT INTO ndb.grants"] = (55,)
+        routes["INSERT INTO ndb.institutions"] = ("https://ror.org/01y2jtd41",)
+        cur = RoutingCursor(routes)
+
+        nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert cur.queries_matching("INSERT INTO ndb.fundinginstitutions") == [
+            {"institutionid": "https://ror.org/01y2jtd41", "grantid": 55}
+        ]
+
+    def test_participant_resolved_and_linked(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(_entry("ndb.projectparticipants.contactname", "Goring, Simon"))
+        routes = dict(BASE_ROUTES)
+        routes["FROM ndb.contacts"] = (42, "Goring, Simon")
+        cur = RoutingCursor(routes)
+
+        response = nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert response.validAll
+        assert cur.queries_matching("INSERT INTO ndb.projectparticipants") == [
+            {"projectid": 900, "contactid": 42}
+        ]
+
+    def test_unknown_participant_is_reported_not_fatal(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(_entry("ndb.projectparticipants.contactname", "Nobody, Real"))
+        cur = RoutingCursor(dict(BASE_ROUTES))
+
+        response = nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert response.validAll
+        assert any("Participant 'Nobody, Real' not found" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.projectparticipants") == []
+
+    def test_keyword_resolved_and_linked(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(_entry("ndb.projectkeywords.keyword", "eDNA | lake"))
+        routes = dict(BASE_ROUTES)
+        routes["ndb.keywords"] = (12,)
+        cur = RoutingCursor(routes)
+
+        nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert cur.queries_matching("INSERT INTO ndb.projectkeywords") == [
+            {"projectid": 900, "keywordid": 12},
+            {"projectid": 900, "keywordid": 12},
+        ]
+
+    def test_unknown_keyword_is_reported(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(_entry("ndb.projectkeywords.keyword", "nonsense"))
+        cur = RoutingCursor(dict(BASE_ROUTES))
+
+        response = nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert any("Keyword 'nonsense' not found" in m for m in response.message)
+
+    def test_contact_institution_linked(self, databus):
+        yml = _project_yml()
+        yml["metadata"].append(
+            _entry(
+                "ndb.contactinstitutions.contactname",
+                "Goring, Simon",
+                institutionid="https://ror.org/01y2jtd41",
+                institutionname="UW-Madison",
+            )
+        )
+        routes = dict(BASE_ROUTES)
+        routes["FROM ndb.contacts"] = (42, "Goring, Simon")
+        routes["INSERT INTO ndb.institutions"] = ("https://ror.org/01y2jtd41",)
+        cur = RoutingCursor(routes)
+
+        nv.valid_project(cur, yml, csv_file=None, databus=databus)
+
+        assert cur.queries_matching("INSERT INTO ndb.contactinstitutions") == [
+            {"institutionid": "https://ror.org/01y2jtd41", "contactid": 42}
+        ]
+
+    def test_project_insert_failure_returns_early(self, databus):
+        class Failing(RoutingCursor):
+            def execute(self, query, params=None):
+                super().execute(query, params)
+                if "INSERT INTO ndb.projects" in query:
+                    raise RuntimeError("permission denied")
+
+        cur = Failing(dict(BASE_ROUTES))
+        response = nv.valid_project(cur, _project_yml(), csv_file=None, databus=databus)
+
+        assert not response.validAll
+        assert any("Cannot insert project" in m for m in response.message)
+        assert response.id_int is None
diff --git a/tests/test_valid_sample.py b/tests/test_valid_sample.py
index c1560d3..a2d7324 100644
--- a/tests/test_valid_sample.py
+++ b/tests/test_valid_sample.py
@@ -84,3 +84,72 @@ def test_id_list_populated_on_insert(self, mock_cur, pb210_pair):
             cur=mock_cur, yml_dict=yml_dict, csv_file=csv_file, databus=databus
         )
         assert isinstance(result.id_list, list)
+
+
+class TestValidSampleRowCountGuard:
+    """Sample names are paired with analysis units by position alone.
+
+    When the rowwise columns cover different numbers of samples the zip used to
+    truncate in silence, dropping samples and leaving the survivors carrying
+    names that belong to other rows. It must fail loudly instead. This is what
+    put WLO56 at 0.5 cm in the Okoboji load.
+    """
+
+    NAMES = ["WLO17", "WLO18", "WLO19"]
+
+    @pytest.fixture
+    def named_samples(self):
+        """A file with one rowwise samplename column, so row counts are visible."""
+        yml_dict = {
+            "metadata": [
+                {
+                    "column": "SampleName",
+                    "neotoma": "ndb.samples.samplename",
+                    "rowwise": True,
+                    "type": "string",
+                }
+            ]
+        }
+        csv_file = [{"SampleName": n} for n in self.NAMES]
+        return csv_file, yml_dict
+
+    def test_fewer_analysis_units_than_samples_is_invalid(self, mock_cur, named_samples):
+        csv_file, yml_dict = named_samples
+        databus = {
+            # One analysis unit short of the three named samples.
+            "analysisunits": MagicMock(id_list=[1, 2], counter=2),
+            "datasets": MagicMock(id_int=1),
+        }
+        result = nv.valid_sample(
+            cur=mock_cur, yml_dict=yml_dict, csv_file=csv_file, databus=databus
+        )
+        assert False in result.valid
+        assert any("mismatched row counts" in m for m in result.message)
+        # No samples are built from a mismatched file, rather than a silent two.
+        assert result.counter == 0
+
+    def test_message_names_both_counts(self, mock_cur, named_samples):
+        csv_file, yml_dict = named_samples
+        databus = {
+            "analysisunits": MagicMock(id_list=[1, 2], counter=2),
+            "datasets": MagicMock(id_int=1),
+        }
+        result = nv.valid_sample(
+            cur=mock_cur, yml_dict=yml_dict, csv_file=csv_file, databus=databus
+        )
+        msg = next(m for m in result.message if "mismatched row counts" in m)
+        assert "samplename=3" in msg
+        assert "analysisunitid=2" in msg
+
+    def test_matching_row_counts_pass_the_guard(self, mock_cur, named_samples):
+        csv_file, yml_dict = named_samples
+        mock_cur.mock_fetchone = (10,)
+        databus = {
+            "analysisunits": MagicMock(id_list=[1, 2, 3]),
+            "datasets": MagicMock(id_int=1),
+        }
+        result = nv.valid_sample(
+            cur=mock_cur, yml_dict=yml_dict, csv_file=csv_file, databus=databus
+        )
+        assert not any("mismatched row counts" in m for m in result.message)
+        assert result.counter == len(self.NAMES)
diff --git a/tests/test_valid_sample_age.py b/tests/test_valid_sample_age.py
index 8aeef6d..8921684 100644
--- a/tests/test_valid_sample_age.py
+++ b/tests/test_valid_sample_age.py
@@ -8,6 +8,60 @@
 from DataBUS import Response
 
 
+class TestChronologyScalarBroadcast:
+    """A chronology-wide scalar must not truncate the per-sample age rows.
+
+    ``agemodel`` is pulled once for the whole chronology while ages are pulled
+    per sample. Wrapping the scalar in a one-element list made the zip stop after
+    a single row, so a 42-sample dataset landed exactly one sample age.
+    """
+
+    AGES = ["100", "200", "300"]
+
+    @pytest.fixture
+    def crs_pair(self):
+        yml_dict = {
+            "metadata": [
+                {
+                    "column": "Age",
+                    "neotoma": "ndb.sampleages.age",
+                    "rowwise": True,
+                    "type": "float",
+                    "chronologyname": "DefaultChronology",
+                },
+                {
+                    "column": "AgeModel",
+                    "neotoma": "ndb.sampleages.agemodel",
+                    "rowwise": False,
+                    "type": "string",
+                    "chronologyname": "DefaultChronology",
+                },
+            ]
+        }
+        csv_file = [{"Age": a, "AgeModel": "CRS"} for a in self.AGES]
+        return csv_file, yml_dict
+
+    def test_every_sample_gets_an_age(self, mock_cur, crs_pair):
+        csv_file, yml_dict = crs_pair
+        mock_cur.mock_fetchone = (1,)
+        # `name` is a reserved MagicMock constructor kwarg, so it has to be
+        # assigned after construction to become a real attribute.
+        chronologies = MagicMock()
+        chronologies.name = {"DefaultChronology": 5}
+        databus = {
+            "chronologies": chronologies,
+            "samples": MagicMock(id_list=[11, 12, 13]),
+        }
+        nv.valid_sample_age(cur=mock_cur, yml_dict=yml_dict, csv_file=csv_file, databus=databus)
+
+        inserts = [
+            params for q, params in mock_cur._execute_calls if "insertsampleage" in q.lower()
+        ]
+        assert len(inserts) == len(self.AGES)
+        assert {p["sampleid"] for p in inserts} == {11, 12, 13}
+        assert {p["age"] for p in inserts} == {100.0, 200.0, 300.0}
+
+
 class TestValidSampleAgeMock:
     def test_returns_response_no_sample_ages(self, mock_cur):
         result = nv.valid_sample_age(
diff --git a/tests/test_valid_sequence.py b/tests/test_valid_sequence.py
new file mode 100644
index 0000000..6d17136
--- /dev/null
+++ b/tests/test_valid_sequence.py
@@ -0,0 +1,336 @@
+"""Tests for the valid_sequence validator and its extraction helpers.
+
+``valid_sequence`` runs after ``valid_data``. For every DNA entry in the
+template it inserts one ``ndb.sequences`` row, one ``ndb.sequencedata`` row per
+dataid the taxon produced, and one ``ndb.aednamodels`` row carrying the taxon
+call. The tests drive it with a routing cursor because each of those steps is a
+different statement.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+from DataBUS.neotomaValidator.valid_sequence import (
+    _build_key_to_dna_map,
+    _extract_dna_entries,
+    _extract_model_info,
+)
+
+
+class RoutingCursor:
+    """Answers fetchone() based on a substring match against the last query."""
+
+    def __init__(self, routes=None, rowcount=0):
+        self.routes = routes or {}
+        self.calls = []
+        self.rowcount = rowcount
+        self._next = None
+
+    def execute(self, query, params=None):
+        self.calls.append((query, params))
+        self._next = None
+        for marker, value in self.routes.items():
+            if marker in query:
+                self._next = value
+                break
+
+    def fetchone(self):
+        return self._next
+
+    def fetchall(self):
+        return []
+
+    def queries_matching(self, marker):
+        return [p for q, p in self.calls if marker in q]
+
+
+def _dna_entry(taxonname, dnasequence, asv, column=None):
+    return {
+        "neotoma": "ndb.data.value",
+        "taxonname": taxonname,
+        "dnasequence": dnasequence,
+        "asv": asv,
+        "column": column or asv,
+    }
+
+
+def _seq_yml(entries=None, model="DADA2", superseeds=None):
+    metadata = list(entries if entries is not None else [_dna_entry("Picea", "ACGTACGT", "ASV1")])
+    if model is not None:
+        entry = {"neotoma": "ndb.aednamodels.modelid", "value": model}
+        if superseeds is not None:
+            entry["superseeds"] = superseeds
+        metadata.append(entry)
+    return {"metadata": metadata}
+
+
+def _databus(id_dict, datasetid=74655):
+    datasets, data = Response(), Response()
+    datasets.id_int = datasetid
+    data.id_dict = id_dict
+    return {"datasets": datasets, "data": data}
+
+
+# Sequence insert -> 500, model insert -> 900, taxonid lookup -> 42.
+BASE_ROUTES = {
+    "INSERT INTO ndb.sequences": (500,),
+    "INSERT INTO ndb.aednamodels": (900,),
+    "SELECT v.taxonid": (42,),
+}
+
+
+class TestExtractDnaEntries:
+    def test_no_entries(self):
+        assert _extract_dna_entries({"metadata": []}) == []
+
+    def test_ignores_entries_without_both_dnasequence_and_asv(self):
+        yml = {
+            "metadata": [
+                {"neotoma": "ndb.data.value", "taxonname": "Picea", "asv": "ASV1"},
+                {"neotoma": "ndb.data.value", "taxonname": "Picea", "dnasequence": "ACGT"},
+                {"neotoma": "ndb.sites.sitename", "dnasequence": "ACGT", "asv": "ASV1"},
+            ]
+        }
+        assert _extract_dna_entries(yml) == []
+
+    def test_collects_complete_entries(self):
+        yml = {"metadata": [_dna_entry("Picea", "ACGTACGT", "ASV1", column="col_a")]}
+        assert _extract_dna_entries(yml) == [
+            {
+                "taxonname": "Picea",
+                "dnasequence": "ACGTACGT",
+                "asv": "ASV1",
+                "column": "col_a",
+            }
+        ]
+
+
+class TestExtractModelInfo:
+    def test_absent_model(self):
+        assert _extract_model_info({"metadata": []}) == (None, [])
+
+    def test_model_without_superseeds(self):
+        yml = {"metadata": [{"neotoma": "ndb.aednamodels.modelid", "value": "DADA2"}]}
+        assert _extract_model_info(yml) == ("DADA2", [])
+
+    def test_superseeds_string_is_split_and_stripped(self):
+        yml = {
+            "metadata": [
+                {
+                    "neotoma": "ndb.aednamodels.modelid",
+                    "value": "DADA2",
+                    "superseeds": "OBITools, mothur ",
+                }
+            ]
+        }
+        assert _extract_model_info(yml) == ("DADA2", ["OBITools", "mothur"])
+
+    def test_superseeds_list_is_kept(self):
+        yml = {
+            "metadata": [
+                {
+                    "neotoma": "ndb.aednamodels.modelid",
+                    "value": "DADA2",
+                    "superseeds": ["OBITools"],
+                }
+            ]
+        }
+        assert _extract_model_info(yml) == ("DADA2", ["OBITools"])
+
+
+class TestBuildKeyToDnaMap:
+    def test_prefers_compound_key(self):
+        entries = _extract_dna_entries(_seq_yml())
+        mapping = _build_key_to_dna_map(entries, {"Picea::ASV1": [1], "Picea": [2]})
+        assert list(mapping) == ["Picea::ASV1"]
+
+    def test_falls_back_to_plain_taxon_key(self):
+        entries = _extract_dna_entries(_seq_yml())
+        mapping = _build_key_to_dna_map(entries, {"Picea": [2]})
+        assert mapping == {"Picea": {"dnasequence": "ACGTACGT", "asv": "ASV1"}}
+
+    def test_unmatched_entry_is_dropped(self):
+        entries = _extract_dna_entries(_seq_yml())
+        assert _build_key_to_dna_map(entries, {"Abies": [1]}) == {}
+
+    def test_two_asvs_for_one_taxon_keep_separate_compound_keys(self):
+        yml = _seq_yml(
+            entries=[
+                _dna_entry("Picea", "ACGT", "ASV1"),
+                _dna_entry("Picea", "TGCA", "ASV2"),
+            ]
+        )
+        entries = _extract_dna_entries(yml)
+        mapping = _build_key_to_dna_map(entries, {"Picea::ASV1": [1], "Picea::ASV2": [2]})
+        assert mapping["Picea::ASV1"]["dnasequence"] == "ACGT"
+        assert mapping["Picea::ASV2"]["dnasequence"] == "TGCA"
+
+    def test_plain_key_taken_only_once(self):
+        """With no compound keys, the first entry wins and the second is dropped."""
+        yml = _seq_yml(
+            entries=[
+                _dna_entry("Picea", "ACGT", "ASV1"),
+                _dna_entry("Picea", "TGCA", "ASV2"),
+            ]
+        )
+        entries = _extract_dna_entries(yml)
+        mapping = _build_key_to_dna_map(entries, {"Picea": [1]})
+        assert mapping == {"Picea": {"dnasequence": "ACGT", "asv": "ASV1"}}
+
+
+class TestValidSequence:
+    def test_no_dna_entries_is_valid(self):
+        cur = RoutingCursor()
+        response = nv.valid_sequence(
+            cur, {"metadata": []}, csv_file=None, databus=_databus({"Picea": [1]})
+        )
+
+        assert response.validAll
+        assert response.message == ["? No aeDNA sequence entries found in template."]
+        assert cur.calls == []
+
+    def test_missing_dataset_id_aborts(self):
+        response = nv.valid_sequence(RoutingCursor(), _seq_yml(), csv_file=None, databus={})
+
+        assert not response.validAll
+        assert any("Dataset ID not available" in m for m in response.message)
+
+    def test_missing_data_ids_aborts(self):
+        datasets = Response()
+        datasets.id_int = 74655
+        response = nv.valid_sequence(
+            RoutingCursor(), _seq_yml(), csv_file=None, databus={"datasets": datasets}
+        )
+
+        assert not response.validAll
+        assert any("Data IDs not available" in m for m in response.message)
+
+    def test_no_matching_data_entries_is_valid(self):
+        response = nv.valid_sequence(
+            RoutingCursor(), _seq_yml(), csv_file=None, databus=_databus({"Abies": [1]})
+        )
+
+        assert response.validAll
+        assert any("Could not match any DNA entries" in m for m in response.message)
+
+    def test_full_insert_chain(self):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        response = nv.valid_sequence(
+            cur, _seq_yml(), csv_file=None, databus=_databus({"Picea::ASV1": [11, 12]})
+        )
+
+        assert response.validAll
+        assert response.id_dict == {"Picea::ASV1": {"sequenceid": 500, "modelid": 900}}
+        assert cur.queries_matching("INSERT INTO ndb.sequences") == [
+            {"datasetid": 74655, "sequence": "ACGTACGT", "asv": "ASV1", "primername": None}
+        ]
+
+    def test_one_sequencedata_row_per_dataid(self):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        nv.valid_sequence(
+            cur, _seq_yml(), csv_file=None, databus=_databus({"Picea::ASV1": [11, 12, 13]})
+        )
+
+        assert cur.queries_matching("INSERT INTO ndb.sequencedata") == [
+            {"dataid": 11, "sequenceid": 500},
+            {"dataid": 12, "sequenceid": 500},
+            {"dataid": 13, "sequenceid": 500},
+        ]
+
+    def test_model_row_carries_taxonid_from_first_dataid(self):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        nv.valid_sequence(
+            cur, _seq_yml(), csv_file=None, databus=_databus({"Picea::ASV1": [11, 12]})
+        )
+
+        assert cur.queries_matching("SELECT v.taxonid") == [{"dataid": 11}]
+        (model,) = cur.queries_matching("INSERT INTO ndb.aednamodels")
+        assert model["sequenceid"] == 500
+        assert model["taxonid"] == 42
+        assert model["model"] == "DADA2"
+
+    def test_no_model_entry_skips_aednamodels(self):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        response = nv.valid_sequence(
+            cur,
+            _seq_yml(model=None),
+            csv_file=None,
+            databus=_databus({"Picea::ASV1": [11]}),
+        )
+
+        assert any("No aeDNA model entry" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.aednamodels") == []
+        assert response.id_dict == {"Picea::ASV1": {"sequenceid": 500, "modelid": None}}
+
+    def test_missing_taxonid_leaves_model_none(self):
+        routes = dict(BASE_ROUTES)
+        routes["SELECT v.taxonid"] = None
+        cur = RoutingCursor(routes)
+        response = nv.valid_sequence(
+            cur, _seq_yml(), csv_file=None, databus=_databus({"Picea::ASV1": [11]})
+        )
+
+        assert not response.validAll
+        assert any("Could not retrieve taxonid for dataid 11" in m for m in response.message)
+        assert response.id_dict == {"Picea::ASV1": {"sequenceid": 500, "modelid": None}}
+
+    def test_superseded_models_are_updated(self):
+        cur = RoutingCursor(dict(BASE_ROUTES), rowcount=2)
+        response = nv.valid_sequence(
+            cur,
+            _seq_yml(superseeds="OBITools"),
+            csv_file=None,
+            databus=_databus({"Picea::ASV1": [11]}),
+        )
+
+        updates = cur.queries_matching("UPDATE ndb.aednamodels")
+        assert updates == [{"new_modelid": 900, "taxonid": 42, "old_model": "OBITools"}]
+        assert any("2 previous model(s) marked as superseded" in m for m in response.message)
+
+    def test_empty_dataid_list_is_skipped(self):
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        response = nv.valid_sequence(
+            cur, _seq_yml(), csv_file=None, databus=_databus({"Picea::ASV1": []})
+        )
+
+        assert response.validAll
+        assert any("No DNA data entries matched" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.sequences") == []
+
+    def test_sequence_insert_failure_skips_the_entry(self):
+        class Failing(RoutingCursor):
+            def execute(self, query, params=None):
+                super().execute(query, params)
+                if "INSERT INTO ndb.sequences" in query:
+                    raise RuntimeError("duplicate key")
+
+        cur = Failing(dict(BASE_ROUTES))
+        response = nv.valid_sequence(
+            cur, _seq_yml(), csv_file=None, databus=_databus({"Picea::ASV1": [11]})
+        )
+
+        assert not response.validAll
+        assert any("Sequence cannot be inserted" in m for m in response.message)
+        assert cur.queries_matching("INSERT INTO ndb.sequencedata") == []
+        assert response.id_dict == {}
+
+    def test_two_taxa_each_get_their_own_sequence(self):
+        yml = _seq_yml(
+            entries=[
+                _dna_entry("Picea", "ACGT", "ASV1"),
+                _dna_entry("Abies", "TGCA", "ASV2"),
+            ]
+        )
+        cur = RoutingCursor(dict(BASE_ROUTES))
+        response = nv.valid_sequence(
+            cur,
+            yml,
+            csv_file=None,
+            databus=_databus({"Picea::ASV1": [11], "Abies::ASV2": [22]}),
+        )
+
+        assert response.validAll
+        assert set(response.id_dict) == {"Picea::ASV1", "Abies::ASV2"}
+        sequences = cur.queries_matching("INSERT INTO ndb.sequences")
+        assert [s["asv"] for s in sequences] == ["ASV1", "ASV2"]
diff --git a/tests/test_valid_speleothem.py b/tests/test_valid_speleothem.py
new file mode 100644
index 0000000..beb5310
--- /dev/null
+++ b/tests/test_valid_speleothem.py
@@ -0,0 +1,193 @@
+"""Tests for valid_speleothem and valid_external_speleothem.
+
+``valid_speleothem`` resolves every ``*id`` field whose CSV value is a name
+(drip type, entity status, rock type, …) against its own vocabulary table
+before building the Speleothem entity. ``valid_external_speleothem`` links that
+entity to a record in an outside database such as SISAL.
+"""
+
+import pytest
+
+import DataBUS.neotomaValidator as nv
+from DataBUS import Response
+
+
+def _entry(table, field, column, type_spec="string"):
+    return {
+        "neotoma": f"ndb.{table}.{field}",
+        "column": column,
+        "rowwise": False,
+        "required": False,
+        "type": type_spec,
+    }
+
+
+def _speleothem(**overrides):
+    """(yml_dict, csv_file) for a speleothem entity.
+
+    ``Speleothem`` refuses to build without a siteid and a speleothemtypeid, so
+    the type is always present unless a test explicitly drops it by passing
+    ``speleothemtypeid=None``.
+    """
+    fields = {"entityname": "Cave-1", "speleothemtypeid": "stalagmite"}
+    fields.update(overrides)
+    fields = {k: v for k, v in fields.items() if v is not None}
+    yml = {"metadata": [_entry("speleothems", k, k) for k in fields]}
+    return yml, [dict(fields)]
+
+
+def _fetches(cur, *values):
+    """Return each value in turn from cur.fetchone(), then None."""
+    it = iter(values)
+    cur.fetchone = lambda: next(it, None)
+
+
+def _external(**fields):
+    yml = {"metadata": [_entry("externalspeleothemdata", k, k) for k in fields]}
+    return yml, [dict(fields)]
+
+
+@pytest.fixture
+def site_databus():
+    sites = Response()
+    sites.id_int = 5001
+    return {"sites": sites}
+
+
+@pytest.fixture
+def speleothem_databus():
+    speleothems = Response()
+    speleothems.id_int = 88
+    return {"speleothems": speleothems}
+
+
+def _params(cur, marker):
+    return [p for q, p in cur._execute_calls if p and marker in q]
+
+
+class TestValidSpeleothem:
+    def test_no_parameters_is_valid(self, mock_cur, site_databus):
+        response = nv.valid_speleothem(
+            mock_cur, {"metadata": []}, csv_file=[], databus=site_databus
+        )
+
+        assert response.validAll
+        assert response.message == ["✔ No speleothem parameters provided."]
+        assert mock_cur._execute_calls == []
+
+    def test_entity_inserted_against_site(self, mock_cur, site_databus):
+        _fetches(mock_cur, (7,), (88,))
+        yml, csv_file = _speleothem(monitoring="yes")
+        response = nv.valid_speleothem(mock_cur, yml, csv_file, databus=site_databus)
+
+        assert response.validAll
+        assert response.id_int == 88
+        assert "✔ Speleothem can be created." in response.message
+        (insert,) = _params(mock_cur, "insert_speleothem(")
+        assert insert["siteid"] == 5001
+        assert insert["entityname"] == "Cave-1"
+
+    def test_monitoring_yes_becomes_true(self, mock_cur, site_databus):
+        _fetches(mock_cur, (7,), (88,))
+        yml, csv_file = _speleothem(monitoring="yes")
+        nv.valid_speleothem(mock_cur, yml, csv_file, databus=site_databus)
+
+        assert _params(mock_cur, "insert_speleothem(")[0]["monitoring"] is True
+
+    def test_monitoring_anything_else_becomes_false(self, mock_cur, site_databus):
+        _fetches(mock_cur, (7,), (88,))
+        yml, csv_file = _speleothem(monitoring="no")
+        nv.valid_speleothem(mock_cur, yml, csv_file, databus=site_databus)
+
+        assert _params(mock_cur, "insert_speleothem(")[0]["monitoring"] is False
+
+    def test_blank_monitoring_becomes_false(self, mock_cur, site_databus):
+        """An unfilled monitoring column must not crash the validator."""
+        _fetches(mock_cur, (7,), (88,))
+        yml, csv_file = _speleothem()
+        response = nv.valid_speleothem(mock_cur, yml, csv_file, databus=site_databus)
+
+        assert response.validAll
+        assert _params(mock_cur, "insert_speleothem(")[0]["monitoring"] is False
+
+    def test_named_vocabulary_field_is_resolved(self, mock_cur, site_databus):
+        _fetches(mock_cur, (7,), (88,))
+        yml, csv_file = _speleothem(speleothemtypeid="stalagmite")
+        response = nv.valid_speleothem(mock_cur, yml, csv_file, databus=site_databus)
+
+        assert _params(mock_cur, "ndb.speleothemtypes") == [{"element": "stalagmite"}]
+        assert _params(mock_cur, "insert_speleothem(")[0]["speleothemtypeid"] == 7
+        assert any("speleothemtypeid for 7 found" in m for m in response.message)
+
+    def test_unknown_vocabulary_value_is_invalid(self, mock_cur, site_databus):
+        mock_cur.mock_fetchone = None
+        yml, csv_file = _speleothem(speleothemtypeid="not a type")
+        response = nv.valid_speleothem(mock_cur, yml, csv_file, databus=site_databus)
+
+        assert not response.validAll
+        assert any("speleothemtypeid for not a type not found" in m for m in response.message)
+
+    def test_missing_site_uses_placeholder(self, mock_cur):
+        _fetches(mock_cur, (7,), (88,))
+        yml, csv_file = _speleothem()
+        response = nv.valid_speleothem(mock_cur, yml, csv_file, databus={})
+
+        assert not response.validAll
+        assert any("Site ID not available" in m for m in response.message)
+        assert _params(mock_cur, "insert_speleothem(")[0]["siteid"] == 1
+
+
+class TestValidExternalSpeleothem:
+    def test_no_parameters_is_valid(self, mock_cur, speleothem_databus):
+        response = nv.valid_external_speleothem(
+            mock_cur, {"metadata": []}, csv_file=[], databus=speleothem_databus
+        )
+
+        assert response.validAll
+        assert any("No external speleothem parameters" in m for m in response.message)
+        assert mock_cur._execute_calls == []
+
+    def test_links_entity_to_external_record(self, mock_cur, speleothem_databus):
+        mock_cur.mock_fetchone = (3,)
+        yml, csv_file = _external(extdatabaseid="SISAL", externalid="entity_42")
+        response = nv.valid_external_speleothem(mock_cur, yml, csv_file, databus=speleothem_databus)
+
+        assert response.validAll
+        assert "✔  ExternalSpeleothem inserted." in response.message
+        (insert,) = _params(mock_cur, "insert_externalspeleothem(")
+        assert insert["entityid"] == 88
+        assert insert["extdatabaseid"] == 3
+        assert insert["externalid"] == "entity_42"
+
+    def test_database_name_lookup_is_lowercased(self, mock_cur, speleothem_databus):
+        mock_cur.mock_fetchone = (3,)
+        yml, csv_file = _external(extdatabaseid="SISAL", externalid="entity_42")
+        nv.valid_external_speleothem(mock_cur, yml, csv_file, databus=speleothem_databus)
+
+        assert _params(mock_cur, "ndb.externaldatabases") == [{"extdatabaseid": "sisal"}]
+
+    def test_unknown_external_database_is_invalid(self, mock_cur, speleothem_databus):
+        mock_cur.mock_fetchone = None
+        yml, csv_file = _external(extdatabaseid="Nowhere", externalid="entity_42")
+        response = nv.valid_external_speleothem(mock_cur, yml, csv_file, databus=speleothem_databus)
+
+        assert not response.validAll
+        assert any("extdatabaseid for Nowhere not found" in m for m in response.message)
+
+    def test_description_is_stripped(self, mock_cur, speleothem_databus):
+        mock_cur.mock_fetchone = (3,)
+        yml, csv_file = _external(
+            extdatabaseid="SISAL", externalid="entity_42", externaldescription="  a cave, "
+        )
+        nv.valid_external_speleothem(mock_cur, yml, csv_file, databus=speleothem_databus)
+
+        assert _params(mock_cur, "insert_externalspeleothem(")[0]["externaldescription"] == "a cave"
+
+    def test_missing_speleothem_uses_placeholder_entity(self, mock_cur):
+        mock_cur.mock_fetchone = (3,)
+        yml, csv_file = _external(extdatabaseid="SISAL", externalid="entity_42")
+        response = nv.valid_external_speleothem(mock_cur, yml, csv_file, databus={})
+
+        assert not response.validAll
+        assert any("Speleothem entity ID not available" in m for m in response.message)
+        assert _params(mock_cur, "insert_externalspeleothem(")[0]["entityid"] == 2