Add project mapping and safer dry-run output

.env.example

@@ -18,3 +18,5 @@ MIGRATION_RETRY_DELAY=1.0
 MIGRATION_MAX_CONCURRENT=10
 MIGRATION_CHECKPOINT_INTERVAL=50
 MIGRATION_STATE_DIR=./checkpoints
+# MIGRATION_PROJECT_MAP={"Source Project":"Destination Project"}
+# MIGRATION_PROJECT_MAP_FILE=./project-map.json

README.md

@@ -122,7 +122,9 @@ All options can be set via environment variables or CLI flags. CLI flags take pr
 | Environment Variable | CLI Flag | Default | Description |
 |---------------------|----------|---------|-------------|
 | `MIGRATION_RESOURCES` | `--resources`, `-r` | `all` | Comma-separated list of resources to migrate. Options: `all`, `ai_secrets`, `roles`, `groups`, `datasets`, `project_tags`, `span_iframes`, `functions`, `prompts`, `project_scores`, `experiments`, `logs`, `views` |
-| `MIGRATION_PROJECTS` | `--projects`, `-p` | *(all projects)* | Comma-separated list of project names to migrate |
+| `MIGRATION_PROJECTS` | `--projects`, `-p` | *(all projects)* | Comma-separated list of source project names to migrate |
+| `MIGRATION_PROJECT_MAP` | `--project-map` | *(same-name fallback)* | JSON object mapping source project names to destination project names, e.g. `{"Source Project":"Destination Project"}` |
+| `MIGRATION_PROJECT_MAP_FILE` | `--project-map-file` | *(none)* | Path to a JSON file mapping source project names to destination project names. Mutually exclusive with `--project-map` / `MIGRATION_PROJECT_MAP` |
 | `MIGRATION_CREATED_AFTER` | `--created-after` | *(none)* | Only applies to resources that support created-time filtering. Currently this affects project logs event streaming and experiment listing. Migrates items with `created >=` this value (**inclusive**). Format: `YYYY-MM-DD` or ISO-8601 |
 | `MIGRATION_CREATED_BEFORE` | `--created-before` | *(none)* | Only applies to resources that support created-time filtering. Currently this affects project logs event streaming and experiment listing. Migrates items with `created <` this value (**exclusive**). Format: `YYYY-MM-DD` or ISO-8601 |
 
@@ -258,6 +260,14 @@ braintrust-migrate migrate --resources ai_secrets,datasets,prompts
 
 # Migrate specific projects only
 braintrust-migrate migrate --projects "Project A","Project B"
+
+# Migrate a source project into a differently named destination project
+braintrust-migrate migrate \
+  --resources logs \
+  --projects "<SOURCE_PROJECT_NAME>" \
+  --project-map '{"<SOURCE_PROJECT_NAME>":"<DEST_PROJECT_NAME>"}' \
+  --created-after "<START_TIME_ISO_8601>" \
+  --created-before "<END_TIME_ISO_8601>"
 ```
 
 **Resume Migration:**
@@ -293,6 +303,15 @@ braintrust-migrate migrate \
 **Dry Run (Validation Only):**
 ```bash
 braintrust-migrate migrate --dry-run
