Cocoindex Extraction Contract — TECH

Cocoindex Extraction Contract — TECH

Status: RATIFIED-S241 (Q-EX2 — discriminated-union Pydantic with ExtractByLlm typed output_type). Companion to ./PRODUCT.md; reference the numbered Behavior invariants there for product intent.

Context

This spec defines the typed extraction contract that gates the LLM extraction stage of the Knowledge Hub cocoindex pipeline (per docs/plans/phase-0-investigation/architecture/02-data-flow.md §3.1 row 3 — “LLM extraction”). Today, the production codebase has no recurring cocoindex flow — scripts/ontology-sync/parse-flow.py is the only file that demonstrates the layered fn-shape pattern (and is itself an executable stub, not a live flow). The legacy extraction surface is split across:

• lib/ai/classify.ts — TypeScript classifyContent chain (procedural per-call extraction; RATIFIED-RETIRE per 02-data-flow.md §3.3 row 1 + COCO.8).
• lib/extraction/ — TypeScript Q&A pair Pattern A/B parsers (RATIFIED-RETIRE-POST-PHEW-MIGRATION per 02-data-flow.md §10.1 + phase-b-prerequisite-2-cocoindex-deep-dive.md §4 Recommendation 1).
• lib/entities/ — TypeScript entity-classification chain (selective-RETIRE per phase-b-prerequisite-2-cocoindex-deep-dive.md §4 Recommendation 3 — KH classification retained, named-entity dedup migrates to cocoindex).

The target state is a Python scripts/cocoindex_pipeline/extraction.py module (path matches 02-data-flow.md §3 + canonical-pipeline PLAN.md §4.8 T8 subtask 5 lines 296-307) hosting the Pydantic class shapes declared in this contract. The LLM-extraction step itself is wired at flow scope inside the sibling scripts/cocoindex_pipeline/flow.py (T8 scope) via doc["..."].transform(cocoindex.functions.ExtractByLlm(...)) per phase-b-prerequisite-2a-cocoindex-examples.md lines 196-203 + 362-368 — see §3 below.

Sibling spec dependency. The failure-write path in §4.2 depends on the record_extraction_failure helper signature owned by docs/specs/id-36-cocoindex-ledger-api/TECH.md (T1.3 — drafted in parallel). This contract is consumed by that spec at the pipeline_runs write-back boundary; resolve the helper signature there before T8 implementation lands.

Relevant code for reference:

• scripts/ontology-sync/parse-flow.py (lines 1-148) — canonical layered fn-shape stub. The inner-tier parse_cv_frontmatter(content_text: str) (lines 98-110) demonstrates the content_text: str (NOT FileLike) signature mandated by S9 spike §7.1 + §7.2.
• lib/validation/schemas.ts:43-52 — VALID_CONTENT_TYPES re-export from lib/ontology/content-type-registry.ts. Source of truth for the classification variant’s content_type enum.
• lib/validation/schemas.ts:1495-1508 — VALID_ENTITY_TYPES (12 values). Source of truth for the entity_mention variant’s entity_type enum.
• docs/ontology/26-form-type.md lines 65-79 — form_type CV baseline values (11 values: 8 procurement + 3 non-procurement). Source of truth for the q_a_form variant’s form_type enum.
• docs/plans/phase-0-investigation/architecture/05-qa-flow.md §7.2 — question_matches.question_kind form-type discriminator (the column the drafter spec collided with — see B-1 fix at §2.1 below).
• lib/anthropic.ts:29 + scripts/kb_pipeline/config.py:29 — production Anthropic model is claude-opus-4-6 (NOT the claude-opus-4-7 Claude-Code-session model). Cited in §3.2 below.
• lib/pipeline/record-run.ts — recordPipelineRun() helper for pipeline_runs inserts (per CLAUDE.md Gotcha “Cron pipeline_runs inserts: Use recordPipelineRun()”). The validation-failure path in PRODUCT.md invariant 13 routes through this helper.

Proposed changes

1. Module layout

New Python module: scripts/cocoindex_pipeline/extraction.py. This file contains all Pydantic class shapes declared in this contract plus any @coco.fn(memo=True) inner-tier post-processing helpers (e.g. span normalisation, confidence-threshold filtering) that consume the typed ExtractByLlm output. The flow-scope wiring (source binding, doc["binary"].transform(ExtractByLlm(...)) call, op_id propagation, target binding) lives in scripts/cocoindex_pipeline/flow.py (covered by T8 of the canonical-pipeline PLAN — illustrated in §3 below for spec completeness, but implementation owned by T8).

Required project files touched in T1.2 (THIS spec’s draft scope is the file shapes; T8 is the implementation):

File	Status	Purpose
scripts/cocoindex_pipeline/__init__.py	NEW (T8)	Package marker
scripts/cocoindex_pipeline/extraction.py	NEW (T8)	Pydantic shapes + inner-tier @coco.fn post-processing helpers per this contract
scripts/cocoindex_pipeline/flow.py	NEW (T8)	Outer-tier flow scaffolding hosting the flow-scope ExtractByLlm(...) calls
scripts/cocoindex_pipeline/prompts.py	NEW (T8)	Prompt-template constants (memo-cascade per S9 §7.4)
scripts/tests/cocoindex_pipeline/test_extraction_contract.py	NEW (T8)	Acceptance tests per the §5 test plan in this spec

2. Pydantic class shapes

All shapes use Pydantic v2 (pydantic>=2.0). The discriminated union uses Annotated[Union[...], Field(discriminator="extraction_kind")] per the Pydantic v2 discriminated-union API.

2.1 The discriminated-union root

