ADR 0002 phase 3: contract_version in run history

README.md

@@ -51,7 +51,8 @@ Scheduling options are documented in [`docs/scheduling.md`](docs/scheduling.md).
 - [x] Webhook alert integration tests against mock server
 - [x] Contract registry catalog (`contracts/registry.yml`) — [ADR 0002](docs/adr/0002-schema-registry-and-contract-versioning.md)
 - [x] CLI resolves `--contract orders` via registry (phase 2)
-- [ ] Run history stores `contract_version` metadata (phase 3)
+- [x] Run history stores `contract_version` metadata (phase 3)
+- [ ] CI registry consistency guards (phase 4)
 
 ## Technology stack
 

docs/adr/0002-schema-registry-and-contract-versioning.md

@@ -111,7 +111,7 @@ When implemented:
 |-------|-------|-------------|
 | **1** | Registry + docs | `contracts/registry.yml`, ADR 0002, README update |
 | **2** | CLI resolution | `--contract orders` resolves via registry; `--version` override |
-| **3** | History metadata | Persist `contract_version` in run history |
+| **3** | History metadata | Persist `contract_version` in run history | ✓ |
 | **4** | CI guards | Registry consistency tests; CHANGELOG requirement |
 | **5** | Pipeline pin (optional) | Config reference from `production-data-pipeline` |
 

src/dqo/cli.py

@@ -89,7 +89,8 @@ def main(argv: list[str] | None = None) -> int:
         runs = store.recent_runs(args.contract, limit=args.limit)
         for run in runs:
             status = "passed" if run["passed"] else "failed"
-            print(f"{run['started_at']}  {run['run_id']}  {status}")
+            version = run.get("contract_version") or "unknown"
+            print(f"{run['started_at']}  {run['run_id']}  v{version}  {status}")
         return 0
 
     parser.error(f"unknown command: {args.command}")

src/dqo/history.py

