crosswatch

Unified observability for DATA and AI products. One plugin-driven API, one config format, one local store, and one CLI across data quality and LLM/agent observability — by wrapping and composing best-in-class OSS rather than reinventing it.

• Data: freshness, volume, schema-change, null/duplicate, and drift checks as portable Ibis expressions that run unchanged on DuckDB, SQLite, Postgres, MySQL, ClickHouse, Snowflake, BigQuery, Databricks — plus dbt artifact ingestion (Great Expectations / Soda providers planned). Failures carry the reason, a suggested fix, and a sample of the exact offending rows/keys/bins.
• AI: OpenTelemetry-native tracing for OpenAI/Anthropic (and LangChain/LlamaIndex via OpenInference), with token / cost / latency on every span using GenAI semantic conventions.
• MCP: trace every tool call and evaluate tool-selection accuracy, argument correctness, and task success — turn recorded sessions into regression suites.
• Evals: evaluator API + CI mode that fails the build on a score regression.
• Store / alerts / UI: default DuckDB/Parquet store, OTLP export to any backend (Phoenix/Langfuse/Jaeger/Datadog), Slack/webhook alerts, and a minimal FastAPI dashboard.

🚧 Alpha. See docs/the architecture notes for the full design.

What's unique: data → AI impact correlation

Data-observability tools (Great Expectations, Soda, Monte Carlo) stop at the warehouse. LLM-observability tools (Langfuse, Phoenix, LangSmith) start at the prompt. Nothing joins them — so when a RAG eval regresses, you can't tell from one tool whether a data problem upstream caused it.

crosswatch records both data-quality results and AI evals in one store, so it correlates them. Tag an eval with the data assets it depends on, and crosswatch links a failing check on that asset to the AI output it degraded:

crosswatch impact -c crosswatch.yml

Data → AI impact (1 link):
  [warehouse.customers] null_duplicate FAILED ⇒ IMPACTED
      rag:answer_quality=0.42 (12 min later)
      data cause: email null_fraction=0.100 > 0.0; 1 duplicate rows over key ['id']

from crosswatch.models import EvalResult
from crosswatch.impact import correlate

# Declare the data an AI output depends on:
EvalResult(evaluator="rag", metric="answer_quality", score=0.42, threshold=0.8,
           assets=["warehouse.customers"])

links = correlate(checks, evals)   # failing check on an asset ⟶ the evals that use it

This answers a question no single-domain tool can: which data problem is degrading my AI, and which AI outputs are at risk from this data problem? It shows up as a dedicated card on the dashboard, and crosswatch impact exits non-zero when a data failure impacted a failing eval — a CI gate across the data/AI boundary.

Why

Teams stitch together Great Expectations for data quality, OpenTelemetry for tracing, DeepEval/Ragas for evals, and bespoke MCP logging — each with its own config, storage, and alerting. crosswatch is the composition layer: it doesn't re-implement expectation engines or tracers, it gives them one surface.

Install

pip install crosswatch                              # thin core (DuckDB + native checks)
pip install 'crosswatch[postgres,anthropic,server]' # add only what you need

Extras: sqlite postgres mysql clickhouse snowflake bigquery databricks openai anthropic langchain llama-index mcp gx soda evidently deepeval ragas otlp phoenix langfuse server slack. Managed with uv.

Connectors

One adapter per source, all exposing the same interface (get_schema / profile / run_check / get_query_history) — so a check definition runs unchanged on any of them. Referenced in config by type:

pip install 'crosswatch[mysql,clickhouse]'

Third parties can publish their own connector by registering a crosswatch.connectors entry point — no change to the core.

5-minute quickstart

1. Describe what to watch in crosswatch.yml:

store: { type: duckdb, path: .crosswatch/crosswatch.duckdb }
telemetry: { exporters: [] }            # add 'phoenix' / 'langfuse' / 'otlp'
connectors:
  warehouse: { type: duckdb, path: data.duckdb }   # or postgres: { dsn: ${POSTGRES_DSN} }
checks:
  - { type: volume,         connector: warehouse, table: orders, min_rows: 1 }
  - { type: freshness,      connector: warehouse, table: orders, column: created_at, max_age: 24h }
  - { type: null_duplicate, connector: warehouse, table: orders, columns: [email], max_null_fraction: 0.0, key: [id] }

2. Run the checks (writes results to the store; non-zero exit on failure — drop it straight into CI):