scripts/cocoindex_pipeline/extraction.py

from __future__ import annotations

from datetime import datetime
from typing import Annotated, Literal, Union
from uuid import UUID

from pydantic import BaseModel, Field, ConfigDict


class _ExtractionBase(BaseModel):
    """Shared fields populated by the outer-tier flow wrapper, not by the LLM.

    Per PRODUCT.md invariant 5 — every variant carries op_id +
    content_items_id + extracted_at. These fields are stamped by the
    cocoindex flow wrapper *after* the LLM response is validated; they
    are NOT part of the LLM's output_type contract per cocoindex's
    ExtractByLlm semantics (the model would otherwise hallucinate
    UUIDs).

    Per S-1 verifier suggestion: extractor_kind (the link to
    q_a_extractions.extractor_kind enum) is stamped at outer-tier
    write time, not on the Pydantic shape — it is a per-target
    column populated by the target-binding adapter in flow.py.
    """

    model_config = ConfigDict(
        # Strict mode — Pydantic refuses to coerce mismatched types.
        # A model returning {"extraction_kind": "q_a_form", ...} where
        # a sub-field is the wrong type fails loud per PRODUCT.md
        # invariant 13.
        strict=True,
        # Forbid unexpected fields — surfacing prompt drift early.
        extra="forbid",
    )

    op_id: UUID = Field(
        description="Cocoindex per-flow op_id — hybrid op_id pattern "
        "per 02-data-flow.md §5.1 N7."
    )
    content_items_id: UUID = Field(
        description="FK to content_items row whose content_text was "
        "the extraction input — source-attribution marker."
    )
    extracted_at: datetime = Field(
        description="UTC timestamp set at LLM-call time by the "
        "outer-tier flow wrapper."
    )


class FormMetadata(BaseModel):
    """Block carried inside the q_a_form variant per PRODUCT.md inv 2.

    form_type values per docs/ontology/26-form-type.md lines 65-79
    (full 11-value canonical CV — 8 procurement form_type values per
    Q-OQR1-02 ratification at line 52 + 3 non-procurement form_type
    values per the same ontology doc).

    PER VERIFIER B-2: The earlier 8-value draft omitted checklist /
    questionnaire / sales_proposal_template. The full 11-value
    Literal is the canonical contract; non-procurement form
    routing for the v1 pipeline is gated on the form_types CV
    instance table landing in T2 per GAP-Q-EX2-001.
    """

    model_config = ConfigDict(strict=True, extra="forbid")

    form_type: Literal[
        "bid", "rfp", "pqq", "itt",
        "tender", "framework", "dps", "gcloud",
        "checklist", "questionnaire", "sales_proposal_template",
    ]
    form_format: Literal["docx", "xlsx", "pdf", "html", "md"]
    form_title: str | None = None
    issuing_organisation: str | None = None
    deadline: datetime | None = None
    evaluation_methodology: str | None = None


class QAPair(BaseModel):
    """One Q&A pair extracted from a form.

    PER VERIFIER B-1: The drafter spec named the obligation field
    `question_kind` and assigned a fabricated 3-value Literal
    (`mandatory` / `optional` / `info_only`). Two problems:

      (a) `question_kind` collides with the canonical column name on
          the `question_matches` table per 05-qa-flow.md §7.2 — that
          column holds form-type discriminator values
          (`bid` / `rfp` / `pqq` / etc.), not obligation values.
      (b) The third value `info_only` was fabricated — the ratified
          shape at phase-b-prerequisite-2a-cocoindex-examples.md
          line 212 is the 2-value `Literal["mandatory", "optional"]`;
          a hyphenated `info-only` appears in unratified pseudo-code
          at line 341 but has no upstream ratification.

    Fix: rename the field to `expected_response_kind` (no collision
    with question_matches) and use the canonical 2-value Literal.
    If a third state ("info-only" or similar) becomes load-bearing
    downstream, it lands via spec update with citation rather than
    prompt drift.
    """

    model_config = ConfigDict(strict=True, extra="forbid")

    question_text: str = Field(min_length=1)
    answer_text: str | None = None
    expected_response_kind: Literal["mandatory", "optional"]
    evaluation_criteria: str | None = None
    evidence_requirements: list[str] = Field(default_factory=list)
    scope_tags: list[str] = Field(default_factory=list)


class QAFormExtraction(_ExtractionBase):
    """The q_a_form discriminated-union variant per PRODUCT.md inv 2.

    Maps downstream to q_a_extractions (per QAPair) + form_templates
    (per FormMetadata).
    """

    extraction_kind: Literal["q_a_form"] = "q_a_form"
    form_metadata: FormMetadata
    qa_pairs: list[QAPair] = Field(default_factory=list)


class EntityMentionExtraction(_ExtractionBase):
    """The entity_mention variant per PRODUCT.md inv 3.

    entity_type values mirror VALID_ENTITY_TYPES in
    lib/validation/schemas.ts:1495-1508 — keep these enums in lockstep.
    A parity test (per §5.4 below) asserts the two lists match.
    """

    extraction_kind: Literal["entity_mention"] = "entity_mention"
    entity_type: Literal[
        "organisation", "certification", "regulation", "framework",
        "capability", "person", "technology", "project",
        "sector", "product", "standard", "methodology",
    ]
    entity_name: str = Field(min_length=1)
    canonical_name: str | None = None
    source_span_start: int = Field(ge=0)
    source_span_end: int = Field(ge=0)
    mention_confidence: float = Field(ge=0.0, le=1.0)


