Metadata-Version: 2.5
Name: crosswatch
Version: 0.1.0
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

# crosswatch

[![CI](https://github.com/RakshRak/crosswatch/actions/workflows/ci.yml/badge.svg)](https://github.com/RakshRak/crosswatch/actions/workflows/ci.yml)
[![License: Apache-2.0](https://img.shields.io/badge/License-Apache%202.0-blue.svg)](LICENSE)
[![Python 3.10+](https://img.shields.io/badge/python-3.10%2B-blue.svg)](pyproject.toml)

**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](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 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:

```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/).

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

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

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

## License

Apache-2.0.
