07 — Schema Contracts: Pydantic Models + Contract Tests

Source of truth for CodingSOPs §7 + TestingSOPs Layer 3 compliance at the L1→L2 boundary. The "schema lock" risk (L1 produces industry, L2 expects industry_vertical) is eliminated here — not in documentation, not in convention, but in code that fails loudly at commit time.

The two-layer type safety model (from CodingSOPs §7)

pyright --strict     → catches field name drift and type mismatches at WRITE TIME
Pydantic v2          → catches bad data shapes at RUNTIME when records cross boundaries
Contract tests       → validates L1 output satisfies L2 input schema on every COMMIT

None of these is optional. All three run before any merge.

Verified schema source

All field names, types, and values below are derived from: - Live query of algolia-central_enterprise_ledger (app 0EXRPAXB56) — 2026-04-30 - n8n Firehose v2 (Normalize fields + Classify Content nodes) - n8n Refinery (Split & Inject, Chunk Configurator nodes)

Not from assumption or training data. See 03-record-extraction.md for the full field reference.

Layer 1 module boundary: CrawlerRecord (L1 output)

This is what recordExtractor returns — expressed as a Pydantic model. The JavaScript extractor produces a dict that lands in Algolia; the Python module that reads from Algolia (for L2 handoff) must parse it through this model.

# packages/l1_crawler/src/l1_crawler/models.py
from __future__ import annotations
from pydantic import BaseModel, Field, model_validator
from enum import Enum
import time


class SourceType(str, Enum):
    """source_type taxonomy — from n8n Firehose URL routing, verified live in index."""
    DOC = "doc"
    SUPPORT = "support"
    BLOG = "blog"
    OTHER = "other"
    DEVELOPER = "developer"
    CUSTOMER_STORY = "customer_story"
    ACADEMY = "academy"
    RESOURCE = "resource"
    CHANGELOG = "changelog"


class RecordStatus(str, Enum):
    PENDING = "pending"     # set by L1; L2 reads this to find work
    INDEXED = "indexed"     # set by L2 after enrichment + chunking


class CrawlerRecord(BaseModel):
    """
    Canonical output of the L1 Algolia Crawler — one parent doc record per indexed page.

    Schema verified against live algolia-central_enterprise_ledger (app 0EXRPAXB56) on 2026-04-30.
    is_chunk is always False here — chunks are created by L2 Enrichment.
    """
    version: str = "1.0.0"

    # Identity
    object_id: str = Field(min_length=1, max_length=64, alias="objectID")
    record_type: str = "enterprise_ledger"
    url: str = Field(min_length=10)

    # Content — L1 populates these
    title: str = Field(min_length=1)
    description: str = ""
    content: str = Field(min_length=300)  # quality gate: <300 chars = skip
    source_type: SourceType
    language_code: str = "en"
    is_chunk: bool = False
    status: RecordStatus = RecordStatus.PENDING
    created_at: int = Field(default_factory=lambda: int(time.time()))
    updated_at: int = Field(default_factory=lambda: int(time.time()))

    # L2 placeholder fields — MUST be None from L1
    # L2 Enrichment fills these; values are proper-case strings (not snake_case)
    summary: str | None = None          # GPT 3-sentence summary
    product_tag: str | None = None      # "AI Search", "InstantSearch", "Analytics", etc.
    feature_tag: str | None = None      # "Rules", "Faceting", "A/B Testing", etc.
    solution_tag: str | None = None     # "Site Search", "Analytics", etc.
    customer_tag: str | None = None
    industry_tag: str | None = None

    model_config = {"populate_by_name": True}

    @model_validator(mode="after")
    def l2_fields_must_be_null(self) -> "CrawlerRecord":
        """L1 must never pre-populate L2 classification fields."""
        assert self.summary is None, "L1 must not set summary — that is L2 (GPT)"
        assert self.product_tag is None, "L1 must not set product_tag — that is L2"
        assert self.feature_tag is None, "L1 must not set feature_tag — that is L2"
        assert self.solution_tag is None, "L1 must not set solution_tag — that is L2"
        assert self.is_chunk is False, "L1 never creates chunk records"
        assert self.status == RecordStatus.PENDING, "L1 must set status=pending"
        return self

Layer 2 module boundary: EnrichmentInput (L2 input)

This is what the L2 Enrichment module accepts. It reads from Algolia where status=pending. It must be parseable from any CrawlerRecord with no field-name translation.

# packages/l2_enrichment/src/l2_enrichment/models.py
from __future__ import annotations
from pydantic import BaseModel, Field
from l1_crawler.models import SourceType, RecordStatus