class ClassificationExtraction(_ExtractionBase):
    """The classification variant per PRODUCT.md inv 4.

    content_type values mirror VALID_CONTENT_TYPES in
    lib/validation/schemas.ts:43-52 (re-export of CONTENT_TYPE_VALUES
    from lib/ontology/content-type-registry.ts). The parity test
    extends to this enum too.
    """

    extraction_kind: Literal["classification"] = "classification"
    content_type: str  # Constrained at runtime — see §2.2 below.
    primary_domain: str
    classification_confidence: float = Field(ge=0.0, le=1.0)
    secondary_classifications: list[str] = Field(default_factory=list)
    rationale: str | None = None


ExtractionOutput = Annotated[
    Union[
        QAFormExtraction,
        EntityMentionExtraction,
        ClassificationExtraction,
    ],
    Field(discriminator="extraction_kind"),
]

2.2 Content-type enum cross-validation

The content_type field on ClassificationExtraction is a plain str in the Pydantic shape rather than a Literal[...] for one reason: the canonical enumeration lives in lib/ontology/content-type-registry.ts and is mirrored into lib/validation/schemas.ts via re-export. Mirroring the same list into Python would create a third source of drift. Instead:

• At module load time, extraction.py reads the canonical list from scripts/tests/fixtures/taxonomy_snapshot.json (per CLAUDE.md Gotcha “Taxonomy dual-source — Python pipeline reads taxonomy from scripts/tests/fixtures/taxonomy_snapshot.json”).
• A runtime validator (added to ClassificationExtraction via @field_validator("content_type")) asserts the LLM-returned value is in the snapshot list. Validation failure raises ValueError, which Pydantic surfaces as the classification variant’s validation error per PRODUCT.md invariant 13.
• A guard test in scripts/tests/cocoindex_pipeline/test_taxonomy_parity.py (the Python sibling of the existing TypeScript markdown-parity.test.ts at __tests__/lib/ontology/markdown-parity.test.ts) asserts the snapshot matches the markdown ontology register.

The entity_type and form_type Literal lists are short, stable, and ratified by spec — these are inline Literal[...] per the shapes above. The parity test in §5.4 below covers them.

3. Extraction integration pattern (Path A — @coco.fn-wrapped anthropic SDK)

API-DEVIATION-S256 — Path A canonical pattern. The cocoindex.functions.ExtractByLlm / cocoindex.LlmSpec / cocoindex.LlmApiType symbols cited in this section’s pre-S256 history are ABSENT in the installed cocoindex==1.0.3 pin (verified empirically per OQ-3 import-and-call check at S256 W1; full provenance in docs/research/cocoindex-1.0.3-extractbyllm-spec-reality-investigation.md). Cocoindex 1.0.0 restructured away from the enum-gated LLM-API surface; the canonical extraction pattern in 1.x is a KH-authored @coco.fn-wrapped function calling the anthropic SDK directly, with Pydantic model_validate_json() for the typed output. Verifier B-3 finding REVERSED: the pre-S256 anti-pattern call-out (“ExtractByLlm inside @coco.fn is API-incorrect”) no longer applies — in 1.x, @coco.fn IS the canonical wrapper (confirmed by live examples-repo patient_intake_extraction_baml + patient_intake_extraction_dspy + paper_metadata which all wrap external SDK calls inside @coco.fn). No litellm shim required — @coco.fn imposes zero LLM-provider gating; direct import anthropic; anthropic.Anthropic().messages.create(...) inside the function body is sanctioned.

The §3.1 + §3.2 + §6 anti-pattern table content below records the pre-S256 ExtractByLlm-centric intent. The actual landed pattern in scripts/cocoindex_pipeline/flow.py (T8 implementation, WP4 of S256) calls anthropic SDK directly inside @coco.fn(memo=True)-decorated extractors. Pydantic validation, Inv-13 failure routing, Inv-21 memoisation, and Inv-22 invariant coverage carry over unchanged — only the LLM-call mechanism switches from ExtractByLlm (which doesn’t exist) to direct SDK call (which works). Treat the code skeletons below as intent record, not as the literal implementation guide.

3.1 The canonical Path A pattern (S256)

Per 02-data-flow.md §3.1 (the 6-stage topology — row 3 “LLM extraction” lists the LLM-call primitive at flow scope), the binary-conversion stage materialises a content_text: str column on the row (typically named markdown or content_text depending on the binary path — Docling-produced GFM markdown for PDF / DOCX / XLSX; pullmd-produced markdown for HTML / URL sources). The flow-scope extraction step consumes that column via a @coco.fn-decorated function that:

1. Loads the prompt constant from scripts/cocoindex_pipeline/prompts.py.
2. Calls anthropic.Anthropic().messages.create(model="claude-opus-4-6", messages=[{"role": "user", "content": f"{PROMPT}\n\n{content_text}"}], ...).
3. Extracts the response text + passes through pydantic.TypeAdapter[ClassificationExtraction].validate_json(...) (or the relevant variant) — failures route via classify_pydantic_error() per §4.1.
4. Returns the validated typed value to cocoindex flow scope.

Memoisation per Inv-21 is @coco.fn(memo=True) + the deterministic memo key (content_text, prompt_version) — cocoindex’s content-hash determinism guarantees memo-hit on unchanged content + unchanged prompt.

# scripts/cocoindex_pipeline/extraction.py (or flow.py — implementer's choice, document in journal)
# WP4 (S256) lands this; WP3 lands only the Pydantic shapes + prompts that this function consumes.

import anthropic
import cocoindex as coco
from pydantic import TypeAdapter, ValidationError