+
+# Preview project mapping, destination existence, selected resources, and logs window
+braintrust-migrate migrate \
+  --dry-run \
+  --resources logs \
+  --projects "<SOURCE_PROJECT_NAME>" \
+  --project-map '{"<SOURCE_PROJECT_NAME>":"<DEST_PROJECT_NAME>"}' \
+  --created-after "<START_TIME_ISO_8601>" \
+  --created-before "<END_TIME_ISO_8601>"
 ```
 
 **Time-based Filtering:**

braintrust_migrate/cli.py

@@ -25,6 +25,8 @@
     Config,
     canonicalize_created_after,
     canonicalize_created_before,
+    load_project_name_mapping_file,
+    parse_project_name_mapping_json,
 )
 from braintrust_migrate.orchestration import MigrationOrchestrator
 
@@ -43,6 +45,24 @@
 console = Console()
 
 
+def _load_project_name_mapping_override(
+    project_map: str | None,
+    project_map_file: Path | None,
+) -> dict[str, str] | None:
+    """Load CLI/env project-name mapping overrides, if provided."""
+    if project_map is not None and project_map_file is not None:
+        raise ValueError("Set only one of --project-map or --project-map-file")
+    if project_map is not None:
+        return parse_project_name_mapping_json(project_map, field_name="--project-map")
+    if project_map_file is not None:
+        return load_project_name_mapping_file(project_map_file)
+    return None
+
+
+def _resources_include(resources: list[str], resource_name: str) -> bool:
+    return "all" in resources or resource_name in resources
+
+
 def setup_logging(log_level: str = "INFO", log_format: str = "json") -> None:
     """Setup structured logging.
 
@@ -89,10 +109,29 @@ def migrate(
         typer.Option(
             "--projects",
             "-p",
-            help="Comma-separated list of project names to migrate (if not specified, all projects will be migrated)",
+            help="Comma-separated list of source project names to migrate (if not specified, all projects will be migrated)",
             envvar="MIGRATION_PROJECTS",
         ),
     ] = None,
+    project_map: Annotated[
+        str | None,
+        typer.Option(
+            "--project-map",
+            help=(
+                "JSON object mapping source project names to destination project names "
+                '(e.g. \'{"Source Project":"Destination Project"}\')'
+            ),
+            envvar="MIGRATION_PROJECT_MAP",
+        ),
+    ] = None,
+    project_map_file: Annotated[
+        Path | None,
+        typer.Option(
+            "--project-map-file",
+            help="Path to a JSON file mapping source project names to destination project names",
+            envvar="MIGRATION_PROJECT_MAP_FILE",
+        ),
+    ] = None,
     state_dir: Annotated[
         Path | None,
         typer.Option(
@@ -245,6 +284,8 @@ def migrate(
         _migrate_main(
             resources,
             projects,
+            project_map,
+            project_map_file,
             state_dir,
             resume_run_dir,
             log_level,
@@ -266,6 +307,8 @@ def migrate(
 async def _migrate_main(
     resources: str,
     projects: str | None,
+    project_map: str | None,
+    project_map_file: Path | None,
     state_dir: Path | None,
     resume_run_dir: Path | None,
     log_level: str,
@@ -302,6 +345,13 @@ async def _migrate_main(
         else:
             config = Config.from_env()
 
+        project_name_mapping_override = _load_project_name_mapping_override(
+            project_map,
+            project_map_file,
+        )
+        if project_name_mapping_override is not None:
+            config.project_name_mapping = project_name_mapping_override
+
         # Checkpoint normalization: allow a *single* checkpoint path (root/run/project)
         # to be provided via CLI or env. CLI takes precedence, but env-driven `--state-dir`
         # is supported as well.
@@ -359,6 +409,7 @@ async def _migrate_main(
             dest_url=str(config.destination.url),
             resources=config.resources,
             projects=getattr(config, "project_names", None),
+            project_name_mapping=config.project_name_mapping,
             state_dir=str(config.state_dir),
             dry_run=dry_run,
             logs_fetch_limit=config.migration.logs_fetch_limit,
@@ -1097,12 +1148,24 @@ async def _run_dry_run(config: Config) -> None:
                     source_client,
                     dest_client,
                     config.project_names,
+                    config.project_name_mapping,
                 )
                 progress.update(
                     validation_task,
                     description=f"✅ Discovered {len(projects)} projects",
                 )
 
+                if _resources_include(config.resources, "logs"):
+                    log_probe_results = await _test_logs_dry_run_probe(
+                        source_client,
+                        projects,
+                        config,
+                        progress,
+                        validation_task,
+                    )
+                else:
+                    log_probe_results = {}
+
                 # Test resource discovery for each migrator type
                 test_results = await _test_resource_discovery(
                     source_client,
@@ -1116,7 +1179,12 @@ async def _run_dry_run(config: Config) -> None:
             progress.update(validation_task, completed=1, total=1)
 
             # Display dry run results
-            _display_dry_run_results(projects, test_results)
+            _display_dry_run_results(
+                config,
+                projects,
+                test_results,
+                log_probe_results,
+            )
 
             console.print("\n[green]✅ Dry run completed successfully![/green]")
             console.print(
@@ -1133,6 +1201,7 @@ async def _discover_projects_read_only(
     source_client,
     dest_client,
     project_names: list[str] | None = None,
+    project_name_mapping: dict[str, str] | None = None,
 ) -> list[dict[str, str]]:
     """Discover projects without mutating destination organization.
 
@@ -1166,19 +1235,80 @@ async def _discover_projects_read_only(
         if selected_names and name not in selected_names:
             continue
 
-        dest_id_raw = dest_id_by_name.get(name)
+        dest_name = (project_name_mapping or {}).get(name, name)
+        dest_id_raw = dest_id_by_name.get(dest_name)
         dest_id = dest_id_raw if isinstance(dest_id_raw, str) else ""
         mappings.append(
             {
                 "source_id": source_id,
                 "dest_id": dest_id,
                 "name": name,
+                "dest_name": dest_name,
+                "dest_status": "exists" if dest_id else "would create",
             }
         )
 
     return mappings
 
 
+async def _test_logs_dry_run_probe(
+    source_client,
+    projects: list[dict],
+    config: Config,
+    progress,
+    validation_task,
+) -> dict[str, dict[str, Any]]:
+    """Probe whether the selected logs time window has at least one matching span."""
+    from braintrust_migrate.btql import (
+        btql_quote,
+        fetch_btql_sorted_page_with_retries,
+    )
+    from braintrust_migrate.streaming_utils import build_btql_sorted_page_query
+
+    results: dict[str, dict[str, Any]] = {}
+    for project in projects:
+        project_name = project["name"]
+        progress.update(
+            validation_task,
+            description=f"🔍 Probing logs window for {project_name}...",
+        )
+        from_expr = f"project_logs('{btql_quote(project['source_id'])}') spans"
+
+        def _query_text_for_limit(n: int, *, _from_expr: str = from_expr) -> str:
+            return build_btql_sorted_page_query(
+                from_expr=_from_expr,
+                limit=n,
+                last_pagination_key=None,
+                created_after=config.migration.created_after,
+                created_before=config.migration.created_before,
+                select="*",
+            )
+
+        try:
+            page = await fetch_btql_sorted_page_with_retries(
+                client=source_client,
+                query_for_limit=_query_text_for_limit,
+                configured_limit=1,
+                operation="btql_project_logs_dry_run_probe",
+                log_fields={"source_project_id": project["source_id"]},
+                floor_limit=1,
+                default_500_retry_limit=1,
+            )
+            rows = page.get("events")
+            match_found = isinstance(rows, list) and len(rows) > 0
+            results[project_name] = {
+                "status": "success",
+                "matching_spans_visible": match_found,
+            }
+        except Exception as e:
+            results[project_name] = {
+                "status": "error",
+                "error": str(e),
+            }
+
+    return results
+
+
 async def _test_resource_discovery(
     source_client,
     dest_client,
@@ -1268,29 +1398,78 @@ async def _test_resource_discovery(
 
 
 def _display_dry_run_results(
-    projects: list[dict], test_results: dict[str, dict]
+    config: Config,
+    projects: list[dict],
+    test_results: dict[str, dict],
+    log_probe_results: dict[str, dict[str, Any]],
 ) -> None:
     """Display dry run results in a formatted table.
 
     Args:
+        config: Migration configuration.
         projects: List of discovered projects
         test_results: Results from resource discovery tests
+        log_probe_results: Read-only logs time-window probe results.
     """
+    plan_table = Table(title="🧭 Migration Plan")
+    plan_table.add_column("Setting", style="cyan")
+    plan_table.add_column("Value", style="yellow")
+    plan_table.add_row("Resources", ", ".join(config.resources))
+    plan_table.add_row("Source Projects", ", ".join(config.project_names or ["all"]))
+    plan_table.add_row("Created After", config.migration.created_after or "none")
+    plan_table.add_row("Created Before", config.migration.created_before or "none")
+    if config.project_name_mapping:
+        map_summary = ", ".join(
+            f"{source} -> {dest}"
+            for source, dest in sorted(config.project_name_mapping.items())
+        )
+    else:
+        map_summary = "same-name fallback"
+    plan_table.add_row("Project Map", map_summary)
+
+    console.print("\n")
+    console.print(plan_table)
+
     # Projects table
     if projects:
-        projects_table = Table(title="📁 Discovered Projects")
-        projects_table.add_column("Project Name", style="cyan")
+        projects_table = Table(title="📁 Project Migration Plan")
+        projects_table.add_column("Source Project", style="cyan")
+        projects_table.add_column("Destination Project", style="yellow")
         projects_table.add_column("Source ID", style="blue")
         projects_table.add_column("Dest ID", style="green")
+        projects_table.add_column("Destination Status", style="magenta")
 
         for project in projects:
             projects_table.add_row(
-                project["name"], project["source_id"], project["dest_id"]
+                project["name"],
+                project.get("dest_name", project["name"]),
+                project["source_id"],
+                project["dest_id"] or "none",
+                project.get("dest_status", "exists" if project["dest_id"] else "would create"),
             )
 
         console.print("\n")
         console.print(projects_table)
 
+    if log_probe_results:
+        logs_table = Table(title="🧾 Logs Time-Window Probe")
+        logs_table.add_column("Project", style="cyan")
+        logs_table.add_column("Status", style="magenta")
+        logs_table.add_column("Result", style="yellow")
+
+        for project_name, result in log_probe_results.items():
+            if result["status"] == "success":
+                visible = bool(result.get("matching_spans_visible"))
+                status = "✅ Success"
+                detail = "matching spans visible" if visible else "no matching spans"
+            else:
+                status = "❌ Error"
+                detail = result.get("error", "Failed")
+            logs_table.add_row(project_name, status, detail)
+
+        console.print("\n")
+        console.print(logs_table)
+
     # Resource discovery results
     if test_results:
         resources_table = Table(title="🔍 Resource Discovery Test Results")