Metadata-Version: 2.5
Name: crosswatch
Version: 0.1.1
Summary: Unified observability for DATA and AI products — wraps and composes best-in-class OSS (OpenTelemetry, Great Expectations, Evidently, DeepEval/Ragas) behind one plugin-driven API.
Project-URL: Homepage, https://github.com/RakshRak/crosswatch
Project-URL: Repository, https://github.com/RakshRak/crosswatch
Project-URL: Issues, https://github.com/RakshRak/crosswatch/issues
Project-URL: Changelog, https://github.com/RakshRak/crosswatch/blob/main/CHANGELOG.md
Author: crosswatch contributors
License-Expression: Apache-2.0
License-File: LICENSE
Keywords: data-quality,evals,llm,mcp,observability,opentelemetry
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: System :: Monitoring
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: duckdb>=0.10
Requires-Dist: ibis-framework[duckdb]>=9.0
Requires-Dist: opentelemetry-api>=1.20
Requires-Dist: opentelemetry-sdk>=1.20
Requires-Dist: pydantic-settings>=2.1
Requires-Dist: pydantic>=2.5
Requires-Dist: pyyaml>=6.0
Requires-Dist: typer>=0.12
Provides-Extra: airflow
Requires-Dist: apache-airflow>=2.8; extra == 'airflow'
Provides-Extra: anthropic
Requires-Dist: anthropic>=0.25; extra == 'anthropic'
Requires-Dist: openinference-instrumentation-anthropic>=0.1; extra == 'anthropic'
Provides-Extra: bigquery
Requires-Dist: ibis-framework[bigquery]>=9.0; extra == 'bigquery'
Provides-Extra: clickhouse
Requires-Dist: clickhouse-connect>=0.7; extra == 'clickhouse'
Requires-Dist: ibis-framework[clickhouse]>=9.0; extra == 'clickhouse'
Provides-Extra: dagster
Requires-Dist: dagster>=1.6; extra == 'dagster'
Provides-Extra: databricks
Requires-Dist: databricks-sql-connector>=3.0; extra == 'databricks'
Requires-Dist: sqlalchemy>=2.0; extra == 'databricks'
Provides-Extra: deepeval
Requires-Dist: deepeval>=1.0; extra == 'deepeval'
Provides-Extra: dev
Requires-Dist: httpx>=0.27; extra == 'dev'
Requires-Dist: mypy>=1.10; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest-cov>=5.0; extra == 'dev'
Requires-Dist: pytest>=8.0; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Provides-Extra: evidently
Requires-Dist: evidently>=0.4; extra == 'evidently'
Provides-Extra: gx
Requires-Dist: great-expectations>=0.18; extra == 'gx'
Provides-Extra: langchain
Requires-Dist: openinference-instrumentation-langchain>=0.1; extra == 'langchain'
Provides-Extra: langfuse
Requires-Dist: langfuse>=2.0; extra == 'langfuse'
Provides-Extra: lineage
Requires-Dist: openlineage-python>=1.9; extra == 'lineage'
Provides-Extra: llama-index
Requires-Dist: openinference-instrumentation-llama-index>=2.0; extra == 'llama-index'
Provides-Extra: mcp
Requires-Dist: mcp>=1.0; extra == 'mcp'
Provides-Extra: mysql
Requires-Dist: cryptography>=42.0; extra == 'mysql'
Requires-Dist: pymysql>=1.1; extra == 'mysql'
Provides-Extra: openai
Requires-Dist: openai>=1.10; extra == 'openai'
Requires-Dist: openinference-instrumentation-openai>=0.1; extra == 'openai'
Provides-Extra: otlp
Requires-Dist: opentelemetry-exporter-otlp>=1.20; extra == 'otlp'
Provides-Extra: phoenix
Requires-Dist: arize-phoenix-otel>=0.5; extra == 'phoenix'
Provides-Extra: postgres
Requires-Dist: ibis-framework[postgres]>=9.0; extra == 'postgres'
Requires-Dist: psycopg[binary]>=3.1; extra == 'postgres'
Requires-Dist: sqlalchemy>=2.0; extra == 'postgres'
Provides-Extra: prefect
Requires-Dist: prefect>=2.16; extra == 'prefect'
Provides-Extra: ragas
Requires-Dist: ragas>=0.2; extra == 'ragas'
Provides-Extra: server
Requires-Dist: fastapi>=0.110; extra == 'server'
Requires-Dist: jinja2>=3.1; extra == 'server'
Requires-Dist: uvicorn>=0.27; extra == 'server'
Provides-Extra: slack
Requires-Dist: httpx>=0.27; extra == 'slack'
Provides-Extra: snowflake
Requires-Dist: ibis-framework[snowflake]>=9.0; extra == 'snowflake'
Provides-Extra: soda
Requires-Dist: soda-core>=3.3; extra == 'soda'
Provides-Extra: spark
Requires-Dist: ibis-framework[pyspark]>=9.0; extra == 'spark'
Provides-Extra: sqlite
Description-Content-Type: text/markdown