from scripts.cocoindex_pipeline.extraction import (
    ClassificationExtraction,
    EntityMentionExtraction,
    QAFormExtraction,
    classify_pydantic_error,
)
from scripts.cocoindex_pipeline.prompts import (
    CLASSIFICATION_PROMPT,
    ENTITY_MENTION_PROMPT,
    Q_A_FORM_PROMPT,
)

ANTHROPIC_MODEL = "claude-opus-4-6"  # production tier per lib/anthropic.ts:29 + scripts/kb_pipeline/config.py:29
_classification_adapter = TypeAdapter(ClassificationExtraction)


@coco.fn(memo=True)
async def extract_classification(content_text: str) -> ClassificationExtraction:
    client = anthropic.AsyncAnthropic()  # picks up ANTHROPIC_API_KEY from env
    response = await client.messages.create(
        model=ANTHROPIC_MODEL,
        max_tokens=4096,
        messages=[{
            "role": "user",
            "content": f"{CLASSIFICATION_PROMPT}\n\n{content_text}",
        }],
    )
    response_text = response.content[0].text
    return _classification_adapter.validate_json(response_text)

The flow-scope wiring is then a plain .transform():

# scripts/cocoindex_pipeline/flow.py — WP4 lands this
flow["classification"] = flow["content_text"].transform(extract_classification)
flow["q_a_form"] = flow["content_text"].transform(extract_qa_form)
flow["entity_mentions"] = flow["content_text"].transform(extract_entity_mentions)

3.1-LEGACY The pre-S256 ExtractByLlm flow-scope pattern (HISTORICAL — DOES NOT LAND)

HISTORICAL — DOES NOT LAND. The code below cites symbols that DO NOT EXIST in cocoindex==1.0.3. Preserved as intent record so the §3.2 / §4 / §6 cross-references remain readable. Implementation follows §3.1 Path A above.

Per 02-data-flow.md §3.1 (the 6-stage topology — row 3 “LLM extraction” lists ExtractByLlm as the cocoindex primitive at flow scope), the binary-conversion stage materialises a content_text: str column on the row (typically named markdown or content_text depending on the binary path — Docling-produced GFM markdown for PDF / DOCX / XLSX; pullmd-produced markdown for HTML / URL sources). The flow-scope ExtractByLlm step consumes that column and produces a typed extracted column. Layered fn-shape (S9 §7.2) applies: any @coco.fn(memo=True) helper that runs after ExtractByLlm takes the typed column (and, where needed, the source content_text: str) as its input parameter — never FileLike.

# scripts/cocoindex_pipeline/flow.py (T8 — shape illustrated for spec completeness)

import cocoindex as coco
from cocoindex import LlmSpec, LlmApiType
from cocoindex.functions import ExtractByLlm

from scripts.cocoindex_pipeline.extraction import (
    ClassificationExtraction,
    EntityMentionExtraction,
    QAFormExtraction,
)
from scripts.cocoindex_pipeline.prompts import (
    CLASSIFICATION_PROMPT,
    ENTITY_MENTION_PROMPT,
    Q_A_FORM_PROMPT,
)


# Production Anthropic model — sourced from `lib/anthropic.ts:29` +
# `scripts/kb_pipeline/config.py:29`. NOT the Claude-Code-session
# model `claude-opus-4-7` (which the drafter spec conflated — see
# verifier N-2). Production uses claude-opus-4-6 as the drafting
# tier; the cocoindex-examples line-365 reference to claude-opus-4-7
# is forward-looking pseudo-code, not a ratified production target.
ANTHROPIC_MODEL = "claude-opus-4-6"


def attach_extraction_stages(flow: coco.Flow) -> None:
    """Wire the three flow-scope ExtractByLlm steps + post-processing.

    Per phase-b-prerequisite-2a-cocoindex-examples.md lines 196-203 +
    362-368, ExtractByLlm is invoked on the flow column (here named
    `content_text`, produced by the binary-conversion stage). The
    pattern mirrors:

        doc["extracted"] = doc["markdown"].transform(
            cocoindex.functions.ExtractByLlm(
                llm_spec=cocoindex.LlmSpec(
                    api_type=cocoindex.LlmApiType.ANTHROPIC,
                    model=ANTHROPIC_MODEL,
                ),
                output_type=Patient,
                instruction="Please extract ..."))

    No @coco.fn wrapper around the ExtractByLlm call itself — the
    cocoindex primitive IS the flow operation.
    """

    # ── Classification — runs on every content_items row ──────────────
    flow["classification"] = flow["content_text"].transform(
        ExtractByLlm(
            llm_spec=LlmSpec(
                api_type=LlmApiType.ANTHROPIC,
                model=ANTHROPIC_MODEL,
            ),
            output_type=ClassificationExtraction,
            instruction=CLASSIFICATION_PROMPT,
        )
    )

    # ── Q&A form — fires only when content_type indicates a form ─────
    # Routing predicate per PRODUCT.md inv 7 is applied at flow
    # composition time (T8 owns the per-content-type fanout — typically
    # a filter operation on flow["classification"].content_type before
    # the q_a_form ExtractByLlm step runs).
    flow["q_a_form"] = flow["content_text"].transform(
        ExtractByLlm(
            llm_spec=LlmSpec(
                api_type=LlmApiType.ANTHROPIC,
                model=ANTHROPIC_MODEL,
            ),
            output_type=QAFormExtraction,
            instruction=Q_A_FORM_PROMPT,
        )
    )

    # ── Entity mentions — runs on every content_items row ────────────
    # output_type=list[EntityMentionExtraction] — cocoindex supports
    # list-typed extraction (per cocoindex-examples line 326:
    # "Supports nested schemas with list[T] / T | None / nested
    # dataclass references").
    flow["entity_mentions"] = flow["content_text"].transform(
        ExtractByLlm(
            llm_spec=LlmSpec(
                api_type=LlmApiType.ANTHROPIC,
                model=ANTHROPIC_MODEL,
            ),
            output_type=list[EntityMentionExtraction],
            instruction=ENTITY_MENTION_PROMPT,
        )
    )

    # Stamp the _ExtractionBase fields (op_id, content_items_id,
    # extracted_at). This is a flow-scope post-processing step, not
    # an @coco.fn — see §3.2 below.
    flow["classification_stamped"] = flow["classification"].transform(
        stamp_extraction_base,
        op_id=flow["op_id"],
        content_items_id=flow["content_items_id"],
    )
    # Repeat for q_a_form + entity_mentions.