@@ -59,6 +59,7 @@ def _ensure_schema(self) -> None:
                     CREATE TABLE IF NOT EXISTS check_runs (
                         run_id TEXT PRIMARY KEY,
                         contract_name TEXT NOT NULL,
+                        contract_version TEXT,
                         started_at TEXT NOT NULL,
                         finished_at TEXT NOT NULL,
                         passed INTEGER NOT NULL
@@ -87,6 +88,7 @@ def _ensure_schema(self) -> None:
                     CREATE TABLE IF NOT EXISTS check_runs (
                         run_id TEXT PRIMARY KEY,
                         contract_name TEXT NOT NULL,
+                        contract_version TEXT,
                         started_at TIMESTAMPTZ NOT NULL,
                         finished_at TIMESTAMPTZ NOT NULL,
                         passed BOOLEAN NOT NULL
@@ -109,17 +111,41 @@ def _ensure_schema(self) -> None:
                     """
                 )
 
+            self._migrate_schema(connection)
+
+    def _migrate_schema(self, connection) -> None:
+        if self._is_sqlite:
+            columns = {
+                row[1]
+                for row in connection.execute("PRAGMA table_info(check_runs)").fetchall()
+            }
+            if "contract_version" not in columns:
+                connection.execute(
+                    "ALTER TABLE check_runs ADD COLUMN contract_version TEXT"
+                )
+            return
+
+        connection.execute(
+            """
+            ALTER TABLE check_runs
+            ADD COLUMN IF NOT EXISTS contract_version TEXT
+            """
+        )
+
     def save_run(self, summary: RunSummary) -> None:
         with self._connection() as connection:
             self._execute(
                 connection,
                 """
-                INSERT INTO check_runs (run_id, contract_name, started_at, finished_at, passed)
-                VALUES (%s, %s, %s, %s, %s)
+                INSERT INTO check_runs (
+                    run_id, contract_name, contract_version, started_at, finished_at, passed
+                )
+                VALUES (%s, %s, %s, %s, %s, %s)
                 """,
                 (
                     summary.run_id,
                     summary.contract_name,
+                    summary.contract_version,
                     summary.started_at.isoformat(),
                     summary.finished_at.isoformat(),
                     summary.passed if not self._is_sqlite else int(summary.passed),
@@ -153,7 +179,7 @@ def recent_runs(self, contract_name: str, *, limit: int = 10) -> list[dict[str,
             cursor = self._query(
                 connection,
                 """
-                SELECT run_id, contract_name, started_at, finished_at, passed
+                SELECT run_id, contract_name, contract_version, started_at, finished_at, passed
                 FROM check_runs
                 WHERE contract_name = %s
                 ORDER BY started_at DESC

src/dqo/models.py

@@ -66,6 +66,7 @@ class CheckResult:
 @dataclass(frozen=True)
 class RunSummary:
     contract_name: str
+    contract_version: str
     run_id: str
     started_at: datetime
     finished_at: datetime

src/dqo/runner.py

@@ -62,6 +62,7 @@ def run_contract_file(
 
     return RunSummary(
         contract_name=loaded.name,
+        contract_version=loaded.version,
         run_id=str(uuid.uuid4()),
         started_at=started_at,
         finished_at=finished_at,

tests/test_alerts.py

@@ -17,6 +17,7 @@
 def _failed_summary() -> RunSummary:
     return RunSummary(
         contract_name="orders",
+        contract_version="1.0",
         run_id="run-alert-1",
         started_at=datetime(2026, 7, 14, 10, 0, tzinfo=timezone.utc),
         finished_at=datetime(2026, 7, 14, 10, 1, tzinfo=timezone.utc),
@@ -98,6 +99,7 @@ def test_alert_router_delivers_failed_checks_to_webhook() -> None:
 def test_alert_router_skips_info_severity_for_webhook() -> None:
     summary = RunSummary(
         contract_name="orders",
+        contract_version="1.0",
         run_id="run-alert-2",
         started_at=datetime(2026, 7, 14, 10, 0, tzinfo=timezone.utc),
         finished_at=datetime(2026, 7, 14, 10, 1, tzinfo=timezone.utc),

tests/test_cli.py

@@ -43,3 +43,29 @@ def test_cli_run_resolves_registry_contract_name(tmp_path: Path) -> None:
     )
 
     assert exit_code == 0
+
+
+def test_cli_history_shows_contract_version(tmp_path: Path, capsys) -> None:
+    db_url = f"sqlite:///{tmp_path / 'history.db'}"
+    main(
+        [
+            "run",
+            "--contract",
+            "orders",
+            "--data",
+            "data/samples/orders.csv",
+            "--references",
+            "data/samples",
+            "--reference-time",
+            "2026-07-14T12:00:00Z",
+            "--no-console-alerts",
+            "--history-db",
+            db_url,
+        ]
+    )
+
+    exit_code = main(["history", "--contract", "orders", "--history-db", db_url])
+    output = capsys.readouterr().out
+
+    assert exit_code == 0
+    assert "v1.0" in output

tests/test_history.py

@@ -7,6 +7,7 @@
 def _sample_summary(run_id: str) -> RunSummary:
     return RunSummary(
         contract_name="orders",
+        contract_version="1.0",
         run_id=run_id,
         started_at=datetime(2026, 7, 14, 10, 0, tzinfo=timezone.utc),
         finished_at=datetime(2026, 7, 14, 10, 1, tzinfo=timezone.utc),
@@ -30,6 +31,30 @@ def _sample_summary(run_id: str) -> RunSummary:
     )
 
 
+def test_history_store_migrates_legacy_schema(tmp_path) -> None:
+    db_path = tmp_path / "legacy.db"
+    connection = __import__("sqlite3").connect(db_path)
+    connection.execute(
+        """
+        CREATE TABLE check_runs (
+            run_id TEXT PRIMARY KEY,
+            contract_name TEXT NOT NULL,
+            started_at TEXT NOT NULL,
+            finished_at TEXT NOT NULL,
+            passed INTEGER NOT NULL
+        )
+        """
+    )
+    connection.commit()
+    connection.close()
+
+    store = HistoryStore(database_url=f"sqlite:///{db_path}")
+    store.save_run(_sample_summary("run-migrated"))
+
+    runs = store.recent_runs("orders")
+    assert runs[0]["contract_version"] == "1.0"
+
+
 def test_history_store_persists_runs(tmp_path) -> None:
     db_url = f"sqlite:///{tmp_path / 'history.db'}"
     store = HistoryStore(database_url=db_url)
@@ -38,6 +63,7 @@ def test_history_store_persists_runs(tmp_path) -> None:
     runs = store.recent_runs("orders")
     assert len(runs) == 1
     assert runs[0]["run_id"] == "run-1"
+    assert runs[0]["contract_version"] == "1.0"
     assert runs[0]["passed"] == 0
 
     failures = store.failure_trend("orders")