class EnrichmentInput(BaseModel):
    """
    What L2 Enrichment reads from Algolia — records where status=pending.
    Mirrors CrawlerRecord exactly for the fields L2 needs; L2 fields are absent (not set yet).

    Verified against live index schema 2026-04-30.
    """
    version: str = "1.0.0"

    object_id: str = Field(alias="objectID")
    record_type: str
    url: str
    title: str
    content: str        # full page text — L2 uses this for summarization + chunking
    source_type: SourceType
    language_code: str
    is_chunk: bool
    status: RecordStatus  # must be "pending" — L2 only processes pending

    @classmethod
    def from_algolia_record(cls, record: dict) -> "EnrichmentInput":
        """Parse a raw Algolia hit dict. Fails loudly if schema has drifted."""
        return cls.model_validate(record)

Layer 2 output: EnrichedRecord

The full record L2 writes back to Algolia after enrichment (parent) + chunks:

# packages/l2_enrichment/src/l2_enrichment/models.py

class EnrichedRecord(BaseModel):
    """Parent doc record after L2 enrichment. Written back via saveObjects."""

    object_id: str = Field(alias="objectID")
    record_type: str = "enterprise_ledger"
    url: str
    title: str
    content: str
    summary: str            # GPT 3-sentence summary — L2 fills
    source_type: SourceType
    language_code: str
    is_chunk: bool = False
    status: RecordStatus = RecordStatus.INDEXED
    updated_at: int

    # Classification — L2 fills, proper case strings
    product_tag: str | None = None
    feature_tag: str | None = None
    solution_tag: str | None = None
    customer_tag: str | None = None
    industry_tag: str | None = None


class ChunkRecord(BaseModel):
    """Chunk record created by L2. objectID = '{parent_uuid}_chunk_{index}'"""

    object_id: str = Field(alias="objectID")   # "{parent_id}_chunk_{chunk_index}"
    record_type: str = "enterprise_ledger"
    parent_id: str
    chunk_index: int = Field(ge=0)
    chunk_total: int = Field(ge=1)
    url: str
    title: str          # "{original_title} (Part {n}/{total})"
    content: str        # "Document Context: {summary}\n\nFragment: {chunk_text}"
    summary: str
    source_type: SourceType
    language_code: str
    is_chunk: bool = True
    status: RecordStatus = RecordStatus.INDEXED

    # Tags inherited from parent
    product_tag: str | None = None
    feature_tag: str | None = None
    solution_tag: str | None = None
    customer_tag: str | None = None
    industry_tag: str | None = None

    @model_validator(mode="after")
    def content_must_have_context_prefix(self) -> "ChunkRecord":
        assert self.content.startswith("Document Context:"), \
            "Chunk content must start with 'Document Context: {summary}'"
        return self

Contract test: L1 output satisfies L2 input

# packages/l1_crawler/tests/contract/test_l1_l2_contract.py
import hashlib
import pytest
import time
from l1_crawler.models import CrawlerRecord, SourceType, RecordStatus
from l2_enrichment.models import EnrichmentInput, EnrichedRecord, ChunkRecord


def make_sample_crawler_record() -> dict:
    """Minimal valid L1 record matching the live RC2 Algolia schema."""
    url = "https://www.algolia.com/doc/guides/building-search-ui/upgrade-guides/js/"
    return {
        "objectID": hashlib.sha256(url.encode()).hexdigest()[:36],
        "record_type": "enterprise_ledger",
        "url": url,
        "title": "Upgrade InstantSearch.js - Algolia",
        "description": "How to upgrade InstantSearch.js",
        "content": "This guide explains how to upgrade InstantSearch.js from v3 to v4. " * 10,
        "source_type": "doc",
        "language_code": "en",
        "is_chunk": False,
        "status": "pending",
        "created_at": int(time.time()),
        "updated_at": int(time.time()),
        # L2 fields — all null from L1
        "summary": None,
        "product_tag": None,
        "feature_tag": None,
        "solution_tag": None,
        "customer_tag": None,
        "industry_tag": None,
    }


