analitiq-validator

The Analitiq artifact validator. It validates Analitiq JSON documents against the contract models — the same Pydantic models the published JSON Schemas are generated from — plus the cross-file and cross-document checks a single-document model cannot express, so authoring, the connector-builder plugin, and downstream consumers all enforce one contract with no drift. It is structured so per-kind validators slot in without touching each other.

Today it covers:

• authored single documents — connector, endpoint, and type-map files, plus connection, stream, and pipeline documents;
• pipeline bundles — the cross-document referential integrity of an assembled run (pipeline + streams + connections + connectors + endpoints).

Single-document validity (structure and every cross-field rule) is delegated to TypeAdapter(...).validate_python from analitiq-contract-models (analitiq.contracts). It runs offline — no schema fetch, no network. On top of the models it adds only what a single-document model cannot express — the connection / stream / pipeline kinds are pure model validation (the model IS the whole contract), so analitiq-validate --document x.json validates every authored kind and emits the same uniform {passed, findings[]}.

Connector-package cross-file checks:

• cross-file coverage — a connector ships a sibling type-map.json carrying the sections its kind calls for, and no other type-map name, and an API connector's read map covers every (native_type, arrow_type) its endpoint files declare;
• filename ↔ id — an endpoint file is named {endpoint_id}.json;
• advisory warnings the contract tolerates — duplicate type-map rules, read patterns spelling a lowercase literal, regex natives spelling a container that renders a scalar, write-map vocabulary gaps.

Pipeline-bundle referential integrity (validate_pipeline_bundle(bundle)) — a bundle is a mapping of already-parsed {pipeline, streams, connections, connectors, endpoints} documents, checked for internal consistency independent of any storage or on-disk layout:

• every pipeline.streams[] ref resolves to exactly one bundled stream document;
• every referenced connection is present in the bundle;
• every stream endpoint_ref.connection_id is one of the pipeline's connections;
• every connection's connector_id is present among the bundled connectors;
• every scope='connection' endpoint_ref resolves to a bundled endpoint document;
• the bundle names a pipeline (has a pipeline_id).

Referential integrity is separate from runnability. require_runnable=True (the default) additionally gates the pipeline on status='active' with at least one runnable stream — the check an executor needs. An authoring tool validating a draft bundle passes require_runnable=False to get the referential checks without the active-status gate.

It does not assume the documents were already contract-validated, so a missing reference field (a connection naming no connector, a stream slot with no endpoint_ref) is reported as an unresolved reference — while per-document shape (field types, lengths, enums) remains each document model's own job. Stream and connection refs match on their base form, so a {id}_v{n} versioned ref resolves the document that declares the bare {id} (connector identities match whole).

Path-free document sets

check_coverage and the pipeline-builder plugin's own bundle assembly both resolve a document's siblings by reading real files off a filesystem path. A consumer with no local filesystem to put those files on — a hosted validator wrapping this package as a remote tool, registry CI, any caller handed document content directly — cannot use either route: it has the documents' content in hand, never a directory they can be read from.

analitiq.validator.document_set fixes the contract such a consumer calls instead. validate_pipeline_package raises NotImplementedError. packages/validator/tests/test_document_set.py is the fixture corpus that fixes what an implementation must satisfy, a case xfail while the function it exercises raises. Read that module and that test file for the contract itself — it is not restated here.

The contract is deliberately narrow. Each entry point takes the request model that names the unit being submitted — ValidateSingleDocumentRequest or ValidatePackageRequest, both from analitiq.contracts.validation_requests, which own the text-only document value type and, for a package, the document-key grammar, and are where a malformed argument is refused. A request never carries a directory to read from, and its keys are never resolved against a filesystem — the offline guarantee stated above for single-document validation, extended to a set of documents instead of one.

Rule cases

Some rules enforced by a cross-document check ship example document sets with the package, each one a directory holding a bundle.json or a connector package rooted at connector.json. A rule is not required to carry any, so the corpus grades only the rules it holds cases for. rule_cases() loads them and case_mismatch(case) grades one against the installed validator, returning None when the validator agrees:

from analitiq.validator.rule_cases import case_mismatch, rule_cases

mismatches = [m for m in map(case_mismatch, rule_cases()) if m]

Install

pip install analitiq-validator

This pulls analitiq-contract-models (and pydantic) transitively.

Use

analitiq-validate --document definition/connector.json
analitiq-validate --document connection.json   # connection / stream / pipeline work too

The kind is detected from the document's shape — no --kind flag. Point --document at definition/connector.json to also trigger cross-file coverage: it discovers the sibling type-map.json and endpoints/*.json from the connector's directory. A connection / stream / pipeline document is validated purely against its contract model. Validation is always model-driven and offline — --document is the only flag.

Output is a JSON report ({"passed": bool, "findings": [...]}) on stdout; the process exits non-zero exactly when passed is false — a fail finding at severity: "error", or a notApplicable finding naming a rule that is (or, naming none, might as well be) error-tier, since a check that could not run gets no benefit of the doubt.

A pipeline bundle is assembled from many documents, so it is validated as a library call rather than from a single file:

from analitiq.validator import validate_pipeline_bundle

findings = validate_pipeline_bundle(
    {
        "pipeline": pipeline_doc,
        "streams": stream_docs,
        "connections": connection_docs,
        "connectors": connector_ids,          # the connector identities present
        "endpoints": connection_endpoint_docs,  # scope='connection', connection_id, endpoint_id
    },
    require_runnable=False,  # authoring a DRAFT bundle: referential checks, no active-status gate
)
if any(f["severity"] == "error" for f in findings):
    raise SystemExit(findings)

Source of truth

The canonical source is analitiq-ai/claude-code-plugins, under packages/validator/ — with the contract models beside it at packages/contract-models/, and the public JSON Schemas rendered from those same models by scripts/render_schemas.py.

The package is authored directly in the public analitiq.validator namespace; it is staged verbatim at build time, not rendered from a private tree. Edit the source there, not the installed copy.