Post-migration validation (--validate)

CHANGELOG.md

@@ -8,7 +8,7 @@ The format is based on Keep a Changelog and this project follows Semantic Versio
 
 ### Added
 
-- N/A
+- Optional post-migration validation (`--validate` / `MIGRATION_VALIDATE`). After migrating, validates the destination against the source: object parity (counts + the specific items missing in the destination) for project-scoped resources, plus event count-parity for datasets, experiments, and logs. Validation mirrors migration policy (e.g. bundle-backed code functions, which are intentionally skipped, are not reported as missing), honors the same `created_after`/`created_before` window, and is reported per-resource in the console and in `migration_report.json`. Note: for high-volume event resources we compare counts only — enumerating *which* events are missing does not scale; counts are reported per dataset/experiment/project so a discrepancy can still be localized. (roles/groups/ai_secrets/ACLs are out of scope.)
 
 ### Changed
 

README.md

@@ -402,6 +402,20 @@ The tool uses **two-level checkpointing** for streaming resources (logs, experim
 
 On resume: skips 1-30 (done), resumes experiment 31 from saved `_pagination_key`, continues with 32-100.
 
+### Validation
+
+Pass `--validate` (or `MIGRATION_VALIDATE=true`) to validate the destination against the source as a final phase after migrating. It runs per project and checks:
+
+- **Object parity** for project-scoped resources (datasets, experiments, prompts, functions, project_scores, views, project_tags, span_iframes): compares source vs destination and reports the **specific items missing** in the destination.
+- **Event count-parity** for datasets, experiments, and logs: compares a cheap `count` of events on each side (per dataset/experiment, and per project for logs).
+
+Results are summarized per resource in the console (mismatches are listed, with a pointer to the full report) and recorded under each project's `validation` block in `migration_report.json`.
+
+Notes and limitations:
+- Validation **mirrors migration policy** — intentionally skipped items (e.g. bundle-backed code functions) are **not** reported as missing — and honors the same `--created-after`/`--created-before` window.
+- For high-volume event resources it compares **counts only**. Enumerating *which* events are missing requires diffing full id sets and does not scale; counts are reported per dataset/experiment/project so a discrepancy can still be localized to a specific object.
+- Org-scoped resources (roles, groups, ai_secrets) and ACLs are **out of scope** for validation.
+
 ## Parallelization
 
 The migration tool currently uses **two active levels of concurrency**. The env vars in [Parallelization Tuning](#parallelization-tuning) still matter, but the within-project resource-type DAG concurrency described below has not been implemented yet.

braintrust_migrate/cli.py

@@ -227,6 +227,18 @@ def migrate(
             envvar="MIGRATION_GROUP_AUTO_INVITE_USERS",
         ),
     ] = None,
+    validate: Annotated[
+        bool | None,
+        typer.Option(
+            "--validate/--no-validate",
+            help=(
+                "After migrating, validate the destination against the source "
+                "(object counts + which items are missing; event count-parity "
+                "for datasets/experiments/logs)."
+            ),
+            envvar="MIGRATION_VALIDATE",
+        ),
+    ] = None,
 ) -> None:
     """Migrate resources from source to destination Braintrust organization.
 
@@ -259,6 +271,7 @@ def migrate(
             acl_auto_invite_users,
             group_map_users,
             group_auto_invite_users,
+            validate,
         )
     )
 
@@ -280,6 +293,7 @@ async def _migrate_main(
     acl_auto_invite_users: bool | None,
     group_map_users: bool | None,
     group_auto_invite_users: bool | None,
+    validate: bool | None = None,
 ) -> None:
     """Async implementation of the migrate command."""
     setup_logging(log_level, log_format)
@@ -352,6 +366,8 @@ async def _migrate_main(
             config.migration.group_map_users = group_map_users
         if group_auto_invite_users is not None:
             config.migration.group_auto_invite_users = group_auto_invite_users
+        if validate is not None:
+            config.migration.validate_migration = validate
 
         logger.info(
             "Starting migration",
@@ -914,6 +930,52 @@ def _display_results(results: dict) -> None:
                 f"  ... and {len(summary['errors']) - MAX_ERRORS_TO_DISPLAY} more errors"
             )
 
+    # Validation summary (only when --validate was used).
+    validations = [
+        (pname, pdata["validation"])
+        for pname, pdata in results.get("projects", {}).items()
+        if pdata.get("validation")
+    ]
+    if validations:
+        total_checks = sum(len(v.get("resources", [])) for _, v in validations)
+        failures = [
+            (pname, r)
+            for pname, v in validations
+            for r in v.get("resources", [])
+            if not r.get("ok")
+        ]
+        console.print("\n[bold]Validation[/bold]")
+        if not failures:
+            console.print(
+                f"  [green]✅ all {total_checks} validated resource(s) match[/green]"
+            )
+        else:
+            console.print(
+                f"  [red]❌ {len(failures)} of {total_checks} validated "
+                f"resource(s) have mismatches[/red]"
+            )
+            for pname, r in failures:
+                console.print(f"  [red]• {pname}/{r['resource_type']}[/red]")
+                if r.get("error"):
+                    console.print(f"      error: {r['error']}")
+                for check in r.get("checks", []):
+                    if check.get("ok"):
+                        continue
+                    if check.get("kind") == "object" and check.get("missing"):
+                        miss = check["missing"]
+                        shown = ", ".join(miss[:MAX_ERRORS_TO_DISPLAY])
+                        more = (
+                            f" (+{len(miss) - MAX_ERRORS_TO_DISPLAY} more)"
+                            if len(miss) > MAX_ERRORS_TO_DISPLAY
+                            else ""
+                        )
+                        console.print(f"      missing {len(miss)}: {shown}{more}")
+                    elif check.get("kind") == "events":
+                        console.print(
+                            f"      {check['scope']}: "
+                            f"source={check['source_count']} dest={check['dest_count']}"
+                        )
+
     # Always point to the full per-item report so users can drill into exactly
     # what was migrated / skipped / failed (and why) — counts above, detail here.
     report_path = results.get("report_path")

braintrust_migrate/config.py

@@ -168,6 +168,15 @@ class MigrationConfig(BaseModel):
         description="Use a SQLite seen-id store to prevent older versions overwriting newer ones during pagination",
     )
 
+    validate_migration: bool = Field(
+        default=False,
+        description=(
+            "After migrating, validate the destination against the source: object "
+            "counts (and which specific items are missing) for project-scoped "
+            "resources, plus event count-parity for datasets/experiments/logs."
+        ),
+    )
+
     created_after: str | None = Field(
         default=None,
         description=(
@@ -403,6 +412,13 @@ def _get_bool(specific_key: str, unified_key: str, default: str) -> bool:
         logs_use_seen_db = _get_bool(
             "MIGRATION_LOGS_USE_SEEN_DB", "MIGRATION_EVENTS_USE_SEEN_DB", "true"
         )
+        validate = os.getenv("MIGRATION_VALIDATE", "false").lower() in {
+            "1",
+            "true",
+            "yes",
+            "y",
+            "on",
+        }
 
         # Optional time filter (applies to logs and experiments)
         created_after = os.getenv("MIGRATION_CREATED_AFTER")
@@ -507,6 +523,7 @@ def _get_bool(specific_key: str, unified_key: str, default: str) -> bool:
                 logs_insert_batch_size=logs_insert_batch_size,
                 logs_use_version_snapshot=logs_use_version_snapshot,
                 logs_use_seen_db=logs_use_seen_db,
+                validate_migration=validate,
                 created_after=created_after,
                 created_before=created_before,
                 experiment_events_fetch_limit=experiment_events_fetch_limit,

braintrust_migrate/orchestration.py

@@ -11,6 +11,8 @@
 
 from braintrust_migrate.client import BraintrustClient, create_client_pair
 from braintrust_migrate.config import Config
+from braintrust_migrate.resources.base import list_resources
+from braintrust_migrate.validation import validate_resource
 from braintrust_migrate.resources import (
     ACLMigrator,
     AISecretMigrator,
@@ -858,6 +860,29 @@ async def _migrate_project(
                         }
                     )
 
+        # Optional post-migration validation (source vs destination parity).
+        if self.config.migration.validate_migration:
+            try:
+                project_results["validation"] = await self._validate_project(
+                    source_client=source_client,
+                    dest_client=dest_client,
+                    source_project_id=source_project_id,
+                    dest_project_id=dest_project_id,
+                    migrated_resource_types=list(project_results["resources"].keys()),
+                    id_mapping=global_id_mappings,
+                )
+            except Exception as e:
+                self._logger.error(
+                    "Project validation phase failed",
+                    project=project_name,
+                    error=str(e),
+                )
+                project_results["validation"] = {
+                    "ok": False,
+                    "error": str(e),
+                    "resources": [],
+                }
+
         self._logger.info(
             f"Completed project migration: {project_name}",
             total_resources=project_results["total_resources"],
@@ -868,6 +893,71 @@ async def _migrate_project(
 
         return project_results
 
+    async def _validate_project(
+        self,
+        *,
+        source_client: BraintrustClient,
+        dest_client: BraintrustClient,
+        source_project_id: str,
+        dest_project_id: str,
+        migrated_resource_types: list[str],
+        id_mapping: dict[str, str],
+    ) -> dict[str, Any]:
+        """Validate a project's migrated resources against the source.
+
+        Object parity (counts + which items are missing) plus event count-parity
+        for datasets/experiments/logs. Out-of-scope resource types are skipped
+        (validate_resource returns None).
+        """
+        results: list[dict[str, Any]] = []
+        for resource_type in migrated_resource_types:
+            try:
+                validation = await validate_resource(
+                    resource_type,
+                    source_client=source_client,
+                    dest_client=dest_client,
+                    source_project_id=source_project_id,
+                    dest_project_id=dest_project_id,
+                    list_fn=list_resources,
+                    id_mapping=id_mapping,
+                    created_after=self.config.migration.created_after,
+                    created_before=self.config.migration.created_before,
+                )
+            except Exception as e:
+                self._logger.error(
+                    "Validation failed for resource",
+                    resource=resource_type,
+                    error=str(e),
+                )
+                results.append(
+                    {
+                        "resource_type": resource_type,
+                        "ok": False,
+                        "error": str(e),
+                        "checks": [],
+                    }
+                )
+                continue
+
+            if validation is None:
+                continue
+            results.append(validation.to_dict())
+            if not validation.ok:
+                self._logger.warning(
+                    "Validation mismatch",
+                    resource=resource_type,
+                    checks=[c.to_dict() for c in validation.checks if not c.ok],
+                )
+
+        ok = all(r.get("ok", False) for r in results)
+        self._logger.info(
+            "Project validation complete",
+            project_id=dest_project_id,
+            ok=ok,
+            validated=len(results),
+        )
+        return {"ok": ok, "resources": results}
+
     def _get_organization_resources_to_migrate(self) -> list[str]:
         """Get list of organization-scoped resource types to migrate based on configuration.
 
@@ -1073,6 +1163,11 @@ def _generate_migration_report(
                         }
                     )
 
+            # Include post-migration validation results when present.
+            validation = project_data.get("validation")
+            if validation is not None:
+                project_summary["validation"] = validation
+
             detailed_report["projects"][project_name] = project_summary
 
         # Write JSON report to file