crosswatch check -c crosswatch.yml

  ✓ volume [orders] pass: row_count=1240
  ✗ freshness [orders] fail: newest created_at = 2026-10-18T09:00:00+00:00
  ✓ null_duplicate [orders] pass: no nulls/duplicates beyond threshold

2/3 checks passed; 1 error-severity failure(s)

3. See it in the dashboard:

crosswatch serve -c crosswatch.yml      # http://127.0.0.1:8000

Or use the Python API

import crosswatch
from crosswatch.connectors.duckdb import DuckDBConnector
from crosswatch.data.checks.freshness import FreshnessCheck

conn = DuckDBConnector(path="data.duckdb")
result = conn.run_check(FreshnessCheck(table="orders", column="created_at", max_age="24h"))
print(result.status, result.message)

cfg = crosswatch.load("crosswatch.yml")   # same config, programmatic access

Configuration — YAML, JSON, or Python

The same config (store, telemetry exporters, connectors, checks, evals, alerts) can be expressed three ways. ${ENV_VAR} references are expanded in all of them.

YAML (crosswatch.yml)

store: { type: duckdb, path: .crosswatch/crosswatch.duckdb }
telemetry: { service_name: shop, exporters: [otlp] }   # or [], [phoenix], [langfuse]
connectors:
  warehouse: { type: postgres, dsn: ${POSTGRES_DSN} }
  events:    { type: clickhouse, host: localhost, port: 8123, user: crosswatch, password: ${CH_PW}, database: crosswatch }
checks:
  - { type: volume,         connector: warehouse, table: orders, min_rows: 1000 }
  - { type: freshness,      connector: warehouse, table: orders, column: created_at, max_age: 24h }
  - { type: null_duplicate, connector: warehouse, table: customers, columns: [email], max_null_fraction: 0.0, key: [id] }
  - { type: drift,          connector: events,    table: events_today, column: latency_ms, reference_table: events_baseline, threshold: 0.2, severity: warn }

crosswatch check -c crosswatch.yml       # exits non-zero on any error-severity failure

JSON (crosswatch.json)

JSON is valid input to the same loader (YAML is a JSON superset), so the CLI and crosswatch.load accept a .json file directly:

{
  "store": { "type": "duckdb", "path": ".crosswatch/crosswatch.duckdb" },
  "connectors": {
    "warehouse": { "type": "mysql", "host": "127.0.0.1", "port": 3306,
                   "user": "root", "password": "${MYSQL_PW}", "database": "shop" }
  },
  "checks": [
    { "type": "volume", "connector": "warehouse", "table": "orders", "min_rows": 1000 },
    { "type": "null_duplicate", "connector": "warehouse", "table": "customers",
      "columns": ["email"], "max_null_fraction": 0.0, "key": ["id"] }
  ]
}

crosswatch check -c crosswatch.json

Python

Load either file format, build from a dict, or construct the objects directly:

import json
import crosswatch
from crosswatch.config import Config
from crosswatch.data import run_checks

# (a) load a YAML or JSON file
cfg = crosswatch.load("crosswatch.yml")            # == Config.from_yaml("crosswatch.yml" | "crosswatch.json")

# (b) build from a Python dict (e.g. parsed JSON)
cfg = Config.from_dict(json.load(open("crosswatch.json")))

# run everything the config describes
store = cfg.build_store()
connectors = cfg.build_connectors()
checks = cfg.build_checks()
results = run_checks(checks, connectors, check_targets=[c.connector for c in cfg.checks])
store.write_check_results(results)
failed = [r for r in results if not r.ok]

# (c) or skip config entirely — construct a connector + checks by hand
from crosswatch.connectors.duckdb import DuckDBConnector
from crosswatch.data.checks.volume import VolumeCheck

conn = DuckDBConnector(path="data.duckdb")
r = conn.run_check(VolumeCheck(table="orders", min_rows=1000))
print(r.status, r.message, r.samples)   # samples pinpoint the offending rows

Examples

dbt — ingest run_results.json

dbt build                                  # produces target/run_results.json
python examples/dbt_quality.py target/run_results.json

Every dbt model/test outcome becomes a crosswatch CheckResult in the same store, dashboard, and alert path as native checks (see examples/dbt_quality.py).

RAG — trace an LLM pipeline with token/cost

from crosswatch.ai import configure, instrument_anthropic, observe

configure(service_name="rag", exporters=["phoenix"])   # OTLP to local Phoenix