3.2 Optional inner-tier post-processing helpers (@coco.fn(memo=True))

When post-processing of the typed ExtractByLlm output is needed (e.g. span-offset normalisation, deterministic confidence-rescaling, evidence-requirement deduplication), it is implemented as a pure-Python @coco.fn(memo=True) function consuming the typed extracted column + the source content_text: str. Per S9 §7.1 + §7.2 the layered fn-shape requirement is load-bearing — the inner tier consumes content (typed or string), NEVER FileLike.

# scripts/cocoindex_pipeline/extraction.py (cont.)

import cocoindex as coco
from scripts.cocoindex_pipeline.extraction import EntityMentionExtraction


@coco.fn(memo=True)
async def normalise_entity_span(
    extraction: EntityMentionExtraction,
    content_text: str,
) -> EntityMentionExtraction:
    """Inner-tier post-processing fn — consumes the typed extracted
    column AND the source content_text. Adjusts span offsets to align
    with whitespace boundaries.

    Per S9 §7.2 layered fn-shape: inputs are content (typed +
    string), NOT FileLike. Metadata-only edits to the source file
    (mtime, owner_change) hit memo cleanly because the memo key is
    (extraction_payload, content_text), not the file handle.
    """
    span = content_text[
        extraction.source_span_start : extraction.source_span_end
    ]
    # ... normalise whitespace boundaries ...
    return extraction.model_copy(
        update={
            "source_span_start": ...,
            "source_span_end": ...,
        }
    )

The _ExtractionBase field stamping (op_id, content_items_id, extracted_at) is also a post-processing operation. It is NOT a memoised @coco.fn (those three values change per flow run, so memoisation would either stale-cache the values or invalidate every run — defeating the purpose). It is instead a plain Python helper invoked at flow scope:

# scripts/cocoindex_pipeline/extraction.py (cont.)

from datetime import datetime, timezone


def stamp_extraction_base(
    extraction: ClassificationExtraction
    | QAFormExtraction
    | EntityMentionExtraction,
    *,
    op_id: UUID,
    content_items_id: UUID,
) -> (
    ClassificationExtraction
    | QAFormExtraction
    | EntityMentionExtraction
):
    """Plain Python helper — NOT @coco.fn. Stamps the _ExtractionBase
    fields with op_id (from cocoindex flow context), content_items_id
    (from the row's primary key in source-binding tier), and
    extracted_at (now-UTC).

    Pydantic v2 model_copy preserves immutability semantics.
    """
    return extraction.model_copy(
        update={
            "op_id": op_id,
            "content_items_id": content_items_id,
            "extracted_at": datetime.now(timezone.utc),
        }
    )

The prompt-template constants (CLASSIFICATION_PROMPT, Q_A_FORM_PROMPT, ENTITY_MENTION_PROMPT) live in scripts/cocoindex_pipeline/prompts.py — prompt design is owned by T8 (cocoindex flow scaffolding) per PLAN.md §4.8. This contract owns only the typed output shape and the integration pattern.

4. Validation behaviour on Anthropic API response mismatch

4.1 Pydantic strict-mode handling

Per the ConfigDict(strict=True, extra="forbid") configuration on _ExtractionBase (§2.1 above), the following Anthropic response shapes fail validation per PRODUCT.md invariant 13:

Response shape	Pydantic error	error_class for pipeline_runs
Missing extraction_kind field	discriminator validation error	invalid_discriminator
extraction_kind: "unknown_variant"	discriminator lookup failure	invalid_discriminator
entity_type: "foo" (not in 12-value list)	Literal validation error	invalid_enum
form_type: "rfx" (not in 11-value list)	Literal validation error	invalid_enum
expected_response_kind: "info_only" (not in 2-value list)	Literal validation error	invalid_enum
mention_confidence: "0.5" (str instead of float)	strict-mode type error	type_coercion
qa_pairs: null (instead of [])	default_factory not applied to null	type_coercion
Extra field llm_internal_thought: "..."	extra="forbid" violation	unexpected_field
Missing required field question_text	required-field validation	missing_required

Per S-2 verifier suggestion, the _classify_pydantic_error helper (referenced at §4.2 below) is implemented as an explicit dict mapping rather than implicit code-table:

# scripts/cocoindex_pipeline/extraction.py (cont.)

from pydantic import ValidationError

_PYDANTIC_ERROR_TO_ERROR_CLASS: dict[str, str] = {
    "missing": "missing_required",
    "literal_error": "invalid_enum",
    "union_tag_invalid": "invalid_discriminator",
    "union_tag_not_found": "invalid_discriminator",
    "extra_forbidden": "unexpected_field",
    "string_type": "type_coercion",
    "int_type": "type_coercion",
    "float_type": "type_coercion",
    "bool_type": "type_coercion",
    "uuid_parsing": "type_coercion",
    "datetime_parsing": "type_coercion",
}