<p align="center">
  <img src="https://raw.githubusercontent.com/RakshRak/crosswatch/main/logo.png" alt="crosswatch logo" width="140">
</p>

<h1 align="center">crosswatch</h1>

<p align="center"><strong>Unified observability for DATA and AI products.</strong></p>

<p align="center">
  <a href="https://github.com/RakshRak/crosswatch/actions/workflows/ci.yml"><img src="https://github.com/RakshRak/crosswatch/actions/workflows/ci.yml/badge.svg" alt="CI"></a>
  <a href="https://pypi.org/project/crosswatch/"><img src="https://img.shields.io/pypi/v/crosswatch.svg" alt="PyPI"></a>
  <a href="LICENSE"><img src="https://img.shields.io/badge/License-Apache%202.0-blue.svg" alt="License: Apache-2.0"></a>
  <a href="pyproject.toml"><img src="https://img.shields.io/badge/python-3.10%2B-blue.svg" alt="Python 3.10+"></a>
</p>

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.

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

## Table of contents

- [Features](#features)
- [What's unique: data → AI impact correlation](#whats-unique-data--ai-impact-correlation)
- [Why](#why)
- [Install](#install)
- [Connectors](#connectors)
- [Checks](#checks)
- [Evals](#evals)
- [Alerts](#alerts)
- [Exporters](#exporters)
- [5-minute quickstart](#5-minute-quickstart)
- [Configuration — YAML, JSON, or Python](#configuration--yaml-json-or-python)
- [Examples](#examples)
- [Local demo services](#local-demo-services)
- [Running in the cloud](#running-in-the-cloud-spark-dataproc-airflow)
- [Architecture at a glance](#architecture-at-a-glance)
- [Known limitations / roadmap](#known-limitations--roadmap)
- [Development](#development)
- [License](#license)

## Features

- **Data**: freshness, volume, schema-change, null/duplicate, and drift checks
  as portable [Ibis](https://ibis-project.org) 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 plugin API + CI mode that fails the build on a score
  regression. Ships an LLM-as-judge evaluator (deterministic or real-LLM
  grading) and the MCP tool-use evals above (DeepEval/Ragas providers planned,
  not yet built — see [Evals](#evals)).
- **Store / alerts / UI**: default DuckDB/Parquet store, OTLP export to any
  backend (Phoenix/Langfuse/Jaeger/Datadog), Slack/webhook alerts, and a minimal
  FastAPI dashboard.

## 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:

```bash
crosswatch impact -c crosswatch.yml
```

```text
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']
```

```python
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

```bash
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](https://docs.astral.sh/uv/).

> `gx`, `soda`, `deepeval`, and `ragas` currently only install the underlying
> library — see [Known limitations / roadmap](#known-limitations--roadmap).

## 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`:

| Source | `type` | Extra to install | Status |
|---|---|---|---|
| DuckDB | `duckdb` | *(core)* | ✅ default; file or in-memory |
| SQLite | `sqlite` | `sqlite` *(no deps)* | ✅ file-based |
| Postgres | `postgres` | `postgres` | ✅ + query history |
| MySQL | `mysql` | `mysql` | ✅ + query history |
| ClickHouse | `clickhouse` | `clickhouse` | ✅ + query history |
| Snowflake | `snowflake` | `snowflake` | ⚙️ needs account + credentials |
| BigQuery | `bigquery` | `bigquery` | ⚙️ needs project + ADC |
| Databricks | `databricks` | `databricks` | ⚙️ needs workspace + token |

```bash
pip install 'crosswatch[mysql,clickhouse]'
```

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

## Checks

Every check is a Pydantic model discovered via the `crosswatch.checks`
entry-point group; set it in config with `type`. List what's available
(built-ins plus any third-party checks installed) at any time:

```python
from crosswatch import plugins

print(plugins.available(plugins.CHECKS))
# ['drift', 'freshness', 'null_duplicate', 'schema_change', 'volume']
```

```bash
python -c "from crosswatch import plugins; print(plugins.available(plugins.CHECKS))"
```

Every check also takes the common fields `table`, `severity`
(`error`/`warn`/`info`, default `error`), and an optional `name`:

| `type` | Asserts | Key fields |
|---|---|---|
| `freshness` | newest row in `column` isn't older than `max_age` | `column`, `max_age` (e.g. `"24h"`) |
| `volume` | row count falls within bounds | `min_rows`, `max_rows` |
| `schema_change` | live schema matches expectations | `expected_columns`, `expected_types`, `allow_new` |
| `null_duplicate` | null fraction + key uniqueness | `columns`, `max_null_fraction`, `key` |
| `drift` | PSI between a reference and current distribution | `column`, `reference_table`, `threshold` |

Third parties can publish their own check by registering a
`crosswatch.checks` entry point — no change to the core.

## Evals

An evaluator scores an `EvalCase` (an AI input/output/context — e.g. a RAG
answer or an agent's tool call) and returns an `EvalResult`. Every evaluator
returns the same result shape, so a rule-based score, an LLM judge, and a
replayed MCP session all flow into the same store, CI mode, and alerts.
Evaluators are discovered via the `crosswatch.evaluators` entry-point group:

```python
from crosswatch import plugins

print(plugins.available(plugins.EVALUATORS))
# ['llm_judge']
```

### LLM-as-judge

`LLMJudge` scores a case `0.0`–`1.0` against a rubric. Inject a deterministic
`score_fn` for CI/tests (no LLM call needed), or pass a real client to grade
with a model:

```python
from crosswatch.evals import LLMJudge, run_dataset
from crosswatch.models import EvalCase

cases = [
    EvalCase(id="1", input="What is the capital of France?",
             context=["Paris is the capital of France."], output="Paris"),
]

# CI-friendly: deterministic, no LLM call
judge = LLMJudge(metric="answer_quality", threshold=0.8,
                  score_fn=lambda c: 1.0 if "Paris" in c.output else 0.0)
results = run_dataset(judge, cases)

# Or grade with a real LLM:
import anthropic
judge = LLMJudge(metric="answer_quality", threshold=0.8,
                  client=anthropic.Anthropic(), model="claude-opus-4-8")
results = run_dataset(judge, cases)
```

> **CLI note:** `crosswatch eval -c crosswatch.yml` currently only dispatches
> `mcp_tool_selection`/`mcp` evaluators configured in `crosswatch.yml`'s
> `evals:` block (see [Examples](#examples) below) — any other evaluator name,
> including `llm_judge`, is skipped with a warning. `LLMJudge` is Python-API-only
> for now.

### MCP tool-use evals

`tool_selection_accuracy`, `argument_correctness`, and `task_success_rate`
score recorded MCP sessions against ground truth, and `run_regression` replays
them through a live agent to catch drift between model/prompt versions — see
[MCP — trace tool calls + eval tool selection](#examples) for a worked example.

> `gx`/`soda`/`deepeval`/`ragas` evaluator providers aren't built yet — see
> [Known limitations / roadmap](#known-limitations--roadmap).

## Alerts

An alert channel delivers a failing check/eval result somewhere (Slack,
webhook). `dispatch` turns results into `AlertEvent`s and fans them out to
whichever channels you pass it — only failures by default, or every result
with `on="always"`. Channels are discovered via the `crosswatch.alerts`
entry-point group:

```python
from crosswatch import plugins

print(plugins.available(plugins.ALERTS))
# ['slack', 'webhook']
```

```python
from crosswatch.alerts.base import dispatch
from crosswatch.alerts.webhook import WebhookChannel

channel = WebhookChannel(url="https://example.com/hooks/crosswatch")
sent = dispatch(results, [channel])              # failures only
sent = dispatch(results, [channel], on="always")  # every result
```

| `channel` | Delivers to | Key fields |
|---|---|---|
| `webhook` | any URL, as a `POST` of JSON | `url`, `timeout` |
| `slack` | a Slack incoming webhook | `url` (formats as an emoji + title/body Slack message) |

> **Config note:** `crosswatch.yml`'s `alerts:` block parses into `AlertConfig`
> objects, but neither `crosswatch check` nor `crosswatch eval` call `dispatch`
> yet — configuring `alerts:` today is a no-op. Call `dispatch` yourself in the
> Python API until the CLI wires it in; see
> [Known limitations / roadmap](#known-limitations--roadmap).

Third parties can publish their own channel by registering a
`crosswatch.alerts` entry point — no change to the core.

## Exporters

Span exporters send OpenTelemetry traces to a tracing backend — set via
`telemetry.exporters` in config (see [Configuration](#configuration--yaml-json-or-python))
or `crosswatch.ai.configure(exporters=[...])` directly. Unlike alerts, this one
*is* fully wired: both the CLI and the Python API call into it. Discover
what's available:

```python
from crosswatch import plugins

print(plugins.available(plugins.EXPORTERS))
# ['langfuse', 'otlp', 'phoenix']
```

| `type` | Sends to | Config via env var |
|---|---|---|
| `otlp` | any OTLP/HTTP collector | `OTEL_EXPORTER_OTLP_TRACES_ENDPOINT` |
| `phoenix` | an [Arize Phoenix](https://docs.arize.com/phoenix) collector | `PHOENIX_COLLECTOR_ENDPOINT` (default `http://localhost:6006/v1/traces`) |
| `langfuse` | [Langfuse](https://langfuse.com), via its OTel endpoint | `LANGFUSE_HOST` (default `https://cloud.langfuse.com`) |

Third parties can publish their own exporter by registering a
`crosswatch.exporters` entry point — no change to the core.

## 5-minute quickstart

**1. Describe what to watch** in `crosswatch.yml`:

```yaml
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):

```bash
crosswatch check -c crosswatch.yml
```

```text
  ✓ 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:**

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

### Or use the Python API

```python
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`)

```yaml
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 }
```

```bash
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:

```json
{
  "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"] }
  ]
}
```

```bash
crosswatch check -c crosswatch.json
```

### Python

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

```python
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`

```bash
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

```python
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

```python
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

```bash
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:

```yaml
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):

```python
# 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]'`):

```yaml
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

```text
connectors/   one adapter per source (Ibis): get_schema/profile/run_check/get_query_history
data/         native checks (Ibis exprs) + dbt ingestion (GE/Soda/Evidently 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, MCP tool-use evals, CI mode
store/         DuckDB/Parquet default; ClickHouse/Postgres optional; OTLP exporter
alerts/        Slack / webhook channels, dispatch on failure or always
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.

## Known limitations / roadmap

crosswatch is alpha software. One honest list of what's actually implemented
vs. scaffolded-but-not-wired, instead of caveats scattered across the doc:

| Area | Status |
|---|---|
| Native checks (`freshness`/`volume`/`schema_change`/`null_duplicate`/`drift`) | ✅ implemented, tested |
| MCP tool-use evals (`tool_selection_accuracy`, `argument_correctness`, `task_success_rate`, `run_regression`) | ✅ implemented, tested |
| `LLMJudge` evaluator | ✅ implemented, tested — **Python API only**; not wired into `crosswatch eval`'s config path yet |
| Alert channels (`webhook`/`slack`) | ✅ implemented, tested — **Python API only**; `alerts:` in `crosswatch.yml` is parsed into config but never dispatched by the CLI yet |
| Span exporters (`otlp`/`phoenix`/`langfuse`) | ✅ implemented, wired into both the CLI and the Python API |
| Great Expectations / Soda check providers | ❌ not built — `gx`/`soda` extras install the library only, no crosswatch wrapper exists |
| DeepEval / Ragas eval providers | ❌ not built — `deepeval`/`ragas` extras install the library only |
| Evidently-based drift reports (`crosswatch.data.profiling`) | ❌ not built — `evidently` extra installs the library only; the module referenced in `drift.py`'s docstring doesn't exist yet |
| Snowflake / BigQuery / Databricks / Spark connectors | ⚙️ wired to the same Ibis backends as the verified ones, but not exercised against a live cluster/account — see [Running in the cloud](#running-in-the-cloud-spark-dataproc-airflow) |
| Email alert channel / threshold+anomaly alert rules | ❌ not built — only `webhook`/`slack` channels and a simple failure-or-always filter exist today |

Hit a gap not listed here? Please open an issue.

## Development

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

## License

Apache-2.0.