@observe(kind="chain")
def answer(q: str) -> str:
    import anthropic
    client = instrument_anthropic(anthropic.Anthropic())   # auto-trace every call
    resp = client.messages.create(model="claude-opus-4-8", max_tokens=512,
                                  messages=[{"role": "user", "content": q}])
    return resp.content[0].text

Each call emits a GenAI-convention span with gen_ai.usage.input_tokens, output_tokens, and crosswatch.cost_usd. Full example: examples/rag_tracing.py.

MCP — trace tool calls + eval tool selection

from crosswatch.ai.mcp import TracedMCPClient, tool_selection_accuracy, MCPSessionRecord

client = TracedMCPClient(mcp_session)          # every call_tool is traced + recorded
await client.call_tool("get_weather", {"city": "Paris"})

# Score recorded sessions against ground-truth expected tools (CI regression gate):
score = tool_selection_accuracy(recorded_sessions, threshold=0.9)

Run crosswatch eval -c crosswatch.yml to gate CI on tool-selection accuracy. Full example: examples/mcp_eval.py.

Local demo services

docker compose up -d        # Postgres + MySQL + ClickHouse + Phoenix
export CROSSWATCH_TEST_PG_DSN=postgresql://crosswatch:crosswatch@localhost:5433/crosswatch
pytest                      # unit tests run on DuckDB/SQLite; -m integration hits the servers

Running in the cloud (Spark, Dataproc, Airflow, …)

crosswatch is a library + CLI, not a server — it runs wherever Python runs: an Airflow/Dagster/Prefect task, a Dataproc/EMR PySpark job, a Cloud Run or Kubernetes job, or CI. Two things change versus a laptop:

1. Use a shared, durable store (not the local DuckDB file). On ephemeral cloud workers, point the store at a managed Postgres so every job writes to one place the dashboard and crosswatch impact can read:

store: { type: postgres, path: ${CROSSWATCH_STORE_DSN} }   # e.g. Cloud SQL / RDS
telemetry: { exporters: [otlp] }                       # spans → your OTel collector

(DuckDB/Parquet = local default; Postgres = shared/cloud; a ClickHouse results store is on the roadmap. Traces always export over OTLP regardless of store.)

2. Checks run on the cluster via the connector. The checks are Ibis expressions, so they execute in the engine you point at — Spark included.

Spark / Dataproc / EMR / Databricks — attach to the SparkSession your job already has (nothing is pulled to the driver except the aggregates):

# inside a Dataproc / EMR PySpark job (or a Databricks notebook)
from crosswatch.connectors.spark import SparkConnector
from crosswatch.data.checks.null_duplicate import NullDuplicateCheck

conn = SparkConnector.from_session(spark)                      # your SparkSession
r = conn.run_check(NullDuplicateCheck(table="events", columns=["user_id"],
                                      max_null_fraction=0.0, key=["event_id"]))

or declaratively (pip install 'crosswatch[spark]'):

connectors:
  lake: { type: spark, master: "local[*]", app_name: crosswatch }   # omit master to attach

Cloud warehouses use the same interface — snowflake, bigquery, databricks connectors (install the matching extra, pass credentials/DSN). A typical pattern: run crosswatch check / crosswatch impact as a step in your orchestrator after the load job, and fail the task on non-zero exit.

Verified locally: DuckDB, SQLite, Postgres, MySQL, ClickHouse. Spark and the cloud warehouses are wired to the same Ibis backends but need a live cluster/account to exercise — treat them as integration-tested-by-you.

Architecture at a glance

connectors/   one adapter per source (Ibis): get_schema/profile/run_check/get_query_history
data/         native checks (Ibis exprs) + dbt + Evidently drift (GE/Soda providers planned)
ai/           @observe + OpenAI/Anthropic auto-instrumentation (OTel GenAI conventions)
ai/mcp/       trace tool calls + tool-selection / argument / task-success evals
evals/        evaluator plugin API, LLM-as-judge, golden datasets, CI mode
store/         DuckDB/Parquet default; ClickHouse/Postgres optional; OTLP exporter
alerts/        Slack / webhook / email, threshold + anomaly rules
cli + server/  crosswatch check | eval | serve (FastAPI dashboard)

Everything pluggable is discovered via importlib.metadata entry points (groups: crosswatch.connectors, crosswatch.checks, crosswatch.check_providers, crosswatch.evaluators, crosswatch.alerts, crosswatch.exporters), so third parties ship compatible plugins without touching the core.

Development

uv venv && uv pip install -e '.[dev,server,postgres]'
pytest
ruff check src tests

License

Apache-2.0.