def classify_pydantic_error(exc: ValidationError) -> str:
    """Map the first error in a ValidationError to an error_class string.

    Returns 'type_coercion' as the default — type-coercion errors are
    the broadest category and the safe fallback for unmapped error
    types. The mapping is exhaustive against Pydantic v2's documented
    error types as of pydantic>=2.0; unmapped errors are logged at
    WARN level so the mapping can be extended.
    """
    if not exc.errors():
        return "type_coercion"
    first_error_type = exc.errors()[0].get("type", "")
    return _PYDANTIC_ERROR_TO_ERROR_CLASS.get(
        first_error_type, "type_coercion"
    )

4.2 The validation-failure write path

# scripts/cocoindex_pipeline/flow.py (T8 illustration)
from pydantic import ValidationError

from scripts.cocoindex_pipeline.extraction import classify_pydantic_error


try:
    classification_typed = stamp_extraction_base(
        flow["classification"].value,  # ExtractByLlm-produced typed value
        op_id=op_id,
        content_items_id=row_id,
    )
    await target_table.upsert(classification_typed)
except ValidationError as exc:
    await record_extraction_failure(
        op_id=op_id,
        content_items_id=row_id,
        extraction_kind="classification",
        error_class=classify_pydantic_error(exc),
        raw_response=exc.input,  # Redacted of PII by record helper
    )
    # NOT raised — pipeline continues processing subsequent rows
    # per PRODUCT.md invariant 13.

record_extraction_failure (T8 subtask) writes to pipeline_runs via recordPipelineRun() per CLAUDE.md Gotcha “Cron pipeline_runs inserts”. The function lives in lib/pipeline/record-run.ts (TypeScript); the Python pipeline either calls it via an internal RPC or duplicates the safe-insert pattern. The choice is owned by T8 + the sibling cocoindex-ledger-api/TECH.md spec (T1.3 — drafted in parallel), not this contract.

4.3 Retry vs validation-failure split

Per PRODUCT.md invariants 17 + 18:

• Transient API failures (network, rate limit, 5xx) — retry per cocoindex native back-off. NOT this contract’s concern.
• Validation failure (Pydantic mismatch) — terminal for this run. Record + continue. NO retry; the malformed response is deterministic given the prompt + content + model version.

The split is enforced by catching ValidationError specifically at the post-ExtractByLlm Pydantic-coercion site — other exceptions (e.g. anthropic.APIError) propagate to cocoindex’s retry machinery.

5. Source-attribution + provenance

Per PRODUCT.md invariants 15, 16, 20:

• op_id stamping happens in the flow-scope post-processing step (stamp_extraction_base, §3.2 above). The same op_id is propagated to downstream q_a_extractions.op_id, entity_mentions.op_id, and content_items.op_id columns (the latter added in T2 per PLAN.md §4.2).
• content_items_id is the only foreign-key link an extraction carries — source_documents_id is reachable only via content_items.source_document_id.
• Failed extractions surface via pipeline_runs only — never as user-facing error envelopes (AI-invisible-infrastructure invariant).

6. Anti-patterns to enforce in code review

These are the negative invariants from PRODUCT.md §“Anti-patterns” — call out in code review:

Anti-pattern	Detection	Fix
output_type=dict[str, Any] on any ExtractByLlm call	grep output_type=dict	Use one of the three typed variants.
ExtractByLlm(...).transform(content_text) invoked inside @coco.fn(memo=True) async def	AST scan of @coco.fn-decorated functions for ExtractByLlm references	Move ExtractByLlm to flow scope per §3.1; inner-tier @coco.fns do post-processing only.
extract_*(file: FileLike, ...) on an inner-tier fn	grep FileLike inside extraction.py	Move file-handling to outer tier; inner tier takes content_text: str and/or the typed extracted value.
Combined “extract everything” prompt with bundled Union[QAForm, Entity, Classification] output_type	Code review — flagged on any extractor that targets multiple variants	Split into three separate flow-scope ExtractByLlm calls per §3.1.
LlmApiType.OLLAMA / LlmApiType.OPENAI in v1 code	grep LlmApiType\.	Anthropic only in v1; OLLAMA + OPENAI deferred to v1.1.
raw_llm_response column anywhere in q_a_extractions / entity_mentions / content_items schema	Migration review	Raw responses live in pipeline_runs for failures only.
Field named question_kind on any extraction Pydantic shape	grep question_kind in scripts/cocoindex_pipeline/	Collides with question_matches.question_kind per 05-qa-flow.md §7.2 — use expected_response_kind or a different non-colliding name.

7. CLAUDE.md gotchas applied

• cocoindex 1.0.3 requires dangerouslyDisableSandbox: true — applies to dev runs of the extraction module. Tests run with the sandbox bypass per the existing scripts/ontology-sync/parse-flow.py precedent (lines 21-22).
• localfs.walk_dir defaults recursive=False — orthogonal to this contract (source-binding stage, not LLM extraction), but flagged because T8 will combine them.
• Python background output: Use PYTHONUNBUFFERED=1 — applies to any backgrounded extraction script; not specific to this contract.
• content_items.content_text_hash is GENERATED ALWAYS — applies whenever an extraction writes back to content_items. The classification variant writes back content_items.content_type / content_items.primary_domain / content_items.classification_confidence. Any UPSERT payload to content_items MUST omit the content_text_hash field — Postgres auto-computes it via md5(normalised content), and explicit values are rejected with cannot insert a non-DEFAULT value into column. T8 implementers: the upsert payload builder must filter content_text_hash out before the postgres.mount_table_target write.
• classifyContent userId must be a UUID — this is the legacy TS path; the cocoindex extraction contract does not carry a userId concept (extraction is system-driven; user attribution is recovered via content_items.created_by).