class TestL1L2Contract:
    def test_crawler_record_is_valid(self):
        """CrawlerRecord parses cleanly from a well-formed record."""
        record = make_sample_crawler_record()
        parsed = CrawlerRecord.model_validate(record)
        assert parsed.status == RecordStatus.PENDING
        assert parsed.is_chunk is False

    def test_crawler_record_satisfies_enrichment_input(self):
        """L1 output is directly parseable as L2 EnrichmentInput — no field mapping needed."""
        record = make_sample_crawler_record()
        enrichment_input = EnrichmentInput.from_algolia_record(record)
        assert enrichment_input.url == record["url"]
        assert enrichment_input.source_type == SourceType.DOC
        assert enrichment_input.status == RecordStatus.PENDING

    def test_l1_cannot_set_summary(self):
        """summary is L2's job — L1 setting it fails the validator."""
        record = make_sample_crawler_record()
        record["summary"] = "This page explains InstantSearch.js."
        with pytest.raises(AssertionError):
            CrawlerRecord.model_validate(record)

    def test_l1_cannot_set_product_tag(self):
        """product_tag is L2's job — L1 setting it fails the validator."""
        record = make_sample_crawler_record()
        record["product_tag"] = "InstantSearch"
        with pytest.raises(AssertionError):
            CrawlerRecord.model_validate(record)

    def test_l1_cannot_set_status_indexed(self):
        """L1 must set status=pending, never indexed."""
        record = make_sample_crawler_record()
        record["status"] = "indexed"
        with pytest.raises(AssertionError):
            CrawlerRecord.model_validate(record)

    def test_missing_title_fails(self):
        """Quality gate: empty title must be rejected."""
        record = make_sample_crawler_record()
        record["title"] = ""
        with pytest.raises(Exception):
            CrawlerRecord.model_validate(record)

    def test_short_content_fails(self):
        """Quality gate: content < 300 chars must be rejected."""
        record = make_sample_crawler_record()
        record["content"] = "Too short."
        with pytest.raises(Exception):
            CrawlerRecord.model_validate(record)

    def test_invalid_source_type_fails(self):
        """source_type must be from the SourceType enum."""
        record = make_sample_crawler_record()
        record["source_type"] = "marketing"  # old n8n content_type value, not valid source_type
        with pytest.raises(Exception):
            CrawlerRecord.model_validate(record)

    def test_objectid_determinism(self):
        """Same URL must always produce same objectID — dedup guarantee."""
        url = "https://www.algolia.com/doc/guides/building-search-ui/upgrade-guides/js/"
        oid1 = hashlib.sha256(url.encode()).hexdigest()[:36]
        oid2 = hashlib.sha256(url.encode()).hexdigest()[:36]
        assert oid1 == oid2

    def test_chunk_content_format(self):
        """Chunk content must start with 'Document Context:' prefix."""
        bad_chunk = {
            "objectID": "some-uuid_chunk_0",
            "parent_id": "some-uuid",
            "chunk_index": 0,
            "chunk_total": 2,
            "url": "https://example.com",
            "title": "Example (Part 1/2)",
            "content": "Fragment text without context prefix",  # wrong format
            "summary": "A 3-sentence summary.",
            "source_type": "doc",
            "language_code": "en",
            "is_chunk": True,
            "status": "indexed",
        }
        with pytest.raises(AssertionError):
            ChunkRecord.model_validate(bad_chunk)

Where this eliminates the schema lock risk

The problem I flagged: L1 produces industry, L2 expects industry_vertical → silent breakage.

How the contract closes it:

1. Both sides define Pydantic models — field name is code, not convention
2. from_crawler_record() in EnrichmentInput fails at parse time if field names don't align
3. test_crawler_record_satisfies_enrichment_input() runs on every commit — if L1 renames a field without updating L2's model, the contract test fails before the commit lands
4. pyright --strict catches record.industry_vertical vs record.industry at write time if someone uses the wrong name in code

The schema lock conversation becomes: "does the contract test pass?" If yes, L1 and L2 are aligned. Full stop.

Pre-commit + CI gates

# Run before every commit (CodingSOPs §9)
ruff format .
ruff check .
uv run pyright packages/ --strict
uv run pytest packages/ -k "contract" -v    # contract tests are fast, run always
uv run pytest packages/ -v                  # full suite

The contract tests (Layer 3) are designed to be fast (no live API calls, no crawl) and run on every commit, not just pre-merge.

What to do when n8n workflow reveals new L2 fields

1. Add field to CrawlerRecord with = None default (L1 remains unaware)
2. Add field to EnrichmentInput as required (L2 will populate it)
3. Update from_crawler_record() to map the field (or leave it as None for L2 to fill)
4. Add a contract test case for the new field
5. Run pyright --strict — it will catch any callers that don't handle the new field

This is the process. No whiteboard, no verbal agreement, no "we'll sync later." Code + test + commit.