Testing and validation

This section maps each numbered Behavior invariant from PRODUCT.md to a concrete verification path. Tests live in scripts/tests/cocoindex_pipeline/test_extraction_contract.py (created in T8 with this spec as the brief).

5.1 Schema-shape invariants (unit tests)

PRODUCT.md inv	Verification	Test type
1 — discriminated union	ExtractionOutput parses each of the three variants via JSON fixtures; parses fail for extraction_kind: null and extraction_kind: "foo"	unit (pytest + Pydantic round-trip)
2 — q_a_form shape	A fixture with FormMetadata + 3 QAPair round-trips through QAFormExtraction.model_validate_json(); a fixture with expected_response_kind: "info_only" fails (canonical 2-value Literal)	unit
3 — entity_mention shape	12 fixtures (one per VALID_ENTITY_TYPES value) all parse; a 13th fixture with entity_type: "junk" fails	unit
4 — classification shape	A fixture with each VALID_CONTENT_TYPES value parses; a fixture with a junk content_type fails the @field_validator per §2.2	unit
5 — _ExtractionBase fields	A fixture missing op_id fails; a fixture with op_id: "not-a-uuid" fails Pydantic v2 UUID parsing. Note (per verifier N-5): Pydantic v2’s strict-mode UUID parses any RFC-4122 form; the test asserts “UUID parse error”, NOT v1/v4 distinction (the CLAUDE.md “Zod UUID validation” gotcha is TypeScript-side and does NOT apply to Python.)	unit

5.2 Validation-behaviour invariants (unit tests against Pydantic)

PRODUCT.md inv	Verification	Test type
11 — pre-UPSERT validation	A malformed response is not surfaced to any downstream UPSERT call site (assert no target_table.upsert call in mocked wrapper)	unit (mock-based)
12 — layered fn-shape	A static-analysis test (mypy + custom AST walker) asserts NO inner-tier @coco.fn function has a FileLike parameter AND asserts NO @coco.fn-decorated function contains an ExtractByLlm(...) call (the verifier B-3 anti-pattern)	unit (AST scan)
13 — validation-failure record	Mock a malformed Anthropic response; assert record_extraction_failure is called once with correct error_class	unit (mock-based)
14 — successful retry overrides	Sequence (failure, then success) on same content_items_id; assert the success row is the final state in target table	integration
18 — no retry on validation mismatch	Mock Pydantic ValidationError; assert ZERO retries (i.e. only one Anthropic call); assert ONE pipeline_runs failure row	unit
19 — invalid_discriminator severity	A response with extraction_kind absent fails with error_class='invalid_discriminator' (the load-bearing prompt-drift signal); a response with extraction_kind: "foo" also maps to invalid_discriminator via classify_pydantic_error	unit

5.3 ExtractByLlm + cocoindex integration (integration tests)

These tests require cocoindex>=1.0.3 installed and a live Anthropic API key (ANTHROPIC_API_KEY env var). They run under dangerouslyDisableSandbox: true per CLAUDE.md Gotcha.

PRODUCT.md inv	Verification	Test type
6 — classification fires every row	Seed 5 content_text fixtures of varying content_type; assert 5 ClassificationExtraction rows produced in one flow run	integration (Anthropic live)
7 — q_a_form routing	Seed 3 fixtures (one form, one Q&A markdown, one non-form); assert exactly 2 QAFormExtraction rows	integration
8 — entity_mention fires regardless of content_type	Seed 3 fixtures (form, policy, methodology); each yields ≥0 EntityMentionExtraction rows (zero is valid for content with no entities)	integration
10 — mixed variants per run	Seed one form-fixture content_text; assert one ClassificationExtraction + one QAFormExtraction + ≥0 EntityMentionExtraction in the same op_id	integration
21 — prompt-template version in code-hash	Bump the CLASSIFICATION_PROMPT constant; re-run the flow against unchanged fixtures; assert ALL classification rows re-extract (memo invalidated per 0.9-spike-S9-cocoindex-idempotency.md §7.4)	integration

Per verifier N-6: Invariant 9 (confidence threshold) is tested by the T8 entity-resolution wiring, NOT by this contract’s acceptance plan. The threshold is applied at the entity-resolution stage (downstream of this contract per PRODUCT.md inv 8 + inv 9); the Pydantic shape only requires mention_confidence: float in [0.0, 1.0] — the persistence rule is owned by T8.

5.4 Enum-parity guards (regression tests)

Per the dual-source taxonomy pattern (CLAUDE.md Gotcha “Taxonomy dual-source — Python pipeline reads taxonomy from scripts/tests/fixtures/taxonomy_snapshot.json”), a parity guard enforces enum lockstep:

Enum	TypeScript / ontology source	Python mirror	Guard test
VALID_ENTITY_TYPES (12 values)	lib/validation/schemas.ts:1495-1508	Inline Literal[...] in EntityMentionExtraction	test_entity_type_parity.py asserts the two lists match
VALID_CONTENT_TYPES (variable)	lib/validation/schemas.ts:43-52 (re-export from lib/ontology/content-type-registry.ts)	Runtime validator reading taxonomy_snapshot.json	test_content_type_parity.py asserts snapshot matches registry
form_type (11 values)	docs/ontology/26-form-type.md lines 65-79 (CV markdown register)	Inline Literal[...] in FormMetadata	test_form_type_parity.py asserts inline list matches the 11-value CV register (8 procurement + 3 non-procurement)
expected_response_kind (2 values)	This spec §2.1 + phase-b-prerequisite-2a-cocoindex-examples.md line 212 (ratified)	Inline Literal[...] in QAPair	test_expected_response_kind_parity.py asserts the 2-value list matches the canonical source

The guard tests live in scripts/tests/cocoindex_pipeline/ and run on every Python test pass (python3 -m pytest scripts/tests/ per CLAUDE.md Commands).

Per S-3 verifier suggestion: parity guards are the load-bearing defence against silent drift. Runbook entry (add to T8 acceptance criteria): “Run python3 -m pytest scripts/tests/cocoindex_pipeline/test_*_parity.py after any change to lib/validation/schemas.ts, docs/ontology/26-form-type.md, or this spec’s §2.1 Literal lists.”

5.5 Anti-pattern enforcement (lint + AST)

PRODUCT.md anti-pattern	Enforcement
22 — no raw_llm_response columns	Migration-review checklist + bun run knip scan of schema TS types
23 — no “extract-everything” variant	AST scan in CI: any flow-scope ExtractByLlm whose output_type argument is ExtractionOutput (the root union) instead of a single variant fails the test
24 — no LlmApiType.OLLAMA / OPENAI in v1	grep -n "LlmApiType\.\(OLLAMA|OPENAI\)" scripts/cocoindex_pipeline/ returns zero hits
§6 row “ExtractByLlm inside @coco.fn”	AST scan: any @coco.fn-decorated function whose body references ExtractByLlm fails the test
§6 row “field named question_kind”	grep -rn "question_kind" scripts/cocoindex_pipeline/ returns zero hits (the canonical name is reserved for question_matches)

5.6 Manual verification (one-off — pre-launch)

• Anthropic prompt-cache passthrough verification per PRODUCT.md gap-flag [GAP-Q-EX2-002]. Owned by T13 of the canonical-pipeline PLAN (§4.13 subtask 1) — not blocking this spec, but the verification result feeds back into the prompt design (no-op if cache passthrough works; prompt-trim pass if it does not).
• Nested-schema depth verification per PRODUCT.md gap-flag [GAP-Q-EX2-003]. Run the QAFormExtraction (FormMetadata + list[QAPair] nesting) against the live Anthropic API with a representative bid fixture; confirm the nesting survives end-to-end. If a depth-flattening pre-stage is needed, lands in T8 flow-scope pre-processing before the ExtractByLlm call.

Risks and mitigations

Risk	Mitigation
Anthropic returns malformed JSON despite output_type declaration	Pydantic strict-mode + extra="forbid" catch the drift at validation; pipeline continues per inv 13. Operator alert via pipeline_runs filter on error_class='invalid_discriminator'.
Prompt-cache passthrough does not work (per [GAP-Q-EX2-002])	Cost projection per phase-b-prerequisite-2-cocoindex-deep-dive.md §5 — non-blocking; affects extraction cost, not correctness. T13 spike resolves.
Nested-schema depth fails on Anthropic (per [GAP-Q-EX2-003])	Fallback: flow-scope pre-stage flattens nesting before ExtractByLlm; flow-scope post-stage reconstructs nested shape post-validation. Adds complexity but preserves contract.
Enum drift between TypeScript and Python sources	Parity guards in §5.4 fail CI; the Python tests run in the quality-test matrix per docs/runbooks/ci.md.
A new content_type added to TS without snapshot update	The taxonomy-sync.yml workflow already exists per CLAUDE.md CI section. Add Python-side check that the Python extraction tests run against the latest snapshot.
cocoindex 1.0.3 sandbox failure in CI	CLAUDE.md Gotcha — dangerouslyDisableSandbox: true required. CI job runs the integration suite with the bypass; unit tests do not need cocoindex installed (pure Pydantic).
T8 implementer copies the broken ExtractByLlm inside @coco.fn pattern (verifier B-3 risk)	Anti-pattern §6 row + AST scan §5.5 row 4 catch the violation; this spec’s §3.1 shows the canonical flow-scope pattern as the only sanctioned shape.

Follow-ups

• docs/specs/id-36-cocoindex-ledger-api/TECH.md (sibling T1.3 spec — drafted in parallel) — owns the pipeline_runs write-back contract that this contract’s failure-path depends on. Cross-reference for the record_extraction_failure helper signature. Per S-4 verifier suggestion, the §1 Context block above surfaces this dependency prominently.
• docs/specs/cocoindex-extraction-prompts/ (POTENTIAL future spec — DEFERRED-v1) — if prompt design becomes a load-bearing concern (e.g. prompt-cache passthrough resolves favourably and prompt-template versioning becomes a substantive surface), spec the prompt-template lifecycle separately.
• Pre-launch verification spike (T13) per PLAN.md §4.13 — Anthropic prompt-cache passthrough + nested-schema depth. Both gap-flagged in PRODUCT.md; both NON-BLOCKING for this contract’s spec land.
• scripts/ontology-sync/parse-flow.py (existing stub) — when the live ontology-sync flow lands (per docs/specs/wp6-ontology-harness/TECH.md §8), the same Pydantic-with-flow-scope-ExtractByLlm pattern from this contract applies (CV files don’t use ExtractByLlm — they parse markdown frontmatter — so the parallel is the layered fn-shape, not the LLM step). Cross-reference for consistency on the layered fn-shape only.
• Production model name — currently bound to claude-opus-4-6 per lib/anthropic.ts:29 + scripts/kb_pipeline/config.py:29. When the production model is upgraded (whether to claude-opus-4-7 1M-context or a successor), ANTHROPIC_MODEL in flow.py updates with citation; this spec’s §3.1 binding is non-load-bearing on the specific model string.