airbyte_ops_mcp.cli.cloud

   1# Copyright (c) 2025 Airbyte, Inc., all rights reserved.
   2"""CLI commands for Airbyte Cloud operations.
   3
   4Commands:
   5    airbyte-ops cloud connector get-version-info - Get connector version info
   6    airbyte-ops cloud connector set-version-override - Set connector version override
   7    airbyte-ops cloud connector clear-version-override - Clear connector version override
   8    airbyte-ops cloud connector regression-test - Run regression tests (single-version or comparison)
   9    airbyte-ops cloud connector fetch-connection-config - Fetch connection config to local file
  10    airbyte-ops cloud organization search - Search organizations by name/email
  11    airbyte-ops cloud organization info - Get organization info
  12    airbyte-ops cloud workspace search - Search workspaces by name/slug or email domain
  13    airbyte-ops cloud workspace info - Get workspace info
  14
  15## CLI reference
  16
  17The commands below are regenerated by `poe docs-generate` via cyclopts's
  18programmatic docs API; see `docs/generate_cli.py`.
  19
  20.. include:: ../../../docs/generated/cli/cloud.md
  21   :start-line: 2
  22"""
  23
  24from __future__ import annotations
  25
  26# Hide Python-level members from the pdoc page for this module; the rendered
  27# docs for this CLI group come entirely from the grafted `.. include::` in
  28# the module docstring above.
  29__all__: list[str] = []
  30
  31import json
  32import logging
  33import os
  34import shutil
  35import signal
  36import socket
  37import subprocess
  38import sys
  39import tempfile
  40import time
  41import uuid
  42from contextlib import suppress
  43from pathlib import Path
  44from typing import Annotated, Any, Literal
  45
  46import requests
  47import yaml
  48from airbyte.cloud.workspaces import CloudWorkspace
  49from airbyte.exceptions import PyAirbyteInputError
  50from airbyte_cdk.models.connector_metadata import MetadataFile
  51from airbyte_cdk.utils.connector_paths import find_connector_root_from_name
  52from airbyte_cdk.utils.docker import build_connector_image, verify_docker_installation
  53from airbyte_protocol.models import ConfiguredAirbyteCatalog
  54from cyclopts import Parameter
  55from fastmcp_extensions.cli import (
  56    exit_with_error,
  57    print_error,
  58    print_json,
  59    print_success,
  60    print_warning,
  61)
  62
  63from airbyte_ops_mcp.cli._base import App, app
  64from airbyte_ops_mcp.cloud_admin.auth import CloudAuthError
  65from airbyte_ops_mcp.cloud_admin.connection_config import fetch_connection_config
  66from airbyte_ops_mcp.cloud_admin.connection_state import (
  67    reset_stream_state as reset_cloud_stream_state,
  68)
  69from airbyte_ops_mcp.cloud_admin.registry_lookup import (
  70    resolve_definition_id_to_canonical_info,
  71)
  72from airbyte_ops_mcp.cloud_admin.version_overrides import (
  73    ResolvedCloudAuth,
  74    VersionOverrideTarget,
  75    get_connector_version_info,
  76)
  77from airbyte_ops_mcp.cloud_admin.version_overrides import (
  78    set_version_override as set_cloud_version_override,
  79)
  80from airbyte_ops_mcp.constants import (
  81    CLOUD_SQL_INSTANCE,
  82    CLOUD_SQL_PROXY_PID_FILE,
  83    DEFAULT_CLOUD_SQL_PROXY_PORT,
  84    ENV_GCP_PROD_DB_ACCESS_CREDENTIALS,
  85)
  86from airbyte_ops_mcp.gcp_logs import GCPSeverity, fetch_error_logs
  87from airbyte_ops_mcp.github_api import resolve_default_github_token
  88from airbyte_ops_mcp.prod_db_access.queries import (
  89    query_workspace_info,
  90    query_workspaces_by_email_domain,
  91)
  92from airbyte_ops_mcp.prod_db_access.queries import (
  93    search_organizations as _search_organizations,
  94)
  95from airbyte_ops_mcp.prod_db_access.queries import (
  96    search_workspaces as _search_workspaces,
  97)
  98from airbyte_ops_mcp.regression_tests.case_gate import (
  99    NAME_PATTERN,
 100    SANDBOX_PIN_FILE,
 101    gate_step_outputs,
 102    read_sandbox_pin,
 103    run_case_dir_gate,
 104    run_case_gate,
 105)
 106from airbyte_ops_mcp.regression_tests.case_validate import (
 107    decide_validation,
 108    gh_merge,
 109    gh_read_pr,
 110    runs_from_needs,
 111    wait_and_merge,
 112)
 113from airbyte_ops_mcp.regression_tests.cdk_secrets import get_first_config_from_secrets
 114from airbyte_ops_mcp.regression_tests.ci_output import (
 115    ComparisonOutcome,
 116    command_result_succeeded,
 117    evaluate_comparison_outcome,
 118    generate_regression_report,
 119    generate_single_version_report,
 120    is_github_actions,
 121    write_github_output,
 122    write_github_outputs,
 123    write_github_summary,
 124    write_json_output,
 125)
 126from airbyte_ops_mcp.regression_tests.config_overrides import (
 127    filter_configured_catalog_file,
 128)
 129from airbyte_ops_mcp.regression_tests.connection_fetcher import (
 130    enrich_catalog_schemas_from_cloud,
 131    fetch_connection_data,
 132    fetch_connection_state,
 133    save_connection_data_to_files,
 134    streams_missing_schemas,
 135)
 136from airbyte_ops_mcp.regression_tests.connection_secret_retriever import (
 137    SecretRetrievalError,
 138    enrich_catalog_schemas_from_platform_db,
 139    enrich_config_with_secrets,
 140    should_use_secret_retriever,
 141)
 142from airbyte_ops_mcp.regression_tests.connector_runner import (
 143    ConnectorRunner,
 144    ensure_image_available,
 145)
 146from airbyte_ops_mcp.regression_tests.http_metrics import (
 147    DEFAULT_MAX_REPLAY_DUMP_SIZE_MB,
 148    DEFAULT_REPLAY_EXTRA,
 149    DEFAULT_STREAM_LARGE_BODIES,
 150    MitmproxyManager,
 151    ReplayOptions,
 152    discard_oversized_dump,
 153    parse_http_dump,
 154)
 155from airbyte_ops_mcp.regression_tests.models import (
 156    Command,
 157    ComparableOutputs,
 158    ConnectorUnderTest,
 159    ExecutionInputs,
 160    TargetOrControl,
 161    duplicate_stream_names,
 162    get_primary_keys_per_stream,
 163    split_records_per_stream,
 164)
 165from airbyte_ops_mcp.regression_tests.regression import (
 166    ComparisonResult,
 167    RecordComparisonSummary,
 168    compare_catalog_schemas,
 169    compare_final_states,
 170    compare_specs,
 171    run_record_comparisons_from_files,
 172)
 173from airbyte_ops_mcp.regression_tests.report import (
 174    HTML_REPORT_FILENAME,
 175    CheckResult,
 176    DiffBlock,
 177    DiffLine,
 178    build_comparison_report_model,
 179    build_diff_block,
 180    build_json_block,
 181    build_single_version_report_model,
 182    consolidate_reports,
 183    failed_checks,
 184    render_inline_summary,
 185    write_html_report,
 186)
 187from airbyte_ops_mcp.regression_tests.repro_case import (
 188    CASE_RESULT_FILE,
 189    SHA_PATTERN,
 190    ReproCase,
 191    ReproductionVerdict,
 192    ReproGoal,
 193    SandboxCli,
 194    SandboxError,
 195    SandboxSide,
 196    case_result,
 197    check_twin_gaps,
 198    evaluate_reproduction,
 199    evaluate_side,
 200    restore_side,
 201    set_up_scenario,
 202)
 203from airbyte_ops_mcp.regression_tests.sandbox_support import (
 204    run_sandbox_support_check,
 205)
 206from airbyte_ops_mcp.telemetry import track_regression_test
 207from airbyte_ops_mcp.tier_cache import resolve_workspace
 208
 209# Path to connectors directory within the airbyte repo
 210CONNECTORS_SUBDIR = Path("airbyte-integrations") / "connectors"
 211
 212# Create the cloud sub-app
 213cloud_app = App(name="cloud", help="Airbyte Cloud operations.")
 214app.command(cloud_app)
 215
 216# Create the connector sub-app under cloud
 217connector_app = App(
 218    name="connector", help="Deployed connector operations in Airbyte Cloud."
 219)
 220cloud_app.command(connector_app)
 221
 222# Create the db sub-app under cloud
 223db_app = App(name="db", help="Database operations for Airbyte Cloud Prod DB Replica.")
 224cloud_app.command(db_app)
 225
 226# Create the connection sub-app under cloud
 227connection_app = App(name="connection", help="Connection operations in Airbyte Cloud.")
 228cloud_app.command(connection_app)
 229
 230# Create the state sub-app under connection
 231state_app = App(name="state", help="Connection state operations in Airbyte Cloud.")
 232connection_app.command(state_app)
 233
 234# Create the catalog sub-app under connection
 235catalog_app = App(
 236    name="catalog", help="Connection catalog operations in Airbyte Cloud."
 237)
 238connection_app.command(catalog_app)
 239
 240# Create the logs sub-app under cloud
 241logs_app = App(name="logs", help="GCP Cloud Logging operations for Airbyte Cloud.")
 242cloud_app.command(logs_app)
 243
 244# Create the organization sub-app under cloud
 245organization_app = App(
 246    name="organization", help="Organization operations in Airbyte Cloud."
 247)
 248cloud_app.command(organization_app)
 249
 250# Create the workspace sub-app under cloud
 251workspace_app = App(name="workspace", help="Workspace operations in Airbyte Cloud.")
 252cloud_app.command(workspace_app)
 253
 254
 255@db_app.command(name="start-proxy")
 256def start_proxy(
 257    port: Annotated[
 258        int,
 259        Parameter(help="Port for the Cloud SQL Proxy to listen on."),
 260    ] = DEFAULT_CLOUD_SQL_PROXY_PORT,
 261    daemon: Annotated[
 262        bool,
 263        Parameter(
 264            help="Run as daemon in background (default). Use --no-daemon for foreground."
 265        ),
 266    ] = True,
 267) -> None:
 268    """Start the Cloud SQL Proxy for database access.
 269
 270    This command starts the Cloud SQL Auth Proxy to enable connections to the
 271    Airbyte Cloud Prod DB Replica. The proxy is required for database query tools.
 272
 273    By default, runs as a daemon (background process). Use --no-daemon to run in
 274    foreground mode where you can see logs and stop with Ctrl+C.
 275
 276    Credentials are read from the GCP_PROD_DB_ACCESS_CREDENTIALS environment variable,
 277    which should contain the service account JSON credentials.
 278
 279    After starting the proxy, set these environment variables to use database tools:
 280        export USE_CLOUD_SQL_PROXY=1
 281        export DB_PORT={port}
 282
 283    Example:
 284        airbyte-ops cloud db start-proxy
 285        airbyte-ops cloud db start-proxy --port 15432
 286        airbyte-ops cloud db start-proxy --no-daemon
 287    """
 288    # Check if proxy is already running on the requested port (idempotency)
 289    try:
 290        with socket.create_connection(("127.0.0.1", port), timeout=0.5):
 291            # Something is already listening on this port
 292            pid_file = Path(CLOUD_SQL_PROXY_PID_FILE)
 293            pid_info = ""
 294            if pid_file.exists():
 295                pid_info = f" (PID: {pid_file.read_text().strip()})"
 296            print_success(
 297                f"Cloud SQL Proxy is already running on port {port}{pid_info}"
 298            )
 299            print_success("")
 300            print_success("To use database tools, set these environment variables:")
 301            print_success("  export USE_CLOUD_SQL_PROXY=1")
 302            print_success(f"  export DB_PORT={port}")
 303            return
 304    except (OSError, TimeoutError, ConnectionRefusedError):
 305        pass  # Port not in use, proceed with starting proxy
 306
 307    # Check if cloud-sql-proxy is installed
 308    proxy_path = shutil.which("cloud-sql-proxy")
 309    if not proxy_path:
 310        exit_with_error(
 311            "cloud-sql-proxy not found in PATH. "
 312            "Install it from: https://cloud.google.com/sql/docs/mysql/sql-proxy"
 313        )
 314
 315    # Get credentials from environment
 316    creds_json = os.getenv(ENV_GCP_PROD_DB_ACCESS_CREDENTIALS)
 317    if not creds_json:
 318        exit_with_error(
 319            f"{ENV_GCP_PROD_DB_ACCESS_CREDENTIALS} environment variable is not set. "
 320            "This should contain the GCP service account JSON credentials."
 321        )
 322
 323    # Build the command using --json-credentials to avoid writing to disk
 324    cmd = [
 325        proxy_path,
 326        CLOUD_SQL_INSTANCE,
 327        f"--port={port}",
 328        f"--json-credentials={creds_json}",
 329    ]
 330
 331    print_success(f"Starting Cloud SQL Proxy on port {port}...")
 332    print_success(f"Instance: {CLOUD_SQL_INSTANCE}")
 333    print_success("")
 334    print_success("To use database tools, set these environment variables:")
 335    print_success("  export USE_CLOUD_SQL_PROXY=1")
 336    print_success(f"  export DB_PORT={port}")
 337    print_success("")
 338
 339    if daemon:
 340        # Run in background (daemon mode) with log file for diagnostics
 341        log_file_path = Path("/tmp/airbyte-cloud-sql-proxy.log")
 342        log_file = log_file_path.open("ab")
 343        process = subprocess.Popen(
 344            cmd,
 345            stdout=subprocess.DEVNULL,
 346            stderr=log_file,
 347            start_new_session=True,
 348        )
 349
 350        # Brief wait to verify the process started successfully
 351        time.sleep(0.5)
 352        if process.poll() is not None:
 353            # Process exited immediately - read any error output
 354            log_file.close()
 355            error_output = ""
 356            if log_file_path.exists():
 357                error_output = log_file_path.read_text()[-1000:]  # Last 1000 chars
 358            exit_with_error(
 359                f"Cloud SQL Proxy failed to start (exit code: {process.returncode}).\n"
 360                f"Check logs at {log_file_path}\n"
 361                f"Recent output: {error_output}"
 362            )
 363
 364        # Write PID to file for stop-proxy command
 365        pid_file = Path(CLOUD_SQL_PROXY_PID_FILE)
 366        pid_file.write_text(str(process.pid))
 367        print_success(f"Cloud SQL Proxy started as daemon (PID: {process.pid})")
 368        print_success(f"Logs: {log_file_path}")
 369        print_success("To stop: airbyte-ops cloud db stop-proxy")
 370    else:
 371        # Run in foreground - replace current process
 372        # Signals (Ctrl+C) will be handled directly by the cloud-sql-proxy process
 373        print_success("Running in foreground. Press Ctrl+C to stop the proxy.")
 374        print_success("")
 375        os.execv(proxy_path, cmd)
 376
 377
 378@db_app.command(name="stop-proxy")
 379def stop_proxy() -> None:
 380    """Stop the Cloud SQL Proxy daemon.
 381
 382    This command stops a Cloud SQL Proxy that was started with 'start-proxy'.
 383    It reads the PID from the PID file and sends a SIGTERM signal to stop the process.
 384
 385    Example:
 386        airbyte-ops cloud db stop-proxy
 387    """
 388    pid_file = Path(CLOUD_SQL_PROXY_PID_FILE)
 389
 390    if not pid_file.exists():
 391        exit_with_error(
 392            f"PID file not found at {CLOUD_SQL_PROXY_PID_FILE}. "
 393            "No Cloud SQL Proxy daemon appears to be running."
 394        )
 395
 396    pid_str = pid_file.read_text().strip()
 397    if not pid_str.isdigit():
 398        pid_file.unlink()
 399        exit_with_error(f"Invalid PID in {CLOUD_SQL_PROXY_PID_FILE}: {pid_str}")
 400
 401    pid = int(pid_str)
 402
 403    # Check if process is still running
 404    try:
 405        os.kill(pid, 0)  # Signal 0 just checks if process exists
 406    except ProcessLookupError:
 407        pid_file.unlink()
 408        print_success(
 409            f"Cloud SQL Proxy (PID: {pid}) is not running. Cleaned up PID file."
 410        )
 411        return
 412    except PermissionError:
 413        exit_with_error(f"Permission denied to check process {pid}.")
 414
 415    # Send SIGTERM to stop the process
 416    try:
 417        os.kill(pid, signal.SIGTERM)
 418        print_success(f"Sent SIGTERM to Cloud SQL Proxy (PID: {pid}).")
 419    except ProcessLookupError:
 420        print_success(f"Cloud SQL Proxy (PID: {pid}) already stopped.")
 421    except PermissionError:
 422        exit_with_error(f"Permission denied to stop process {pid}.")
 423
 424    # Clean up PID file
 425    pid_file.unlink(missing_ok=True)
 426    print_success("Cloud SQL Proxy stopped.")
 427
 428
 429@connector_app.command(name="get-version-info")
 430def get_version_info(
 431    workspace_id: Annotated[
 432        str,
 433        Parameter(help="The Airbyte Cloud workspace ID."),
 434    ],
 435    connector_id: Annotated[
 436        str,
 437        Parameter(help="The ID of the deployed connector (source or destination)."),
 438    ],
 439    connector_type: Annotated[
 440        Literal["source", "destination"],
 441        Parameter(help="The type of connector."),
 442    ],
 443) -> None:
 444    """Get the current version information for a deployed connector."""
 445    result = get_connector_version_info(
 446        auth=_resolve_cli_cloud_auth(),
 447        workspace_id=workspace_id,
 448        actor_id=connector_id,
 449        actor_type=connector_type,
 450    )
 451    print_json(result.model_dump())
 452
 453
 454_OverrideLevel = Literal["actor", "workspace", "organization"]
 455_TierFilter = Literal["TIER_0", "TIER_1", "TIER_2", "UNKNOWN", "ALL"]
 456
 457
 458def _resolve_cli_cloud_auth() -> ResolvedCloudAuth:
 459    """Resolve Airbyte Cloud auth from environment variables for CLI use.
 460
 461    Mirrors the priority order the MCP server uses:
 462
 463    1. `AIRBYTE_CLOUD_BEARER_TOKEN`
 464    2. `AIRBYTE_CLOUD_CLIENT_ID` + `AIRBYTE_CLOUD_CLIENT_SECRET`
 465
 466    Raises a CLI error via `exit_with_error` if no usable credentials are present.
 467    """
 468    bearer_token = os.environ.get("AIRBYTE_CLOUD_BEARER_TOKEN")
 469    if bearer_token:
 470        return ResolvedCloudAuth(bearer_token=bearer_token)
 471
 472    client_id = os.environ.get("AIRBYTE_CLOUD_CLIENT_ID")
 473    client_secret = os.environ.get("AIRBYTE_CLOUD_CLIENT_SECRET")
 474    if client_id and client_secret:
 475        return ResolvedCloudAuth(client_id=client_id, client_secret=client_secret)
 476
 477    exit_with_error(
 478        "Missing Airbyte Cloud credentials. Set AIRBYTE_CLOUD_BEARER_TOKEN, or "
 479        "both AIRBYTE_CLOUD_CLIENT_ID and AIRBYTE_CLOUD_CLIENT_SECRET."
 480    )
 481
 482
 483def _validate_override_scope_args(
 484    *,
 485    override_level: _OverrideLevel,
 486    workspace_id: str | None,
 487    organization_id: str | None,
 488    connector_id: str | None,
 489    connector_type: Literal["source", "destination"] | None,
 490    actor_definition_id: str | None,
 491) -> None:
 492    """Validate scope-specific arguments for `set/clear-version-override`.
 493
 494    Calls `exit_with_error` (which terminates the process) if the supplied
 495    arguments are inconsistent with the requested `override_level`.
 496    """
 497    if override_level == "actor":
 498        missing = [
 499            name
 500            for name, value in (
 501                ("--workspace-id", workspace_id),
 502                ("--connector-id", connector_id),
 503                ("--connector-type", connector_type),
 504            )
 505            if not value
 506        ]
 507        if missing:
 508            exit_with_error(
 509                "Override level 'actor' requires " + ", ".join(missing) + "."
 510            )
 511        if actor_definition_id is not None:
 512            exit_with_error(
 513                "Override level 'actor' must not be combined with "
 514                "--actor-definition-id."
 515            )
 516        if organization_id is not None:
 517            exit_with_error(
 518                "Override level 'actor' must not be combined with --organization-id."
 519            )
 520        return
 521
 522    if override_level == "workspace":
 523        missing = [
 524            name
 525            for name, value in (
 526                ("--workspace-id", workspace_id),
 527                ("--actor-definition-id", actor_definition_id),
 528            )
 529            if not value
 530        ]
 531        if missing:
 532            exit_with_error(
 533                "Override level 'workspace' requires " + ", ".join(missing) + "."
 534            )
 535        if connector_id is not None or connector_type is not None:
 536            exit_with_error(
 537                "Override level 'workspace' must not be combined with "
 538                "--connector-id or --connector-type."
 539            )
 540        if organization_id is not None:
 541            exit_with_error(
 542                "Override level 'workspace' must not be combined with "
 543                "--organization-id."
 544            )
 545        return
 546
 547    # override_level == "organization"
 548    missing = [
 549        name
 550        for name, value in (
 551            ("--organization-id", organization_id),
 552            ("--actor-definition-id", actor_definition_id),
 553        )
 554        if not value
 555    ]
 556    if missing:
 557        exit_with_error(
 558            "Override level 'organization' requires " + ", ".join(missing) + "."
 559        )
 560    if connector_id is not None or connector_type is not None:
 561        exit_with_error(
 562            "Override level 'organization' must not be combined with "
 563            "--connector-id or --connector-type."
 564        )
 565    if workspace_id is not None:
 566        exit_with_error(
 567            "Override level 'organization' must not be combined with --workspace-id."
 568        )
 569
 570
 571def _dispatch_version_override(
 572    *,
 573    override_level: _OverrideLevel,
 574    workspace_id: str | None,
 575    organization_id: str | None,
 576    connector_id: str | None,
 577    connector_type: Literal["source", "destination"] | None,
 578    actor_definition_id: str | None,
 579    version: str | None,
 580    unset: bool,
 581    reason: str | None,
 582    reason_url: str | None,
 583    issue_url: str,
 584    approval_comment_url: str,
 585    ai_agent_session_url: str | None,
 586    customer_tier_filter: _TierFilter,
 587) -> None:
 588    """Route a version-override request through the normalized core helper.
 589
 590    Resolves `--actor-definition-id` to its canonical name and connector type
 591    via the cloud registry for `workspace`/`organization` scopes. Prints the
 592    operation result as JSON.
 593    """
 594    _validate_override_scope_args(
 595        override_level=override_level,
 596        workspace_id=workspace_id,
 597        organization_id=organization_id,
 598        connector_id=connector_id,
 599        connector_type=connector_type,
 600        actor_definition_id=actor_definition_id,
 601    )
 602
 603    auth = _resolve_cli_cloud_auth()
 604
 605    if override_level == "actor":
 606        assert workspace_id is not None
 607        assert connector_id is not None
 608        assert connector_type is not None
 609        assert organization_id is None
 610        try:
 611            ws_resolution = resolve_workspace(workspace_id)
 612            if not ws_resolution.organization_id:
 613                exit_with_error("Could not resolve organization for workspace.")
 614            result = set_cloud_version_override(
 615                auth=auth,
 616                target=VersionOverrideTarget(
 617                    scope="actor",
 618                    organization_id=ws_resolution.organization_id,
 619                    workspace_id=workspace_id,
 620                    actor_id=connector_id,
 621                    connector_type=connector_type,
 622                ),
 623                approval_comment_url=approval_comment_url,
 624                version=version,
 625                unset=unset,
 626                override_reason=reason,
 627                override_reason_reference_url=reason_url,
 628                issue_url=issue_url,
 629                ai_agent_session_url=ai_agent_session_url,
 630                customer_tier_filter=customer_tier_filter,
 631            )
 632        except (PyAirbyteInputError, CloudAuthError) as e:
 633            exit_with_error(str(e))
 634        _print_override_result(result.success, result.message)
 635        print_json(result.model_dump())
 636        return
 637
 638    try:
 639        assert actor_definition_id is not None
 640        canonical_name, resolved_connector_type = (
 641            resolve_definition_id_to_canonical_info(actor_definition_id)
 642        )
 643    except (PyAirbyteInputError, requests.RequestException) as e:
 644        exit_with_error(f"Failed to resolve --actor-definition-id: {e}")
 645    typed_connector_type: Literal["source", "destination"] = (
 646        "source" if resolved_connector_type == "source" else "destination"
 647    )
 648
 649    if override_level == "workspace":
 650        assert workspace_id is not None
 651        try:
 652            ws_resolution = resolve_workspace(workspace_id)
 653            if not ws_resolution.organization_id:
 654                exit_with_error("Could not resolve organization for workspace.")
 655            target = VersionOverrideTarget(
 656                scope="workspace",
 657                organization_id=ws_resolution.organization_id,
 658                workspace_id=workspace_id,
 659                connector_name=canonical_name,
 660                connector_type=typed_connector_type,
 661            )
 662        except PyAirbyteInputError as e:
 663            exit_with_error(str(e))
 664    else:
 665        assert organization_id is not None
 666        target = VersionOverrideTarget(
 667            scope="organization",
 668            organization_id=organization_id,
 669            connector_name=canonical_name,
 670            connector_type=typed_connector_type,
 671        )
 672
 673    try:
 674        result = set_cloud_version_override(
 675            auth=auth,
 676            target=target,
 677            approval_comment_url=approval_comment_url,
 678            version=version,
 679            unset=unset,
 680            override_reason=reason,
 681            override_reason_reference_url=reason_url,
 682            issue_url=issue_url,
 683            ai_agent_session_url=ai_agent_session_url,
 684            customer_tier_filter=customer_tier_filter,
 685        )
 686    except (PyAirbyteInputError, CloudAuthError) as e:
 687        exit_with_error(str(e))
 688    _print_override_result(result.success, result.message)
 689    print_json(result.model_dump())
 690
 691
 692def _print_override_result(success: bool, message: str) -> None:
 693    """Emit an override-operation status message via the shared CLI helpers."""
 694    if success:
 695        print_success(message)
 696    else:
 697        print_error(message)
 698
 699
 700@connector_app.command(name="set-version-override")
 701def set_version_override(
 702    version: Annotated[
 703        str,
 704        Parameter(
 705            help="The semver version string to pin to (e.g., '2.1.5-preview.abc1234')."
 706        ),
 707    ],
 708    reason: Annotated[
 709        str,
 710        Parameter(help="Explanation for the override (min 10 characters)."),
 711    ],
 712    issue_url: Annotated[
 713        str,
 714        Parameter(help="GitHub issue URL providing context for this operation."),
 715    ],
 716    approval_comment_url: Annotated[
 717        str,
 718        Parameter(
 719            help="Slack approval record URL where admin authorized this deployment."
 720        ),
 721    ],
 722    override_level: Annotated[
 723        _OverrideLevel,
 724        Parameter(
 725            help=(
 726                "Scope at which to apply the version override: 'actor' (single "
 727                "deployed connector instance), 'workspace' (all instances of a "
 728                "connector type within a workspace), or 'organization' (all "
 729                "instances across an organization). Defaults to 'actor'."
 730            ),
 731        ),
 732    ] = "actor",
 733    workspace_id: Annotated[
 734        str | None,
 735        Parameter(
 736            help=(
 737                "The Airbyte Cloud workspace ID. Required for 'actor' and "
 738                "'workspace' override levels."
 739            )
 740        ),
 741    ] = None,
 742    organization_id: Annotated[
 743        str | None,
 744        Parameter(
 745            help=(
 746                "The Airbyte Cloud organization ID. Required for "
 747                "'organization' override level."
 748            )
 749        ),
 750    ] = None,
 751    connector_id: Annotated[
 752        str | None,
 753        Parameter(
 754            help=(
 755                "The ID of the deployed connector (source or destination). "
 756                "Required for 'actor' override level."
 757            )
 758        ),
 759    ] = None,
 760    connector_type: Annotated[
 761        Literal["source", "destination"] | None,
 762        Parameter(help="The type of connector. Required for 'actor' override level."),
 763    ] = None,
 764    actor_definition_id: Annotated[
 765        str | None,
 766        Parameter(
 767            help=(
 768                "The connector definition UUID. Required for 'workspace' and "
 769                "'organization' override levels."
 770            )
 771        ),
 772    ] = None,
 773    ai_agent_session_url: Annotated[
 774        str | None,
 775        Parameter(
 776            help="URL to AI agent session driving this operation (for auditability)."
 777        ),
 778    ] = None,
 779    reason_url: Annotated[
 780        str | None,
 781        Parameter(help="Optional URL with more context (e.g., issue link)."),
 782    ] = None,
 783    customer_tier_filter: Annotated[
 784        _TierFilter,
 785        Parameter(
 786            help=(
 787                "Tier filter: 'TIER_0', 'TIER_1', 'TIER_2', 'UNKNOWN', or 'ALL'. "
 788                "The operation will be rejected if the actual customer tier does not match. "
 789                "Defaults to 'TIER_2' (non-sensitive customers)."
 790            ),
 791        ),
 792    ] = "TIER_2",
 793) -> None:
 794    """Set a version override for a deployed connector.
 795
 796    Requires admin authentication via `AIRBYTE_INTERNAL_ADMIN_FLAG` and
 797    `AIRBYTE_INTERNAL_ADMIN_USER` environment variables.
 798
 799    The `--override-level` flag selects the scope at which the pin is applied:
 800
 801    - `actor` (default): pins a single deployed connector instance. Requires
 802      `--workspace-id`, `--connector-id`, and `--connector-type`.
 803    - `workspace`: pins ALL instances of a connector type within a workspace.
 804      Requires `--workspace-id` and `--actor-definition-id`.
 805    - `organization`: pins ALL instances of a connector type across an
 806      organization. Requires `--organization-id` and `--actor-definition-id`.
 807
 808    The `customer_tier_filter` gates the operation: the call fails if the
 809    actual tier of the target organization does not match. Use `ALL` to
 810    proceed regardless of tier (a warning is shown for sensitive tiers).
 811    """
 812    _dispatch_version_override(
 813        override_level=override_level,
 814        workspace_id=workspace_id,
 815        organization_id=organization_id,
 816        connector_id=connector_id,
 817        connector_type=connector_type,
 818        actor_definition_id=actor_definition_id,
 819        version=version,
 820        unset=False,
 821        reason=reason,
 822        reason_url=reason_url,
 823        issue_url=issue_url,
 824        approval_comment_url=approval_comment_url,
 825        ai_agent_session_url=ai_agent_session_url,
 826        customer_tier_filter=customer_tier_filter,
 827    )
 828
 829
 830@connector_app.command(name="clear-version-override")
 831def clear_version_override(
 832    issue_url: Annotated[
 833        str,
 834        Parameter(help="GitHub issue URL providing context for this operation."),
 835    ],
 836    approval_comment_url: Annotated[
 837        str,
 838        Parameter(
 839            help="Slack approval record URL where admin authorized this deployment."
 840        ),
 841    ],
 842    override_level: Annotated[
 843        _OverrideLevel,
 844        Parameter(
 845            help=(
 846                "Scope at which to clear the version override: 'actor', "
 847                "'workspace', or 'organization'. Defaults to 'actor'."
 848            ),
 849        ),
 850    ] = "actor",
 851    workspace_id: Annotated[
 852        str | None,
 853        Parameter(
 854            help=(
 855                "The Airbyte Cloud workspace ID. Required for 'actor' and "
 856                "'workspace' override levels."
 857            )
 858        ),
 859    ] = None,
 860    organization_id: Annotated[
 861        str | None,
 862        Parameter(
 863            help=(
 864                "The Airbyte Cloud organization ID. Required for "
 865                "'organization' override level."
 866            )
 867        ),
 868    ] = None,
 869    connector_id: Annotated[
 870        str | None,
 871        Parameter(
 872            help=(
 873                "The ID of the deployed connector (source or destination). "
 874                "Required for 'actor' override level."
 875            )
 876        ),
 877    ] = None,
 878    connector_type: Annotated[
 879        Literal["source", "destination"] | None,
 880        Parameter(help="The type of connector. Required for 'actor' override level."),
 881    ] = None,
 882    actor_definition_id: Annotated[
 883        str | None,
 884        Parameter(
 885            help=(
 886                "The connector definition UUID. Required for 'workspace' and "
 887                "'organization' override levels."
 888            )
 889        ),
 890    ] = None,
 891    ai_agent_session_url: Annotated[
 892        str | None,
 893        Parameter(
 894            help="URL to AI agent session driving this operation (for auditability)."
 895        ),
 896    ] = None,
 897    customer_tier_filter: Annotated[
 898        _TierFilter,
 899        Parameter(
 900            help=(
 901                "Tier filter: 'TIER_0', 'TIER_1', 'TIER_2', 'UNKNOWN', or 'ALL'. "
 902                "The operation will be rejected if the actual customer tier does not match. "
 903                "Defaults to 'TIER_2' (non-sensitive customers)."
 904            ),
 905        ),
 906    ] = "TIER_2",
 907) -> None:
 908    """Clear a version override from a deployed connector.
 909
 910    Requires admin authentication via `AIRBYTE_INTERNAL_ADMIN_FLAG` and
 911    `AIRBYTE_INTERNAL_ADMIN_USER` environment variables.
 912
 913    The `--override-level` flag selects the scope at which the pin is removed:
 914
 915    - `actor` (default): clears a single deployed connector instance pin.
 916      Requires `--workspace-id`, `--connector-id`, and `--connector-type`.
 917    - `workspace`: clears the workspace-level pin for a connector type.
 918      Requires `--workspace-id` and `--actor-definition-id`.
 919    - `organization`: clears the organization-level pin for a connector type.
 920      Requires `--organization-id` and `--actor-definition-id`.
 921    """
 922    _dispatch_version_override(
 923        override_level=override_level,
 924        workspace_id=workspace_id,
 925        organization_id=organization_id,
 926        connector_id=connector_id,
 927        connector_type=connector_type,
 928        actor_definition_id=actor_definition_id,
 929        version=None,
 930        unset=True,
 931        reason=None,
 932        reason_url=None,
 933        issue_url=issue_url,
 934        approval_comment_url=approval_comment_url,
 935        ai_agent_session_url=ai_agent_session_url,
 936        customer_tier_filter=customer_tier_filter,
 937    )
 938
 939
 940def _load_json_file(file_path: Path) -> dict | None:
 941    """Load a JSON file and return its contents.
 942
 943    Returns None if the file doesn't exist or contains invalid JSON.
 944    """
 945    if not file_path.exists():
 946        return None
 947    try:
 948        return json.loads(file_path.read_text())
 949    except json.JSONDecodeError as e:
 950        print_error(f"Failed to parse JSON in file: {file_path}\nError: {e}")
 951        return None
 952
 953
 954def _parse_extra_env(pairs: list[str] | None) -> dict[str, str]:
 955    """Parse repeatable KEY=VALUE arguments for connector container env."""
 956    parsed: dict[str, str] = {}
 957    for item in pairs or []:
 958        if "=" not in item:
 959            exit_with_error(f"Invalid --extra-env value (expected KEY=VALUE): {item}")
 960        key, value = item.split("=", 1)
 961        key = key.strip()
 962        if not key:
 963            exit_with_error(f"Invalid --extra-env value (expected KEY=VALUE): {item}")
 964        parsed[key] = value
 965    return parsed
 966
 967
 968def _run_connector_command(
 969    connector_image: str,
 970    command: Command,
 971    output_dir: Path,
 972    target_or_control: TargetOrControl,
 973    config_path: Path | None = None,
 974    catalog_path: Path | None = None,
 975    state_path: Path | None = None,
 976    proxy_url: str | None = None,
 977    ca_cert_path: Path | None = None,
 978    enable_debug_logs: bool = False,
 979    extra_env: dict[str, str] | None = None,
 980) -> tuple[dict, ComparableOutputs]:
 981    """Run a connector command and return its results and its declared objects.
 982
 983    The dict is a flattened view of `ExecutionResult`: counts, exit status and
 984    artifact paths, which is all the reports need. What a comparison needs
 985    instead -- the spec, the discovered catalog -- travels beside it as
 986    `ComparableOutputs` rather than growing a dict that several consumers read.
 987    The `ExecutionResult` itself is still dropped here: it holds every parsed
 988    message, and a comparison keeps both sides alive at once.
 989
 990    Args:
 991        connector_image: Full connector image name with tag.
 992        command: The Airbyte command to run.
 993        output_dir: Directory to store output files.
 994        target_or_control: Whether this is target or control version.
 995        config_path: Path to connector config JSON file.
 996        catalog_path: Path to configured catalog JSON file.
 997        state_path: Path to state JSON file.
 998        proxy_url: Optional HTTP proxy URL for traffic capture.
 999        ca_cert_path: Path to the proxy's CA certificate, so the container trusts
1000            it. Without it every HTTPS request through the proxy fails TLS
1001            verification and the capture comes back near-empty.
1002        enable_debug_logs: If `True`, pass `LOG_LEVEL=DEBUG` to the connector container.
1003        extra_env: Additional environment variables for the connector container.
1004
1005    Returns:
1006        The flattened results, and the objects the connector declared.
1007    """
1008    connector = ConnectorUnderTest.from_image_name(connector_image, target_or_control)
1009
1010    config = _load_json_file(config_path) if config_path else None
1011    state = _load_json_file(state_path) if state_path else None
1012
1013    configured_catalog = None
1014    if catalog_path and catalog_path.exists():
1015        catalog_json = catalog_path.read_text()
1016        configured_catalog = ConfiguredAirbyteCatalog.parse_raw(catalog_json)
1017
1018    # Pass log level as an environment variable to the connector container
1019    container_env: dict[str, str] = {}
1020    if enable_debug_logs:
1021        container_env["LOG_LEVEL"] = "DEBUG"
1022    container_env.update(extra_env or {})
1023
1024    execution_inputs = ExecutionInputs(
1025        connector_under_test=connector,
1026        command=command,
1027        output_dir=output_dir,
1028        config=config,
1029        configured_catalog=configured_catalog,
1030        state=state,
1031        environment_variables=container_env or None,
1032    )
1033
1034    runner = ConnectorRunner(
1035        execution_inputs, proxy_url=proxy_url, ca_cert_path=ca_cert_path
1036    )
1037    result = runner.run()
1038
1039    result.save_artifacts(output_dir)
1040
1041    result_dict: dict[str, object] = {
1042        "connector": connector_image,
1043        "command": command.value,
1044        "success": result.success,
1045        "exit_code": result.exit_code,
1046        "stdout_file": str(result.stdout_file_path),
1047        "stderr_file": str(result.stderr_file_path),
1048        "message_counts": {
1049            k.value: v for k, v in result.get_message_count_per_type().items()
1050        },
1051        "record_counts_per_stream": result.get_record_count_per_stream(),
1052    }
1053
1054    # For CHECK, include the parsed connectionStatus from the Airbyte protocol.
1055    # The CDK exits 0 regardless of check outcome; the real pass/fail signal
1056    # lives in the CONNECTION_STATUS message (status=SUCCEEDED or FAILED).
1057    if command == Command.CHECK:
1058        conn_status = result.get_connection_status()
1059        if conn_status is not None:
1060            result_dict["connection_status"] = conn_status.status.value
1061            result_dict["connection_status_message"] = conn_status.message or ""
1062        else:
1063            result_dict["connection_status"] = None
1064            result_dict["connection_status_message"] = ""
1065
1066    return result_dict, ComparableOutputs.from_execution_result(result)
1067
1068
1069def _annotate_run_verdict(command: str, result: dict[str, Any]) -> dict[str, Any]:
1070    """Copy a run result with `success` as the verdict, exit status kept separately.
1071
1072    `ExecutionResult.success` is the container exit status, which for `check` says
1073    nothing about whether the connector could connect -- the CDK exits 0 even when a
1074    check fails. Any payload read as the verdict must therefore not expose the exit
1075    status under `success`, at any nesting level. The raw value stays available as
1076    `exited_cleanly`; for every command but `check` the two are equal.
1077
1078    Returns a copy: callers pass the unannotated originals to the report generator,
1079    which reads `success` as the exit status.
1080    """
1081    return {
1082        **result,
1083        "success": command_result_succeeded(command, result),
1084        "exited_cleanly": result["success"],
1085    }
1086
1087
1088# The checks the output comparisons contribute, named for the row a reviewer
1089# reads in the report and in the step summary.
1090_SPEC_CHECK = "Spec compatibility"
1091_CATALOG_SCHEMA_CHECK = "Catalog schema"
1092_FINAL_STATE_CHECK = "Final state per stream"
1093
1094# How many individual changes a check's one-line summary names before it says
1095# how many more there are. The full list is in the report and in the run's JSON
1096# payload, and the comparators order their findings worst-first, so the ones
1097# named here are the ones worth naming.
1098_MAX_LISTED_CHANGES = 3
1099
1100# How many changes the JSON payload lists per comparison. Same reason as
1101# `_MAX_DIFF_PAYLOAD_CHARS`: the payload is written to `GITHUB_OUTPUT`, and a
1102# reordered `oneOf` on a wide spec produces hundreds of sentences.
1103_MAX_PAYLOAD_CHANGES = 100
1104
1105# How many per-stream entries it lists. A wide source has thousands of streams,
1106# and one line each is enough to exceed the same output budget without a single
1107# diff being attached.
1108_MAX_PAYLOAD_STREAMS = 100
1109
1110# How much of a comparison's detail the run's JSON payload carries. That payload
1111# is written to `GITHUB_OUTPUT`, which is size-limited, and a wide connector's
1112# catalog diff has no natural bound -- so the payload keeps as many stream diffs
1113# as fit and says how many it dropped. The catalogs themselves are saved
1114# artifacts, which is where an unbounded diff belongs.
1115_MAX_DIFF_PAYLOAD_CHARS = 50_000
1116
1117
1118def _compare_outputs(
1119    *,
1120    command: str,
1121    outcome: ComparisonOutcome,
1122    control_outputs: ComparableOutputs,
1123    target_outputs: ComparableOutputs,
1124) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1125    """Compare what the two versions produced, for the commands that produce it.
1126
1127    This is the content check for `spec`, `discover` and `read`: without it their
1128    verdict is the exit code, and a connector that drops a config field, changes
1129    a stream's schema or corrupts the state the next sync resumes from exits 0
1130    all the way to a release. `check` reports its own status, so it is the one
1131    command with nothing to diff here.
1132
1133    Only runs when both commands succeeded. A failed run has nothing to compare,
1134    and reporting "no spec to compare" beside the command failure that caused it
1135    would state the same problem twice, in weaker terms the second time.
1136
1137    A comparator that raises is left to propagate. The comparators normalise what
1138    they are given, so the only ways here are a programming error in a future
1139    caller or a schema nested past the recursion limit -- and the workflow
1140    already has the right vocabulary for both: a step that dies is an
1141    `internal_failure`, reported as infrastructure rather than as a regression,
1142    with the connector's artifacts uploaded either way. Catching it here would
1143    only let a tooling bug read as "the new version broke its spec".
1144
1145    Args:
1146        command: The Airbyte command that was run, as the CLI names it. This is
1147            always `"read"` for a read, including one given a state file: the
1148            `READ` -> `READ_WITH_STATE` upgrade only changes the `Command` the
1149            container is run with.
1150        outcome: The already-evaluated command outcome.
1151        control_outputs: What the control version produced.
1152        target_outputs: What the target version produced.
1153
1154    Returns:
1155        The checks to fold into the verdict, the diff blocks that show what they
1156        found, and the comparison detail for the run's JSON payload -- all empty
1157        when there was nothing to compare.
1158    """
1159    if not outcome.success:
1160        return (), (), {}
1161
1162    if command == "spec":
1163        return _spec_comparison(control_outputs.spec, target_outputs.spec)
1164
1165    if command == "discover":
1166        return _catalog_comparison(control_outputs.catalog, target_outputs.catalog)
1167
1168    if command == "read":
1169        return _final_state_comparison(
1170            control_outputs.final_states, target_outputs.final_states
1171        )
1172
1173    return (), (), {}
1174
1175
1176def _spec_comparison(
1177    control_spec: Any, target_spec: Any
1178) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1179    """Run the spec comparison and shape it for every surface."""
1180    result = compare_specs(control_spec, target_spec)
1181
1182    check = CheckResult(
1183        name=_SPEC_CHECK,
1184        passed=result.passed,
1185        summary=_change_summary(
1186            result.message,
1187            # A passing spec check still says what was added: an additive change
1188            # is the thing a reviewer most wants to see confirmed.
1189            result.errors if result.errors else result.warnings,
1190        ),
1191        details=tuple(result.errors),
1192    )
1193    # The incompatible changes are the check's `details`, which the report renders
1194    # under the check row and the step summary lists -- a block would repeat them.
1195    # The compatible ones have nowhere else to go, and a reviewer asking "what
1196    # else moved?" is asking about exactly those.
1197    blocks = tuple(
1198        block
1199        for block in (
1200            _change_block("Spec — compatible changes", result.warnings, "add"),
1201        )
1202        if block is not None
1203    )
1204
1205    return (check,), blocks, {"spec": _spec_detail(result)}
1206
1207
1208def _catalog_comparison(
1209    control_catalog: Any, target_catalog: Any
1210) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1211    """Run the discovered-catalog comparison and shape it for every surface."""
1212    result = compare_catalog_schemas(control_catalog, target_catalog)
1213
1214    check = CheckResult(
1215        name=_CATALOG_SCHEMA_CHECK,
1216        passed=result.passed,
1217        # A catalog that only grew -- new streams, new fields, nothing removed
1218        # or retyped -- is reported and does not gate the release. Connectors
1219        # add streams constantly, and a check that goes red on every one of them
1220        # is a check people learn to click past.
1221        severity="diagnostic" if result.additive_only else "strict",
1222        summary=_change_summary(result.message, result.errors),
1223        details=tuple(result.errors),
1224    )
1225
1226    # The findings themselves are the check's `details`, which the report renders
1227    # under the check row: a stream that is missing or new has no schema diff, so
1228    # that list is the only place it appears.
1229    blocks = _stream_diff_blocks(
1230        result,
1231        block_title="Schema diff",
1232        omission_title="Schema diffs omitted for size",
1233        artifact_hint=(
1234            "Both discovered catalogs are in the artifact, under "
1235            "airbyte_messages/catalog.jsonl."
1236        ),
1237    )
1238
1239    return (check,), blocks, {"catalog_schema": _catalog_schema_detail(result)}
1240
1241
1242def _final_state_comparison(
1243    control_states: dict[tuple[str | None, str], Any] | None,
1244    target_states: dict[tuple[str | None, str], Any] | None,
1245) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1246    """Run the final-state comparison and shape it for every surface."""
1247    result = compare_final_states(control_states, target_states)
1248
1249    # Structural findings first, then the value-only ones: the check lists
1250    # everything it found, and the summary quotes the front of that list.
1251    findings = [*result.errors, *result.warnings]
1252
1253    check = CheckResult(
1254        name=_FINAL_STATE_CHECK,
1255        passed=result.passed,
1256        # Two ways a non-passing state check does not gate. A cursor that only
1257        # advanced: until HTTP replay lands, the two versions call the live API
1258        # in separate runs and a timestamp cursor moves between them on its own.
1259        # And a read with no state on either side, where there was nothing to
1260        # compare rather than something wrong.
1261        severity=(
1262            "diagnostic" if result.value_only or result.inconclusive else "strict"
1263        ),
1264        summary=_change_summary(result.message, findings),
1265        details=tuple(findings),
1266    )
1267
1268    blocks = _stream_diff_blocks(
1269        result,
1270        block_title="Final state diff",
1271        omission_title="Final state diffs omitted for size",
1272        artifact_hint=(
1273            "Both versions' state messages are in the artifact, under "
1274            "airbyte_messages/state.jsonl."
1275        ),
1276    )
1277
1278    return (check,), blocks, {"final_state": _final_state_detail(result)}
1279
1280
1281def _stream_diff_blocks(
1282    result: ComparisonResult,
1283    *,
1284    block_title: str,
1285    omission_title: str,
1286    artifact_hint: str,
1287) -> tuple[DiffBlock, ...]:
1288    """One block per changed stream, within the report's diff budget.
1289
1290    In comparison order -- only a *changed* stream has a diff, so there is no
1291    severity left to sort by here -- and within the same budget the payload uses:
1292    a report that inlines every schema of a 200-stream source is not a report
1293    anyone reads. What the budget drops is named rather than skipped, since a
1294    block that is simply absent reads as "there was nothing to see".
1295    """
1296    blocks: list[DiffBlock] = []
1297    budget = _MAX_DIFF_PAYLOAD_CHARS
1298    dropped: list[str] = []
1299
1300    for name in result.failed_streams:
1301        stream = result.stream_results[name]
1302        if not stream.schema_diff:
1303            continue
1304
1305        payload = json.dumps(stream.schema_diff, indent=2, sort_keys=True)
1306        if len(payload) > budget:
1307            dropped.append(name)
1308            continue
1309
1310        budget -= len(payload)
1311        blocks.append(build_json_block(f"{block_title} — {name}", payload))
1312
1313    if dropped:
1314        blocks.append(
1315            build_diff_block(
1316                omission_title,
1317                [
1318                    DiffLine(
1319                        kind="meta",
1320                        text=(
1321                            "These streams changed, and their diffs were too "
1322                            f"large for this report. {artifact_hint}"
1323                        ),
1324                    ),
1325                    *(DiffLine(kind="del", text=name) for name in dropped),
1326                ],
1327                language="text",
1328            )
1329        )
1330
1331    return tuple(blocks)
1332
1333
1334def _change_block(title: str, changes: list[str], kind: str) -> DiffBlock | None:
1335    """One diff block listing a comparison's findings, or `None` if there are none."""
1336    if not changes:
1337        return None
1338
1339    return build_diff_block(
1340        title,
1341        [DiffLine(kind=kind, text=change) for change in changes],
1342        language="text",
1343    )
1344
1345
1346def _change_summary(message: str, changes: list[str]) -> str:
1347    """One line: what the comparison concluded, and the first few reasons."""
1348    if not changes:
1349        return message
1350
1351    listed = "; ".join(changes[:_MAX_LISTED_CHANGES])
1352    remaining = len(changes) - _MAX_LISTED_CHANGES
1353    if remaining > 0:
1354        listed = f"{listed}; and {remaining} more"
1355
1356    return f"{message}: {listed}"
1357
1358
1359def _bounded_changes(changes: list[str]) -> list[str]:
1360    """A change list the JSON payload can carry, saying what it left out."""
1361    if len(changes) <= _MAX_PAYLOAD_CHANGES:
1362        return changes
1363
1364    dropped = len(changes) - _MAX_PAYLOAD_CHANGES
1365
1366    return [
1367        *changes[:_MAX_PAYLOAD_CHANGES],
1368        f"... and {dropped} more, in the report and the console output",
1369    ]
1370
1371
1372def _spec_detail(result: ComparisonResult) -> dict[str, Any]:
1373    """The spec comparison, for the run's JSON payload."""
1374    return {
1375        "passed": result.passed,
1376        "message": result.message,
1377        "breaking_changes": _bounded_changes(result.errors),
1378        "compatible_changes": _bounded_changes(result.warnings),
1379    }
1380
1381
1382def _catalog_schema_detail(result: ComparisonResult) -> dict[str, Any]:
1383    """The catalog comparison, for the run's JSON payload."""
1384    streams, omitted, dropped = _bounded_stream_diffs(
1385        result,
1386        omission_note="Too large for this payload; diff the catalogs",
1387        artifact_hint="Both discovered catalogs are in the saved artifacts.",
1388    )
1389
1390    detail: dict[str, Any] = {
1391        "passed": result.passed,
1392        # Why a `passed: false` did not fail the run: everything found was an
1393        # addition, which is reported as a warning rather than a regression.
1394        "additive_only": result.additive_only,
1395        "message": result.message,
1396        "changed_streams": _bounded_changes(result.failed_streams),
1397        "streams": streams,
1398    }
1399    if omitted:
1400        detail["streams_with_omitted_diffs"] = omitted
1401    if dropped:
1402        detail["streams_omitted"] = dropped
1403
1404    return detail
1405
1406
1407def _final_state_detail(result: ComparisonResult) -> dict[str, Any]:
1408    """The final-state comparison, for the run's JSON payload."""
1409    streams, omitted, dropped = _bounded_stream_diffs(
1410        result,
1411        omission_note="Too large for this payload; diff the state messages",
1412        artifact_hint="Both versions' state messages are in the saved artifacts.",
1413    )
1414
1415    detail: dict[str, Any] = {
1416        "passed": result.passed,
1417        # The two reasons a `passed: false` did not fail the run, kept apart so
1418        # the payload never says "every state that moved moved in value only"
1419        # about a read where nothing moved because nothing was compared.
1420        "value_only": result.value_only,
1421        "inconclusive": result.inconclusive,
1422        "message": result.message,
1423        "changed_streams": _bounded_changes(result.failed_streams),
1424        "streams": streams,
1425    }
1426    if omitted:
1427        detail["streams_with_omitted_diffs"] = omitted
1428    if dropped:
1429        detail["streams_omitted"] = dropped
1430
1431    return detail
1432
1433
1434def _bounded_stream_diffs(
1435    result: ComparisonResult,
1436    *,
1437    omission_note: str,
1438    artifact_hint: str,
1439) -> tuple[dict[str, Any], list[str], int]:
1440    """A comparison's per-stream verdicts and diffs, sized for the JSON payload.
1441
1442    Every stream that changed is named with its verdict -- so the payload says
1443    what moved, not just that something did -- while the diffs themselves are
1444    kept only while they fit in `_MAX_DIFF_PAYLOAD_CHARS`, and the number of
1445    entries is capped in turn: a thousand-stream source would otherwise put a
1446    thousand of them into `GITHUB_OUTPUT`. Both bounds say what they dropped,
1447    on the console and in the payload: a bound nobody mentions reads as "there
1448    was nothing more to see".
1449    """
1450    streams: dict[str, Any] = {}
1451    omitted: list[str] = []
1452    budget = _MAX_DIFF_PAYLOAD_CHARS
1453
1454    # Non-passing first, so the cap falls on the streams with nothing to say.
1455    ordered = sorted(result.stream_results.items(), key=lambda item: item[1].passed)
1456    kept, dropped_entries = (
1457        ordered[:_MAX_PAYLOAD_STREAMS],
1458        ordered[_MAX_PAYLOAD_STREAMS:],
1459    )
1460
1461    for name, stream in kept:
1462        entry: dict[str, Any] = {"passed": stream.passed, "message": stream.message}
1463        streams[name] = entry
1464
1465        if not stream.schema_diff:
1466            continue
1467
1468        rendered = len(json.dumps(stream.schema_diff, separators=(",", ":")))
1469        if rendered > budget:
1470            omitted.append(name)
1471            entry["diff_omitted"] = omission_note
1472            continue
1473
1474        budget -= rendered
1475        entry["diff"] = stream.schema_diff
1476
1477    if omitted:
1478        print_warning(
1479            f"Comparison detail for {len(omitted)} streams was too large for the "
1480            f"run's JSON output ({', '.join(omitted)}); their diffs are omitted "
1481            f"there. {artifact_hint}"
1482        )
1483
1484    if dropped_entries:
1485        print_warning(
1486            f"The run's JSON output lists {_MAX_PAYLOAD_STREAMS} of "
1487            f"{len(ordered)} streams, changed ones first. The report carries "
1488            f"them all. {artifact_hint}"
1489        )
1490
1491    return streams, omitted, len(dropped_entries)
1492
1493
1494def _build_connector_image_from_source(
1495    connector_name: str,
1496    repo_root: Path | None = None,
1497    tag: str = "dev",
1498    *,
1499    prefer_repo_root: bool = False,
1500) -> str | None:
1501    """Build a connector image from source code.
1502
1503    Args:
1504        connector_name: Name of the connector (e.g., 'source-pokeapi').
1505        repo_root: Optional path to the airbyte repo root. If not provided,
1506            will attempt to auto-detect from current directory.
1507        tag: Tag to apply to the built image (default: 'dev').
1508        prefer_repo_root: Use `repo_root` even when auto-discovery would find
1509            a connector directory. Reproduction cases set it: auto-discovery
1510            searches from the working directory and can pick up a different
1511            airbyte checkout on the machine, silently building the wrong source.
1512
1513    Returns:
1514        The full image name with tag if successful, None if build fails.
1515    """
1516    if not verify_docker_installation():
1517        print_error("Docker is not installed or not running")
1518        return None
1519
1520    connector_directory: Path | None = None
1521    if not (prefer_repo_root and repo_root):
1522        with suppress(FileNotFoundError):
1523            connector_directory = find_connector_root_from_name(connector_name)
1524    if connector_directory is None:
1525        if repo_root:
1526            connector_directory = repo_root / CONNECTORS_SUBDIR / connector_name
1527            if not connector_directory.exists():
1528                print_error(f"Connector directory not found: {connector_directory}")
1529                return None
1530        else:
1531            print_error(
1532                f"Could not find connector '{connector_name}'. "
1533                "Try providing --repo-root to specify the airbyte repo location."
1534            )
1535            return None
1536    metadata_file_path = connector_directory / "metadata.yaml"
1537    if not metadata_file_path.exists():
1538        print_error(f"metadata.yaml not found at {metadata_file_path}")
1539        return None
1540
1541    metadata = MetadataFile.from_file(metadata_file_path)
1542    print_success(f"Building image for connector: {connector_name}")
1543
1544    built_image = build_connector_image(
1545        connector_name=connector_name,
1546        connector_directory=connector_directory,
1547        metadata=metadata,
1548        tag=tag,
1549        no_verify=False,
1550    )
1551    print_success(f"Successfully built image: {built_image}")
1552    return built_image
1553
1554
1555def _fetch_control_image_from_metadata(connector_name: str) -> str | None:
1556    """Fetch the current released connector image from metadata.yaml on main branch.
1557
1558    This fetches the connector's metadata.yaml from the airbyte monorepo's master branch
1559    and extracts the dockerRepository and dockerImageTag to construct the control image.
1560
1561    Args:
1562        connector_name: The connector name (e.g., 'source-github').
1563
1564    Returns:
1565        The full connector image with tag (e.g., 'airbyte/source-github:1.0.0'),
1566        or None if the metadata could not be fetched or parsed.
1567    """
1568    metadata_url = (
1569        f"https://raw.githubusercontent.com/airbytehq/airbyte/master/"
1570        f"airbyte-integrations/connectors/{connector_name}/metadata.yaml"
1571    )
1572    response = requests.get(metadata_url, timeout=30)
1573    if not response.ok:
1574        print_error(
1575            f"Failed to fetch metadata for {connector_name}: "
1576            f"HTTP {response.status_code} from {metadata_url}"
1577        )
1578        return None
1579
1580    metadata = yaml.safe_load(response.text)
1581    if not isinstance(metadata, dict):
1582        print_error(f"Invalid metadata format for {connector_name}: expected dict")
1583        return None
1584
1585    data = metadata.get("data", {})
1586    docker_repository = data.get("dockerRepository")
1587    docker_image_tag = data.get("dockerImageTag")
1588
1589    if not docker_repository or not docker_image_tag:
1590        print_error(
1591            f"Could not find dockerRepository/dockerImageTag in metadata for {connector_name}"
1592        )
1593        return None
1594
1595    return f"{docker_repository}:{docker_image_tag}"
1596
1597
1598def _run_with_optional_http_metrics(
1599    connector_image: str,
1600    command: Command,
1601    output_dir: Path,
1602    target_or_control: TargetOrControl,
1603    enable_http_metrics: bool,
1604    config_path: Path | None,
1605    catalog_path: Path | None,
1606    state_path: Path | None,
1607    enable_debug_logs: bool = False,
1608    extra_env: dict[str, str] | None = None,
1609    replay_expected: bool = False,
1610    replay_from: Path | None = None,
1611    replay_options: ReplayOptions | None = None,
1612    stream_large_bodies: str = DEFAULT_STREAM_LARGE_BODIES,
1613) -> tuple[dict, ComparableOutputs]:
1614    """Run a connector command with optional HTTP metrics capture.
1615
1616    When enable_http_metrics is True, starts mitmproxy to capture HTTP traffic.
1617    If mitmproxy fails to start, falls back to running without metrics.
1618
1619    Args:
1620        connector_image: Full connector image name with tag.
1621        command: The Airbyte command to run.
1622        output_dir: Directory to store output files.
1623        target_or_control: Whether this is target or control version.
1624        enable_http_metrics: Whether to capture HTTP metrics via mitmproxy.
1625        config_path: Path to connector config JSON file.
1626        catalog_path: Path to configured catalog JSON file.
1627        state_path: Path to state JSON file.
1628        enable_debug_logs: If `True`, pass `LOG_LEVEL=DEBUG` to the connector container.
1629        extra_env: Additional environment variables for the connector container.
1630        replay_expected: Whether this run was *meant* to replay. Separate from
1631            `replay_from`, which is only non-None once a recording actually
1632            exists -- when the proxy is unavailable there is no recording to
1633            point at, and that is exactly the case that has to be reported.
1634        replay_from: A previous run's HTTP dump to serve responses from, so this
1635            run sees the same upstream data the recorded one did. Requires
1636            `enable_http_metrics`; ignored without it.
1637        replay_options: Replay tuning; defaults apply when omitted.
1638        stream_large_bodies: Body-size threshold above which mitmproxy streams
1639            rather than records a body.
1640
1641    Returns:
1642        What `_run_connector_command` returns, with `http_metrics` added to the
1643        results dict when they were captured -- plus `http_dump_file`, the dump
1644        this run recorded, which is what a subsequent run replays from.
1645    """
1646    if not enable_http_metrics:
1647        return _run_connector_command(
1648            connector_image=connector_image,
1649            command=command,
1650            output_dir=output_dir,
1651            target_or_control=target_or_control,
1652            config_path=config_path,
1653            catalog_path=catalog_path,
1654            state_path=state_path,
1655            enable_debug_logs=enable_debug_logs,
1656            extra_env=extra_env,
1657        )
1658
1659    # The manager rather than `MitmproxyManager.start`: the reason a startup
1660    # failed and the fact that a shutdown had to be killed are only reachable
1661    # through it -- there is no session to carry the first, and the second is
1662    # not known until the session is gone.
1663    manager = MitmproxyManager(
1664        output_dir,
1665        replay_from=replay_from,
1666        replay_options=replay_options,
1667        stream_large_bodies=stream_large_bodies,
1668    )
1669    with manager.running() as session:
1670        if session is None:
1671            startup_failure = (
1672                manager.startup_failure_reason or "mitmproxy is unavailable"
1673            )
1674            print_error(
1675                f"Mitmproxy unavailable ({startup_failure}), "
1676                "running without HTTP metrics"
1677            )
1678            result, comparable = _run_connector_command(
1679                connector_image=connector_image,
1680                command=command,
1681                output_dir=output_dir,
1682                target_or_control=target_or_control,
1683                config_path=config_path,
1684                catalog_path=catalog_path,
1685                state_path=state_path,
1686                enable_debug_logs=enable_debug_logs,
1687                extra_env=extra_env,
1688            )
1689            if replay_expected:
1690                # The one skip path that produces no session, and so reported
1691                # nothing at all: no metrics, no reason, an ordinary green
1692                # comparison against live data. Keyed on `replay_expected` and
1693                # not on `replay_from`: when mitmdump is missing the *control*
1694                # run records no dump either, so there is nothing to point at
1695                # and the condition that looks natural here is never true in
1696                # the failure it defends against.
1697                #
1698                # The precise reason, not a fixed "mitmproxy is unavailable":
1699                # `wait_for_proxy_listener` distinguishes a proxy that died on
1700                # startup and a corpus that took longer than its ceiling to
1701                # load, and those point at a code fix where the hardcoded
1702                # string tells a reader to install mitmproxy.
1703                result["http_metrics"] = {
1704                    "replay_skipped_reason": (
1705                        f"{startup_failure}, so nothing could be replayed"
1706                    )
1707                }
1708            return result, comparable
1709
1710        if session.replay_source:
1711            print_success(
1712                f"Started mitmproxy on {session.proxy_url} "
1713                f"(replaying {session.replay_source}, "
1714                f"{session.unreplayable_flow_count} flow(s) not replayable)"
1715            )
1716        else:
1717            print_success(f"Started mitmproxy on {session.proxy_url}")
1718            if session.replay_skipped_reason:
1719                print_warning(
1720                    f"HTTP replay unavailable ({session.replay_skipped_reason}); "
1721                    "this run hits the live API, so it cannot be compared against "
1722                    "identical upstream data"
1723                )
1724
1725        result, comparable = _run_connector_command(
1726            connector_image=connector_image,
1727            command=command,
1728            output_dir=output_dir,
1729            target_or_control=target_or_control,
1730            config_path=config_path,
1731            catalog_path=catalog_path,
1732            state_path=state_path,
1733            proxy_url=session.proxy_url,
1734            ca_cert_path=session.ca_cert_path,
1735            enable_debug_logs=enable_debug_logs,
1736            extra_env=extra_env,
1737        )
1738
1739        # The path, not the metrics: this is what the next run replays from.
1740        result["http_dump_file"] = str(session.dump_file_path)
1741
1742        # Everything the metrics are built from, carried out of the block. The
1743        # dump itself cannot be read here: mitmproxy's `save` addon writes each
1744        # flow through a block-buffered file and only flushes and closes it in
1745        # `done()`, holding in-flight flows until shutdown -- so parsing inside
1746        # the `with` misses the buffered tail, and `live_flow_count` is the
1747        # feature's acceptance criterion, not a rough count.
1748        dump_file_path = session.dump_file_path
1749        replay_source = session.replay_source
1750        replay_skipped_reason = session.replay_skipped_reason
1751        unreplayable_flow_count = session.unreplayable_flow_count
1752
1753    http_metrics = parse_http_dump(dump_file_path, replay_source)
1754    metrics_payload: dict[str, object] = {
1755        "flow_count": http_metrics.flow_count,
1756        "duplicate_flow_count": http_metrics.duplicate_flow_count,
1757        # Carried into the report because it is the number that tells a
1758        # reviewer whether a record difference is a real regression or
1759        # upstream drift. Today it is a duplicate-URL heuristic; it only
1760        # becomes exact once HTTP replay lands.
1761        "cache_hits_count": http_metrics.cache_hits_count,
1762        "cache_hit_ratio": http_metrics.cache_hit_ratio,
1763        # On a replaying target run, `live_flow_count` is the acceptance
1764        # criterion for replay -- so it belongs in the report rather than
1765        # only in the raw dump.
1766        "replayed_flow_count": http_metrics.replayed_flow_count,
1767        "live_flow_count": http_metrics.live_flow_count,
1768        "replay_hit_ratio": http_metrics.replay_hit_ratio,
1769    }
1770    if replay_from is not None:
1771        # Keyed on replay having been *asked* for, not on a corpus having
1772        # survived: when every recorded flow is unreplayable there is no
1773        # corpus, and that is exactly the run whose dropped count matters.
1774        # Poor replay coverage is usually this, and the fix is a bigger
1775        # `--http-replay-max-body`.
1776        metrics_payload["unreplayable_flow_count"] = unreplayable_flow_count
1777    if replay_source:
1778        metrics_payload["replay_source"] = str(replay_source)
1779        # Only on a replaying run, where a live request is a request the corpus
1780        # could not answer and so the one thing that explains a coverage
1781        # shortfall. On a recording run every request is live, so the same list
1782        # would be a request log, which is not a surface to widen for no
1783        # diagnostic gain. Query-parameter values are masked either way -- these
1784        # URLs reach the step summary, the workflow log and `report.html`, and
1785        # plenty of APIs put credentials in the query string; see `redact_url`.
1786        live_urls = http_metrics.top_live_urls()
1787        if live_urls:
1788            metrics_payload["live_urls"] = live_urls
1789            metrics_payload["live_unique_url_count"] = len(http_metrics.live_url_counts)
1790    if manager.metrics_incomplete_reason:
1791        # Read after the block, not inside it: a killed shutdown is only known
1792        # once `_stop` has run. Recorded because every count below is then a
1793        # lower bound -- and short in the direction that makes a run look closer
1794        # to the replay acceptance criterion than it was.
1795        metrics_payload["metrics_incomplete"] = manager.metrics_incomplete_reason
1796    skipped_reason = replay_skipped_reason
1797    if replay_expected and replay_source is None and not skipped_reason:
1798        # The proxy came up but there was no recording to replay -- the
1799        # control run captured nothing. Same outcome as any other skip, so
1800        # it has to read the same way rather than as a clean replayed run.
1801        skipped_reason = "the control run recorded no HTTP dump to replay from"
1802    if skipped_reason:
1803        metrics_payload["replay_skipped_reason"] = skipped_reason
1804    result["http_metrics"] = metrics_payload
1805
1806    print_success(
1807        f"Captured {http_metrics.flow_count} HTTP flows "
1808        f"({http_metrics.duplicate_flow_count} duplicates)"
1809    )
1810    if replay_source:
1811        print_success(
1812            f"Replayed {http_metrics.replayed_flow_count} of "
1813            f"{http_metrics.flow_count} request(s) "
1814            f"({http_metrics.replay_hit_ratio}); "
1815            f"{http_metrics.live_flow_count} went live"
1816        )
1817    return result, comparable
1818
1819
1820def _discard_oversized_http_dumps(*results: dict, max_bytes: int) -> None:
1821    """Drop recorded dumps too large to keep, and say so in the report.
1822
1823    "Too large to keep" is about the runner's disk, not the artifact upload:
1824    dumps are excluded from every upload step regardless of size.
1825
1826    Deleting is not silent: the size that was discarded lands in the run's
1827    `http_metrics`, so a reader looking for the dump learns why it is not there
1828    instead of finding a dangling path.
1829    """
1830    for result in results:
1831        dump_file = result.get("http_dump_file")
1832        if not dump_file:
1833            continue
1834
1835        discarded = discard_oversized_dump(Path(dump_file), max_bytes)
1836        if discarded is None:
1837            continue
1838
1839        result.pop("http_dump_file", None)
1840        metrics = result.get("http_metrics")
1841        if isinstance(metrics, dict):
1842            metrics["discarded_dump_mb"] = round(discarded / 1024 / 1024)
1843        print_warning(
1844            f"Discarded {dump_file}: {discarded / 1024 / 1024:.0f} MB exceeds the "
1845            "retention cap, and a dump this large is what fills the runner's disk. "
1846            "Dumps are never uploaded as artifacts; raise "
1847            "--http-replay-max-dump-size-mb to keep it for a local run"
1848        )
1849
1850
1851def _compare_read_records(
1852    target_result: dict,
1853    control_result: dict,
1854    catalog_file: Path | None,
1855) -> tuple[RecordComparisonSummary | None, str | None]:
1856    """Run record-level comparisons between control and target read outputs.
1857
1858    Uses primary keys from the configured catalog (source-defined primary
1859    keys come from the discovered catalog) to compare record counts, PK
1860    presence, PK integrity, and field values. Streams without a primary key
1861    fall back to a directional record count comparison.
1862
1863    Each run's stdout is first split into per-stream record files (a single
1864    streaming pass), and the comparison then loads one stream at a time, so
1865    memory stays bounded by the largest stream instead of the full output
1866    of both runs. The per-stream files land next to each run's stdout and
1867    double as debugging artifacts. Within one stream, both sides are
1868    materialized as parsed messages plus dict copies plus PK indexes --
1869    measured at roughly 13 bytes resident per byte of record JSON, so a
1870    single stream of ~500 MB of records is the practical ceiling on a
1871    standard 7 GB runner. Start here when debugging an OOM.
1872
1873    Returns:
1874        `(summary, skip_reason)`. The summary is None when the comparison
1875        cannot run at all -- then `skip_reason` says why, so the caller can
1876        surface a diagnostic check instead of letting the run read as a pass
1877        that compared records. An unexpected error during the comparison
1878        itself returns a FAILED summary instead of raising: by this point
1879        both (expensive, live-API) runs have completed, and a comparison bug
1880        must fail the verdict, not destroy the run's outputs and reports.
1881    """
1882    if catalog_file is None or not catalog_file.exists():
1883        reason = "no configured catalog available"
1884        print_warning(f"Skipping record-level comparison: {reason}")
1885        return None, reason
1886
1887    try:
1888        configured_catalog = ConfiguredAirbyteCatalog.model_validate_json(
1889            catalog_file.read_text()
1890        )
1891    except Exception as exc:
1892        reason = f"configured catalog could not be parsed ({exc})"
1893        print_warning(f"Skipping record-level comparison: {reason}")
1894        return None, reason
1895
1896    try:
1897        primary_keys = get_primary_keys_per_stream(configured_catalog)
1898        # A stream name that repeats across namespaces merges two streams in
1899        # this name-keyed pipeline; those names degraded to count-only above,
1900        # and the reader must hear why from the summary, not just stdout.
1901        collision_warnings = [
1902            f"Stream name {name!r} appears in more than one namespace; the "
1903            "record comparison keys streams by name, so its records are "
1904            "compared by count only"
1905            for name in duplicate_stream_names(configured_catalog)
1906        ]
1907        control_record_files = split_records_per_stream(
1908            Path(control_result["stdout_file"])
1909        )
1910        target_record_files = split_records_per_stream(
1911            Path(target_result["stdout_file"])
1912        )
1913
1914        summary = run_record_comparisons_from_files(
1915            control_record_files=control_record_files,
1916            target_record_files=target_record_files,
1917            primary_keys_per_stream=primary_keys,
1918        )
1919        summary.warnings[:0] = collision_warnings
1920    except Exception as exc:
1921        summary = RecordComparisonSummary(
1922            passed=False,
1923            errored=True,
1924            count_comparison=ComparisonResult(
1925                passed=False, message="Record comparison errored"
1926            ),
1927            errors=[f"Record comparison errored: {exc!r}"],
1928        )
1929
1930    if summary.passed:
1931        print_success("Record comparison passed (counts, PKs, field values)")
1932    else:
1933        print_error("Record comparison failed:")
1934        for error in summary.errors[:20]:
1935            print_error(f"  - {error}")
1936    for warning in summary.warnings[:20]:
1937        print_warning(warning)
1938
1939    return summary, None
1940
1941
1942@connector_app.command(name="regression-test")
1943def regression_test(
1944    skip_compare: Annotated[
1945        bool,
1946        Parameter(
1947            help="If True, skip comparison and run single-version tests only. "
1948            "If False (default), run comparison tests (target vs control)."
1949        ),
1950    ] = False,
1951    skip_record_comparison: Annotated[
1952        bool,
1953        Parameter(
1954            help="If True, skip the record-level comparison for read commands "
1955            "(counts, PK presence, PK integrity, field values) and gate the "
1956            "verdict on command outcomes only. Escape hatch for sources whose "
1957            "data legitimately changes between the two live runs (rolling "
1958            "windows, feeds that age out records) until HTTP request caching "
1959            "makes strict comparison reliable."
1960        ),
1961    ] = False,
1962    test_image: Annotated[
1963        str | None,
1964        Parameter(
1965            help="Test connector image with tag (e.g., airbyte/source-github:1.0.0). "
1966            "This is the image under test - in comparison mode, it's compared against control_image."
1967        ),
1968    ] = None,
1969    control_image: Annotated[
1970        str | None,
1971        Parameter(
1972            help="Control connector image (baseline version) with tag (e.g., airbyte/source-github:1.0.0). "
1973            "Ignored if `skip_compare=True`."
1974        ),
1975    ] = None,
1976    connector_name: Annotated[
1977        str | None,
1978        Parameter(
1979            help="Connector name to build image from source (e.g., 'source-pokeapi'). "
1980            "If provided, builds the image locally with tag 'dev'. "
1981            "For comparison tests (default), this builds the target image. "
1982            "For single-version tests (skip_compare=True), this builds the test image."
1983        ),
1984    ] = None,
1985    repo_root: Annotated[
1986        str | None,
1987        Parameter(
1988            help="Path to the airbyte repo root. Required if connector_name is provided "
1989            "and the repo cannot be auto-detected."
1990        ),
1991    ] = None,
1992    command: Annotated[
1993        Literal["spec", "check", "discover", "read"],
1994        Parameter(help="The Airbyte command to run."),
1995    ] = "check",
1996    connection_id: Annotated[
1997        str | None,
1998        Parameter(
1999            help="Airbyte Cloud connection ID to fetch config/catalog from. "
2000            "Mutually exclusive with config-path/catalog-path. "
2001            "If provided, test_image/control_image can be auto-detected."
2002        ),
2003    ] = None,
2004    config_path: Annotated[
2005        str | None,
2006        Parameter(help="Path to the connector config JSON file."),
2007    ] = None,
2008    catalog_path: Annotated[
2009        str | None,
2010        Parameter(help="Path to the configured catalog JSON file (required for read)."),
2011    ] = None,
2012    state_path: Annotated[
2013        str | None,
2014        Parameter(help="Path to the state JSON file (optional for read)."),
2015    ] = None,
2016    output_dir: Annotated[
2017        str,
2018        Parameter(help="Directory to store test artifacts."),
2019    ] = "/tmp/regression_test_artifacts",
2020    enable_http_metrics: Annotated[
2021        bool,
2022        Parameter(
2023            help="Capture HTTP traffic metrics via mitmproxy. Requires mitmdump to "
2024            "be installed; without it the run proceeds without metrics. Only used "
2025            "in comparison mode, where HTTP replay turns it on anyway -- set this "
2026            "to record traffic while running both versions live "
2027            "(--no-http-replay)."
2028        ),
2029    ] = False,
2030    enable_http_replay: Annotated[
2031        bool | None,
2032        Parameter(
2033            negative="--no-http-replay",
2034            help="Replay the control run's recorded HTTP responses for the target "
2035            "run, so both versions see identical upstream data. Implies "
2036            "--enable-http-metrics. Defaults to `True` in comparison mode; has no "
2037            "effect with `--skip-compare`, which has no control run to replay "
2038            "from. Pass --no-http-replay to run both versions against the live "
2039            "API, which is the fallback whenever replay cannot be delivered "
2040            "anyway.",
2041        ),
2042    ] = None,
2043    http_replay_max_body: Annotated[
2044        str,
2045        Parameter(
2046            help="Response bodies larger than this are streamed rather than "
2047            "recorded, and therefore re-fetched live instead of replayed. Raise it "
2048            "for a connector with large pages, at the cost of replay memory. "
2049            "Accepts a mitmproxy size such as '1m' or '10m' -- never a bare number, "
2050            "which means bytes. Unlike the other --http-replay-* options this one "
2051            "governs recording, so it applies with --enable-http-metrics alone."
2052        ),
2053    ] = DEFAULT_STREAM_LARGE_BODIES,
2054    http_replay_max_dump_size_mb: Annotated[
2055        int,
2056        Parameter(
2057            help="Skip replay and run the target live when the filtered replay "
2058            "corpus is larger than this, and discard a recorded dump this large "
2059            "once the run is done with it. mitmproxy holds the whole corpus in "
2060            "memory for the entire run, and a dump grows with everything the "
2061            "connector downloads, so this is what keeps a big connector from "
2062            "exhausting the CI runner's memory and disk. Dumps are never "
2063            "uploaded as artifacts, so on a local run -- where the dump is the "
2064            "only thing that can diagnose a coverage shortfall offline -- raise "
2065            "it to keep the dump."
2066        ),
2067    ] = DEFAULT_MAX_REPLAY_DUMP_SIZE_MB,
2068    http_replay_reuse: Annotated[
2069        bool,
2070        Parameter(
2071            help="Serve a recorded response more than once. By default each "
2072            "recorded flow is consumed on first match, which preserves recorded "
2073            "ordering and lets memory fall as the run proceeds; reuse keeps the "
2074            "whole corpus resident for the entire run."
2075        ),
2076    ] = False,
2077    http_replay_ignore_params: Annotated[
2078        str | None,
2079        Parameter(
2080            help="Comma-separated query parameters to exclude from the replay match "
2081            "key, for connectors that put a timestamp or nonce in the URL."
2082        ),
2083    ] = None,
2084    strict_replay: Annotated[
2085        bool,
2086        Parameter(
2087            help="Kill requests that match nothing in the replay corpus instead of "
2088            "forwarding them upstream. Diagnostic only: it turns unmatched requests "
2089            "into connector errors, which is how you count them. mitmproxy stops "
2090            "intervening once its flowmap is empty, so combine this with "
2091            "--http-replay-reuse to cover the whole run rather than only the part "
2092            "before the corpus is consumed."
2093        ),
2094    ] = False,
2095    selected_streams: Annotated[
2096        str | None,
2097        Parameter(
2098            help="Comma-separated list of stream names to include in the read. "
2099            "Only these streams will be included in the configured catalog. "
2100            "This is useful to limit data volume by testing only specific streams."
2101        ),
2102    ] = None,
2103    enable_debug_logs: Annotated[
2104        bool,
2105        Parameter(
2106            help="Enable debug-level logging for regression test output. "
2107            "Also passed as `LOG_LEVEL=DEBUG` to the connector Docker container."
2108        ),
2109    ] = False,
2110    extra_env: Annotated[
2111        list[str] | None,
2112        Parameter(
2113            help=(
2114                "Additional KEY=VALUE environment variables for the connector "
2115                "containers. In comparison mode, applies to both target and "
2116                "control. Values are visible in the debug log of the `docker run` "
2117                "command; do not use for secrets."
2118            )
2119        ),
2120    ] = None,
2121    with_state: Annotated[
2122        bool | None,
2123        Parameter(
2124            negative="--no-state",
2125            help="Fetch and pass the connection's current state to the read command, "
2126            "producing a warm read instead of a cold read. Defaults to `True` when "
2127            "`--connection-id` is provided, `False` otherwise. Has no effect unless "
2128            "the command is `read`. Ignored when `--state-path` is explicitly provided.",
2129        ),
2130    ] = None,
2131    repro_case_dir: Annotated[
2132        str | None,
2133        Parameter(
2134            help="Reproduction case directory (see `regression-case-setup`). Its "
2135            "catalog.json, state.json and expectations.json replace --catalog-path "
2136            "and --state-path, HTTP replay is turned off, and a read passes only "
2137            "when the control violates the case's expectations and the target "
2138            "satisfies them (see --repro-goal). Writes the verdict to "
2139            "`result.json` in the output directory. Requires `--command read`, "
2140            "--config-path, --sandbox-scenario and comparison mode; run the "
2141            "command under `sandbox exec`."
2142        ),
2143    ] = None,
2144    repro_goal: Annotated[
2145        Literal["fixed", "reproduced"] | None,
2146        Parameter(
2147            help="What a case read must show. `fixed` (default): the control "
2148            "violates the case and the target satisfies it. `reproduced`: both "
2149            "violate it -- the case gate's check, with the unmodified source of "
2150            "the affected release as the target. Requires --repro-case-dir."
2151        ),
2152    ] = None,
2153    sandbox_scenario: Annotated[
2154        str | None,
2155        Parameter(
2156            help="SaaS Sandbox scenario UUID the case was set up in. For a read, "
2157            "its baseline snapshot is restored immediately before each side runs, "
2158            "so control and target start from identical data and failure rules."
2159        ),
2160    ] = None,
2161) -> None:
2162    """Run regression tests on connectors.
2163
2164    This command supports two modes:
2165
2166    Comparison mode (skip_compare=False, default):
2167        Runs the specified Airbyte protocol command against both the target (new)
2168        and control (baseline) connector versions, then compares the results.
2169        This helps identify regressions between versions.
2170
2171    Single-version mode (skip_compare=True):
2172        Runs the specified Airbyte protocol command against a single connector
2173        and validates the output. No comparison is performed.
2174
2175    Results are written to the output directory and to GitHub Actions outputs
2176    if running in CI.
2177
2178    You can provide the test image in three ways:
2179    1. --test-image: Use a pre-built image from Docker registry
2180    2. --connector-name: Build the image locally from source code
2181    3. --connection-id: Auto-detect from an Airbyte Cloud connection
2182
2183    You can provide config/catalog either via file paths OR via a connection_id
2184    that fetches them from Airbyte Cloud.
2185    """
2186    parsed_extra_env = _parse_extra_env(extra_env)
2187
2188    # Configure debug logging for the regression test harness when requested
2189    if enable_debug_logs:
2190        logging.basicConfig(
2191            level=logging.DEBUG,
2192            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
2193            force=True,
2194        )
2195
2196    output_path = Path(output_dir)
2197    output_path.mkdir(parents=True, exist_ok=True)
2198
2199    cmd = Command(command)
2200
2201    repro_case: ReproCase | None = None
2202    sandbox: SandboxCli | None = None
2203    if repro_case_dir or sandbox_scenario or repro_goal:
2204        case_error = _repro_case_argument_error(
2205            command=cmd,
2206            repro_case_dir=repro_case_dir,
2207            sandbox_scenario=sandbox_scenario,
2208            skip_compare=skip_compare,
2209            connection_id=connection_id,
2210            config_path=config_path,
2211            catalog_path=catalog_path,
2212            state_path=state_path,
2213            selected_streams=selected_streams,
2214        )
2215        if case_error is None and repro_case_dir:
2216            try:
2217                repro_case = ReproCase.load(Path(repro_case_dir))
2218                sandbox = SandboxCli()
2219            except (OSError, ValueError, SandboxError) as exc:
2220                case_error = f"Invalid reproduction case: {exc}"
2221        if case_error is None and repro_case is not None:
2222            case_error = _repro_case_command_error(cmd, repro_case.expectations.command)
2223        if case_error is not None:
2224            write_github_output("success", False)
2225            write_github_output("error", case_error)
2226            exit_with_error(case_error)
2227        assert repro_case is not None
2228        if repro_case.expectations.command == "read":
2229            catalog_path = str(repro_case.catalog_path)
2230            state_path = str(repro_case.state_path) if repro_case.state_path else None
2231        # The sandbox is the controlled upstream; replaying one side's traffic
2232        # to the other would bypass the per-side baseline restore.
2233        if enable_http_replay or enable_http_metrics:
2234            print_warning("HTTP replay and metrics are off for reproduction cases")
2235        enable_http_replay = False
2236        enable_http_metrics = False
2237
2238    # On by default, but only where it can work: single-version mode has no
2239    # control run to replay from. Asking for it there explicitly is a mistake
2240    # worth naming; the default landing there is not, so it is not a warning.
2241    if skip_compare:
2242        if enable_http_replay:
2243            print_warning("--enable-http-replay needs a control run; ignoring it")
2244        replay_enabled = False
2245    else:
2246        replay_enabled = True if enable_http_replay is None else enable_http_replay
2247
2248    # Replay is a use of the capture, not a separate mechanism, so it turns the
2249    # capture on rather than erroring.
2250    if replay_enabled:
2251        enable_http_metrics = True
2252    elif (
2253        http_replay_reuse
2254        or strict_replay
2255        or http_replay_ignore_params
2256        or http_replay_max_dump_size_mb != DEFAULT_MAX_REPLAY_DUMP_SIZE_MB
2257    ):
2258        # Otherwise someone debugging poor coverage with --strict-replay alone
2259        # concludes strict mode found nothing wrong. `--http-replay-max-body` is
2260        # deliberately not in this list: it governs recording and applies anyway.
2261        print_warning(
2262            "Replay tuning flags do nothing with --no-http-replay; ignoring them"
2263        )
2264
2265    replay_options = ReplayOptions(
2266        reuse=http_replay_reuse,
2267        extra="kill" if strict_replay else DEFAULT_REPLAY_EXTRA,
2268        ignore_params=tuple(
2269            param.strip()
2270            for param in (http_replay_ignore_params or "").split(",")
2271            if param.strip()
2272        ),
2273        max_dump_bytes=http_replay_max_dump_size_mb * 1024 * 1024,
2274    )
2275
2276    config_file: Path | None = None
2277    catalog_file: Path | None = None
2278    state_file = Path(state_path) if state_path else None
2279
2280    selected_stream_names: set[str] = (
2281        {s.strip() for s in selected_streams.split(",") if s.strip()}
2282        if selected_streams
2283        else set()
2284    )
2285
2286    # Resolve the test image (used in both single-version and comparison modes)
2287    resolved_test_image: str | None = test_image
2288    resolved_control_image: str | None = control_image
2289
2290    # Validate conflicting parameters
2291    # Single-version mode: reject comparison-specific parameters
2292    if skip_compare and control_image:
2293        write_github_output("success", False)
2294        write_github_output(
2295            "error", "Cannot specify control_image with skip_compare=True"
2296        )
2297        exit_with_error(
2298            "Cannot specify --control-image with --skip-compare. "
2299            "Control image is only used in comparison mode."
2300        )
2301
2302    # If connector_name is provided, build the image from source
2303    if connector_name:
2304        if resolved_test_image:
2305            write_github_output("success", False)
2306            write_github_output(
2307                "error", "Cannot specify both test_image and connector_name"
2308            )
2309            exit_with_error("Cannot specify both --test-image and --connector-name")
2310
2311        repo_root_path = Path(repo_root) if repo_root else None
2312        built_image = _build_connector_image_from_source(
2313            connector_name=connector_name,
2314            repo_root=repo_root_path,
2315            tag="dev",
2316            prefer_repo_root=repro_case is not None,
2317        )
2318        if not built_image:
2319            write_github_output("success", False)
2320            write_github_output("error", f"Failed to build image for {connector_name}")
2321            exit_with_error(f"Failed to build image for {connector_name}")
2322        resolved_test_image = built_image
2323
2324    if connection_id:
2325        if config_path or catalog_path:
2326            write_github_output("success", False)
2327            write_github_output(
2328                "error", "Cannot specify both connection_id and file paths"
2329            )
2330            exit_with_error(
2331                "Cannot specify both connection_id and config_path/catalog_path"
2332            )
2333
2334        print_success(f"Fetching config/catalog from connection: {connection_id}")
2335        connection_data = fetch_connection_data(connection_id)
2336
2337        # Check if we should retrieve unmasked secrets
2338        if should_use_secret_retriever():
2339            print_success(
2340                "USE_CONNECTION_SECRET_RETRIEVER enabled - enriching config with unmasked secrets..."
2341            )
2342            try:
2343                connection_data = enrich_config_with_secrets(
2344                    connection_data,
2345                    retrieval_reason="Regression test with USE_CONNECTION_SECRET_RETRIEVER=true",
2346                )
2347                print_success("Successfully retrieved unmasked secrets from database")
2348            except SecretRetrievalError as e:
2349                write_github_output("success", False)
2350                write_github_output("error", str(e))
2351                exit_with_error(
2352                    f"{e}\n\n"
2353                    f"This connection cannot be used for regression testing. "
2354                    f"Please use a connection from a non-EU workspace, or use GSM-based "
2355                    f"integration test credentials instead (by omitting --connection-id)."
2356                )
2357
2358        # Fill in stream schemas: the minimal catalog from the public API has
2359        # empty json_schema values, which silently yields zero records for
2360        # connectors that derive request fields from the catalog schema.
2361        # Only the read command consumes the catalog, so spec/check/discover
2362        # skip the extra Cloud API round trip.
2363        enrichment_state: list[dict[str, Any]] | None = None
2364        enrichment_state_fetched = False
2365        if cmd.needs_catalog():
2366            merged_count = 0
2367            try:
2368                enrichment = enrich_catalog_schemas_from_cloud(connection_data)
2369                connection_data = enrichment.connection_data
2370                merged_count = enrichment.merged_count
2371                enrichment_state = enrichment.state
2372                enrichment_state_fetched = True
2373            except Exception as exc:
2374                print_error(f"Failed to fetch stream schemas from Cloud API: {exc}")
2375            missing_schemas = streams_missing_schemas(connection_data.catalog)
2376            if selected_stream_names:
2377                missing_schemas = [
2378                    s for s in missing_schemas if s in selected_stream_names
2379                ]
2380            if missing_schemas and should_use_secret_retriever():
2381                try:
2382                    connection_data, db_merged_count = (
2383                        enrich_catalog_schemas_from_platform_db(
2384                            connection_data,
2385                            retrieval_reason="Regression test catalog schema enrichment",
2386                        )
2387                    )
2388                    merged_count += db_merged_count
2389                except Exception as exc:
2390                    print_error(f"Failed to merge stored stream schemas: {exc}")
2391                missing_schemas = streams_missing_schemas(connection_data.catalog)
2392                if selected_stream_names:
2393                    missing_schemas = [
2394                        s for s in missing_schemas if s in selected_stream_names
2395                    ]
2396            total_streams = len(connection_data.catalog.get("streams", []))
2397            checked_streams = (
2398                f"{len(missing_schemas)} of {len(selected_stream_names)} "
2399                "selected stream(s)"
2400                if selected_stream_names
2401                else f"{len(missing_schemas)} of {total_streams} stream(s)"
2402            )
2403            if not total_streams:
2404                error_msg = (
2405                    "No configured streams found in the connection catalog. "
2406                    "A read with zero streams would produce zero records on "
2407                    "both versions, so failing instead of producing a false pass."
2408                )
2409                write_github_output("success", False)
2410                write_github_output("error", error_msg)
2411                exit_with_error(error_msg)
2412            if missing_schemas:
2413                error_msg = (
2414                    f"Stream schemas missing for {checked_streams}: "
2415                    f"{', '.join(missing_schemas)}. "
2416                    "Connectors that derive request fields from the catalog "
2417                    "schema would silently return zero records, so failing "
2418                    "instead of producing a false pass."
2419                )
2420                write_github_output("success", False)
2421                write_github_output("error", error_msg)
2422                exit_with_error(error_msg)
2423            print_success(
2424                f"Stream schemas ready: merged {merged_count} of {total_streams} "
2425                "stream(s)"
2426                + (
2427                    f"; checked {len(selected_stream_names)} selected stream(s)"
2428                    if selected_stream_names
2429                    else ""
2430                )
2431            )
2432
2433        # Resolve with_state default: True when connection_id is provided
2434        resolved_with_state = with_state if with_state is not None else True
2435
2436        if not (resolved_with_state and not state_file and command == "read"):
2437            connection_data.state = None
2438        elif enrichment_state_fetched:
2439            connection_data.state = enrichment_state
2440            if connection_data.state:
2441                print_success(
2442                    f"Using state fetched during schema enrichment "
2443                    f"({len(connection_data.state)} stream(s))"
2444                )
2445            else:
2446                print_success("No state available for this connection (initial sync)")
2447        else:
2448            print_success("Fetching connection state for warm read...")
2449            try:
2450                connection_data.state = fetch_connection_state(connection_data)
2451                if connection_data.state:
2452                    print_success(
2453                        f"Fetched state with {len(connection_data.state)} stream(s)"
2454                    )
2455                else:
2456                    connection_data.state = None
2457                    print_success(
2458                        "No state available for this connection (initial sync)"
2459                    )
2460            except Exception as exc:
2461                print_error(f"Failed to fetch connection state: {exc}")
2462                print_success("Falling back to cold read (no state)")
2463
2464        config_file, catalog_file, state_file_from_conn = save_connection_data_to_files(
2465            connection_data, output_path / "connection_data"
2466        )
2467
2468        if state_file_from_conn and not state_file:
2469            state_file = state_file_from_conn
2470
2471        print_success(
2472            f"Fetched config for source: {connection_data.source_name} "
2473            f"with {len(connection_data.stream_names)} streams"
2474        )
2475
2476        # Auto-detect test/control image from connection if not provided
2477        if not resolved_test_image and connection_data.connector_image:
2478            resolved_test_image = connection_data.connector_image
2479            print_success(f"Auto-detected test image: {resolved_test_image}")
2480
2481        if (
2482            not skip_compare
2483            and not resolved_control_image
2484            and connection_data.connector_image
2485        ):
2486            resolved_control_image = connection_data.connector_image
2487            print_success(f"Auto-detected control image: {resolved_control_image}")
2488    elif config_path:
2489        config_file = Path(config_path)
2490        catalog_file = Path(catalog_path) if catalog_path else None
2491    elif connector_name:
2492        # Fallback: fetch integration test secrets from GSM using PyAirbyte API
2493        print_success(
2494            f"No connection_id or config_path provided. "
2495            f"Attempting to fetch integration test config from GSM for {connector_name}..."
2496        )
2497        gsm_config = get_first_config_from_secrets(connector_name)
2498        if gsm_config:
2499            # Write config to a temp file (not in output_path to avoid artifact upload)
2500            gsm_config_dir = Path(
2501                tempfile.mkdtemp(prefix=f"gsm-config-{connector_name}-")
2502            )
2503            gsm_config_dir.chmod(0o700)
2504            gsm_config_file = gsm_config_dir / "config.json"
2505            gsm_config_file.write_text(json.dumps(gsm_config, indent=2))
2506            gsm_config_file.chmod(0o600)
2507            config_file = gsm_config_file
2508            # Use catalog_path if provided (e.g., generated from discover output)
2509            catalog_file = Path(catalog_path) if catalog_path else None
2510            print_success(
2511                f"Fetched integration test config from GSM for {connector_name}"
2512            )
2513        else:
2514            print_error(
2515                f"Failed to fetch integration test config from GSM for {connector_name}."
2516            )
2517            config_file = None
2518            # Use catalog_path if provided (e.g., generated from discover output)
2519            catalog_file = Path(catalog_path) if catalog_path else None
2520    else:
2521        config_file = None
2522        catalog_file = Path(catalog_path) if catalog_path else None
2523
2524    # Auto-detect control_image from metadata.yaml if connector_name is provided (comparison mode only)
2525    if not skip_compare and not resolved_control_image and connector_name:
2526        resolved_control_image = _fetch_control_image_from_metadata(connector_name)
2527        if resolved_control_image:
2528            print_success(
2529                f"Auto-detected control image from metadata.yaml: {resolved_control_image}"
2530            )
2531
2532    # Validate that we have the required images
2533    if not resolved_test_image:
2534        write_github_output("success", False)
2535        write_github_output("error", "No test image specified")
2536        exit_with_error(
2537            "You must provide one of the following: a test_image, a connector_name "
2538            "to build the image from source, or a connection_id to auto-detect the image."
2539        )
2540
2541    if not skip_compare and not resolved_control_image:
2542        write_github_output("success", False)
2543        write_github_output("error", "No control image specified")
2544        exit_with_error(
2545            "You must provide one of the following: a control_image, a connection_id "
2546            "for a connection that has an associated connector image, or a connector_name "
2547            "to auto-detect the control image from the airbyte repo's metadata.yaml."
2548        )
2549
2550    # Pull images if they weren't just built locally
2551    # If connector_name was provided, we just built the test image locally
2552    if not connector_name and not ensure_image_available(resolved_test_image):
2553        write_github_output("success", False)
2554        write_github_output("error", f"Failed to pull image: {resolved_test_image}")
2555        exit_with_error(f"Failed to pull test image: {resolved_test_image}")
2556
2557    if (
2558        not skip_compare
2559        and resolved_control_image
2560        and not ensure_image_available(resolved_control_image)
2561    ):
2562        write_github_output("success", False)
2563        write_github_output("error", f"Failed to pull image: {resolved_control_image}")
2564        exit_with_error(
2565            f"Failed to pull control connector image: {resolved_control_image}"
2566        )
2567
2568    # Apply selected_streams filter to catalog if requested
2569    if selected_stream_names and catalog_file:
2570        print_success(
2571            f"Filtering catalog to {len(selected_stream_names)} selected streams: "
2572            f"{', '.join(sorted(selected_stream_names))}"
2573        )
2574        filter_configured_catalog_file(catalog_file, selected_stream_names)
2575
2576    # Upgrade READ → READ_WITH_STATE when a state file is available
2577    if cmd == Command.READ and state_file:
2578        cmd = Command.READ_WITH_STATE
2579        print_success("State available — using read-with-state (warm read)")
2580
2581    # Track telemetry for the regression test
2582    # Extract version from image tag (e.g., "airbyte/source-github:1.0.0" -> "1.0.0")
2583    target_version = (
2584        resolved_test_image.rsplit(":", 1)[-1]
2585        if ":" in resolved_test_image
2586        else "unknown"
2587    )
2588    control_version = None
2589    if resolved_control_image and ":" in resolved_control_image:
2590        control_version = resolved_control_image.rsplit(":", 1)[-1]
2591
2592    # Get tester identity from environment (GitHub Actions sets GITHUB_ACTOR)
2593    tester = os.getenv("GITHUB_ACTOR") or os.getenv("USER")
2594
2595    track_regression_test(
2596        user_id=tester,
2597        connector_image=resolved_test_image,
2598        command=command,
2599        target_version=target_version,
2600        control_version=control_version,
2601        additional_properties={
2602            "connection_id": connection_id,
2603            "skip_compare": skip_compare,
2604            "with_state": cmd == Command.READ_WITH_STATE,
2605        },
2606    )
2607
2608    # Execute the appropriate mode
2609    if skip_compare:
2610        # Single-version mode: run only the connector image. There is no control
2611        # to compare its spec or catalog against, so the declared objects go
2612        # unread here.
2613        result, _ = _run_connector_command(
2614            connector_image=resolved_test_image,
2615            command=cmd,
2616            output_dir=output_path,
2617            target_or_control=TargetOrControl.TARGET,
2618            config_path=config_file,
2619            catalog_path=catalog_file,
2620            state_path=state_file,
2621            enable_debug_logs=enable_debug_logs,
2622            extra_env=parsed_extra_env,
2623        )
2624
2625        # Same per-version rule as comparison mode: for CHECK the exit code alone
2626        # is not the verdict, connectionStatus is.
2627        annotated_result = _annotate_run_verdict(command, result)
2628        single_success = bool(annotated_result["success"])
2629
2630        print_json(annotated_result)
2631
2632        single_outputs: dict[str, object] = {
2633            "success": single_success,
2634            "connector": resolved_test_image,
2635            "command": command,
2636            "exit_code": result["exit_code"],
2637        }
2638        if command == "check":
2639            # Named to match the comparison-mode output the workflow summary reads.
2640            single_outputs["target_connection_status"] = result.get("connection_status")
2641
2642        write_github_outputs(single_outputs)
2643
2644        # Generate report.md with detailed metrics
2645        report_path = generate_single_version_report(
2646            connector_image=resolved_test_image,
2647            command=command,
2648            result=result,
2649            output_dir=output_path,
2650        )
2651        print_success(f"Generated report: {report_path}")
2652
2653        # The detailed surface is report.html in the uploaded artifact; the step
2654        # summary gets the concise block instead of the whole report. Both are
2655        # built from the unannotated result: the model derives the verdict itself.
2656        report_model = build_single_version_report_model(
2657            connector_image=resolved_test_image,
2658            command=command,
2659            result=result,
2660            configured_streams=_configured_stream_names(cmd, catalog_file),
2661            connection_objects=_collect_connection_objects(
2662                catalog_file, state_file, output_path / "airbyte_messages"
2663            ),
2664        )
2665        html_report_path = write_html_report(report_model, output_path)
2666        print_success(f"Generated HTML report: {html_report_path}")
2667
2668        # Write the concise summary to GITHUB_STEP_SUMMARY (if env var exists)
2669        write_github_summary(render_inline_summary(report_model))
2670
2671        if single_success:
2672            print_success(
2673                f"Single-version regression test passed for {resolved_test_image}"
2674            )
2675        elif command == "check" and result["success"]:
2676            print_error(
2677                f"Single-version regression test failed for {resolved_test_image}: "
2678                f"connectionStatus {result.get('connection_status') or 'missing'} "
2679                "(the connector itself exited 0)."
2680            )
2681        else:
2682            print_error(
2683                f"Single-version regression test failed for {resolved_test_image}"
2684            )
2685    else:
2686        # Comparison mode: run both control and target images. The control
2687        # version runs FIRST on purpose: the record comparison is
2688        # directional (extra records in the target are allowed, missing
2689        # ones fail), so data created upstream between the two live runs
2690        # must land in the target output — where it is tolerated — rather
2691        # than in the control output, where it would read as records
2692        # missing from the target and fail the comparison. Accepted side
2693        # effect (applies to every command, not just read): a hanging or
2694        # timing-out control run means the target never runs, so such a
2695        # run says nothing about the version under test.
2696        #
2697        # HTTP replay needs the same order for its own reason: the target
2698        # replays the dump the control run recorded, so that dump has to
2699        # exist before the target starts.
2700        target_output = output_path / "target"
2701        control_output = output_path / "control"
2702
2703        control_sandbox = _restore_sandbox_baseline(sandbox, sandbox_scenario)
2704        control_result, control_outputs = _run_with_optional_http_metrics(
2705            connector_image=resolved_control_image,  # type: ignore[arg-type]
2706            command=cmd,
2707            output_dir=control_output,
2708            target_or_control=TargetOrControl.CONTROL,
2709            enable_http_metrics=enable_http_metrics,
2710            config_path=config_file,
2711            catalog_path=catalog_file,
2712            state_path=state_file,
2713            enable_debug_logs=enable_debug_logs,
2714            extra_env=parsed_extra_env,
2715            stream_large_bodies=http_replay_max_body,
2716        )
2717        control_sandbox = _check_sandbox_twin_gaps(
2718            sandbox, sandbox_scenario, control_sandbox, repro_case
2719        )
2720
2721        replay_from: Path | None = None
2722        if replay_enabled:
2723            control_dump = control_result.get("http_dump_file")
2724            if control_dump:
2725                replay_from = Path(control_dump)
2726            else:
2727                print_warning(
2728                    "Control run recorded no HTTP dump; the target will run live"
2729                )
2730
2731        target_sandbox = _restore_sandbox_baseline(sandbox, sandbox_scenario)
2732        target_result, target_outputs = _run_with_optional_http_metrics(
2733            connector_image=resolved_test_image,
2734            command=cmd,
2735            output_dir=target_output,
2736            target_or_control=TargetOrControl.TARGET,
2737            enable_http_metrics=enable_http_metrics,
2738            config_path=config_file,
2739            catalog_path=catalog_file,
2740            state_path=state_file,
2741            enable_debug_logs=enable_debug_logs,
2742            extra_env=parsed_extra_env,
2743            replay_expected=replay_enabled,
2744            replay_from=replay_from,
2745            replay_options=replay_options,
2746            stream_large_bodies=http_replay_max_body,
2747        )
2748        target_sandbox = _check_sandbox_twin_gaps(
2749            sandbox, sandbox_scenario, target_sandbox, repro_case
2750        )
2751
2752        # Both dumps have served their purpose by now -- metrics parsed, replay
2753        # corpus built -- so anything too large to be a usable artifact goes,
2754        # rather than being uploaded from a runner with ~14 GB of disk.
2755        _discard_oversized_http_dumps(
2756            control_result,
2757            target_result,
2758            max_bytes=http_replay_max_dump_size_mb * 1024 * 1024,
2759        )
2760
2761        outcome = evaluate_comparison_outcome(command, target_result, control_result)
2762
2763        # A reproduction case replaces the record and final-state comparisons
2764        # for a read: each side is held to the case's expectations instead of
2765        # to the other side.
2766        reproduction: ReproductionVerdict | None = None
2767        if repro_case is not None and sandbox is not None:
2768            reproduction = evaluate_reproduction(
2769                evaluate_side(
2770                    control_result,
2771                    repro_case,
2772                    sandbox_side=control_sandbox,
2773                    final_states=control_outputs.final_states,
2774                ),
2775                evaluate_side(
2776                    target_result,
2777                    repro_case,
2778                    sandbox_side=target_sandbox,
2779                    final_states=target_outputs.final_states,
2780                ),
2781                ReproGoal(repro_goal or ReproGoal.FIXED.value),
2782            )
2783
2784        # Record-level comparison for READ commands (counts, PK presence,
2785        # PK integrity, field values). Only meaningful when both runs
2786        # succeeded; a comparison failure is a regression and fails the
2787        # verdict even though both containers exited cleanly. `outcome`
2788        # itself stays the pure exit-status/connectionStatus layer — the
2789        # folded verdict lives in `success`/`regression_detected` below.
2790        record_comparison: RecordComparisonSummary | None = None
2791        # A skipped comparison must be visible on the report surfaces, not
2792        # only on stdout: without this row the run reads as a pass that
2793        # compared records. Diagnostic, because the skip is not a regression.
2794        record_comparison_skipped: CheckResult | None = None
2795        if (
2796            cmd in (Command.READ, Command.READ_WITH_STATE)
2797            and outcome.both_succeeded
2798            and not skip_record_comparison
2799            and reproduction is None
2800        ):
2801            record_comparison, skip_reason = _compare_read_records(
2802                target_result=target_result,
2803                control_result=control_result,
2804                catalog_file=catalog_file,
2805            )
2806            if record_comparison is None and skip_reason is not None:
2807                record_comparison_skipped = CheckResult(
2808                    name="Record comparison",
2809                    passed=False,
2810                    severity="diagnostic",
2811                    summary=f"Skipped — {skip_reason}; records were not compared",
2812                )
2813        elif (
2814            skip_record_comparison
2815            and cmd in (Command.READ, Command.READ_WITH_STATE)
2816            and reproduction is None
2817        ):
2818            print_warning(
2819                "Record-level comparison skipped (--skip-record-comparison); "
2820                "the verdict gates on command outcomes only."
2821            )
2822            record_comparison_skipped = CheckResult(
2823                name="Record comparison",
2824                passed=False,
2825                severity="diagnostic",
2826                summary="Skipped (--skip-record-comparison); records were not compared",
2827            )
2828        record_comparison_failed = (
2829            record_comparison is not None and not record_comparison.passed
2830        )
2831        # An errored comparison fails the run (fail closed) but as
2832        # INCONCLUSIVE, like both_failed: a comparator crash must not tell a
2833        # connector developer their change caused a regression.
2834        comparison_errored = record_comparison is not None and record_comparison.errored
2835
2836        # Spec/catalog-schema comparison of what the two versions declared
2837        # about themselves; a failed check here is a regression as much as
2838        # a dropped record.
2839        comparison_checks: tuple[CheckResult, ...] = ()
2840        comparison_diff_blocks: tuple[DiffBlock, ...] = ()
2841        comparison_detail: dict[str, Any] = {}
2842        if reproduction is None:
2843            (
2844                comparison_checks,
2845                comparison_diff_blocks,
2846                comparison_detail,
2847            ) = _compare_outputs(
2848                command=command,
2849                outcome=outcome,
2850                control_outputs=control_outputs,
2851                target_outputs=target_outputs,
2852            )
2853        failed_comparisons = failed_checks(comparison_checks)
2854
2855        success = (
2856            outcome.success and not record_comparison_failed and not failed_comparisons
2857        )
2858        regression_detected = (
2859            outcome.regression_detected
2860            or (record_comparison_failed and not comparison_errored)
2861            or bool(failed_comparisons)
2862        )
2863        if reproduction is not None:
2864            success = reproduction.success
2865            regression_detected = reproduction.regression_detected
2866        # In case mode a failing control is the evidence sought, so "both
2867        # failed" would misread a "not fixed" verdict as inconclusive.
2868        both_failed = outcome.both_failed and reproduction is None
2869
2870        # Annotate copies of each side, never the originals: `generate_regression_report`
2871        # below reads `result["success"]` as the exit status, so handing it a dict whose
2872        # `success` is the verdict would corrupt the report's per-version wording.
2873        combined_result = {
2874            "target": _annotate_run_verdict(command, target_result),
2875            "control": _annotate_run_verdict(command, control_result),
2876            "both_succeeded": outcome.both_succeeded,
2877            "regression_detected": regression_detected,
2878            # The authoritative verdict, so consumers of this payload (stdout and
2879            # the `regression_report` output) never have to re-derive it from
2880            # `regression_detected` -- which is false for an inconclusive run.
2881            "success": success,
2882            "both_failed": both_failed,
2883        }
2884        if record_comparison is not None:
2885            combined_result["record_comparison"] = record_comparison.to_dict()
2886        if comparison_detail:
2887            combined_result["comparison"] = comparison_detail
2888        if reproduction is not None:
2889            combined_result["reproduction"] = reproduction.to_dict()
2890
2891        print_json(combined_result)
2892
2893        gh_outputs: dict[str, object] = {
2894            "success": success,
2895            "target_image": resolved_test_image,
2896            "control_image": resolved_control_image,
2897            "command": command,
2898            "target_exit_code": target_result["exit_code"],
2899            "control_exit_code": control_result["exit_code"],
2900            "regression_detected": regression_detected,
2901            "both_failed": both_failed,
2902        }
2903
2904        if command == "check":
2905            gh_outputs["target_connection_status"] = target_result.get(
2906                "connection_status"
2907            )
2908            gh_outputs["control_connection_status"] = control_result.get(
2909                "connection_status"
2910            )
2911
2912        if record_comparison is not None:
2913            gh_outputs["record_comparison_passed"] = record_comparison.passed
2914
2915        if reproduction is not None:
2916            gh_outputs["reproduction_verdict"] = reproduction.label
2917
2918        write_github_outputs(gh_outputs)
2919
2920        write_json_output("regression_report", combined_result)
2921
2922        if reproduction is not None and repro_case is not None:
2923            result_path = output_path / CASE_RESULT_FILE
2924            result_path.write_text(
2925                json.dumps(
2926                    case_result(
2927                        reproduction,
2928                        repro_case,
2929                        control_image=resolved_control_image or "",
2930                        target_image=resolved_test_image,
2931                    ),
2932                    indent=2,
2933                )
2934                + "\n"
2935            )
2936            print_success(f"Wrote reproduction result: {result_path}")
2937
2938        report_path = generate_regression_report(
2939            target_image=resolved_test_image,
2940            control_image=resolved_control_image,  # type: ignore[arg-type]
2941            command=command,
2942            target_result=target_result,
2943            control_result=control_result,
2944            output_dir=output_path,
2945            record_comparison=(
2946                record_comparison.to_dict() if record_comparison is not None else None
2947            ),
2948            # Without these the markdown report would announce "no regression"
2949            # over the same run the HTML report and the CLI call a failure --
2950            # and, for a warning, would show no finding at all where the other
2951            # two surfaces show one.
2952            comparison_findings=[
2953                (check.status, f"{check.name}: {check.summary}")
2954                for check in comparison_checks
2955                if check.status in {"fail", "warn"}
2956            ],
2957            reproduction=reproduction,
2958        )
2959        print_success(f"Generated regression report: {report_path}")
2960
2961        # Same rule as `generate_regression_report` above: the model reads
2962        # `result["success"]` as the exit status and derives the per-version
2963        # verdict itself, so it gets the unannotated originals. `outcome` is
2964        # passed through so the run is evaluated exactly once.
2965        report_model = build_comparison_report_model(
2966            target_image=resolved_test_image,
2967            control_image=resolved_control_image,  # type: ignore[arg-type]
2968            command=command,
2969            target_result=target_result,
2970            control_result=control_result,
2971            outcome=outcome,
2972            record_comparison=record_comparison,
2973            extra_checks=(
2974                *comparison_checks,
2975                *([record_comparison_skipped] if record_comparison_skipped else []),
2976            ),
2977            extra_diff_blocks=comparison_diff_blocks,
2978            configured_streams=_configured_stream_names(cmd, catalog_file),
2979            connection_objects=_collect_connection_objects(
2980                catalog_file, state_file, target_output / "airbyte_messages"
2981            ),
2982            reproduction=reproduction,
2983        )
2984        html_report_path = write_html_report(report_model, output_path)
2985        print_success(f"Generated regression HTML report: {html_report_path}")
2986
2987        # Write the concise summary to GITHUB_STEP_SUMMARY (if env var exists)
2988        write_github_summary(render_inline_summary(report_model))
2989
2990        if reproduction is not None:
2991            report_line = (
2992                f"Reproduction case {repro_case.name if repro_case else ''}: "
2993                f"{reproduction.label} -- {reproduction.detail}"
2994            )
2995            if reproduction.success:
2996                print_success(report_line)
2997            else:
2998                print_error(report_line)
2999                for finding in reproduction.details:
3000                    print_error(f"  - {finding}")
3001        elif failed_comparisons:
3002            # Only reachable when both commands ran cleanly -- a comparison is
3003            # skipped otherwise -- so this says what the exit codes could not.
3004            print_error(
3005                f"Regression detected between {resolved_test_image} and "
3006                f"{resolved_control_image}: "
3007                + "; ".join(
3008                    f"{check.name.lower()} — {check.summary}"
3009                    for check in failed_comparisons
3010                )
3011            )
3012        elif outcome.check_improvement:
3013            print_success(
3014                f"Regression test passed for {resolved_test_image} vs "
3015                f"{resolved_control_image}: target connectionStatus SUCCEEDED "
3016                "while control FAILED (improvement)."
3017            )
3018        elif outcome.both_succeeded and comparison_errored:
3019            print_error(
3020                f"Record comparison errored for {resolved_test_image} vs "
3021                f"{resolved_control_image}. Test failed/inconclusive; this "
3022                "is a tooling failure, not evidence of a regression."
3023            )
3024        elif outcome.both_succeeded and record_comparison_failed:
3025            print_error(
3026                f"Regression detected between {resolved_test_image} and "
3027                f"{resolved_control_image}: both versions ran, but the "
3028                "record comparison failed."
3029            )
3030        elif outcome.both_succeeded:
3031            print_success(
3032                f"Regression test passed for {resolved_test_image} vs {resolved_control_image}"
3033            )
3034        elif outcome.regression_detected:
3035            print_error(
3036                f"Regression detected between {resolved_test_image} and {resolved_control_image}"
3037            )
3038        elif outcome.both_failed:
3039            print_error(
3040                f"Both versions failed for {resolved_test_image} vs {resolved_control_image}. "
3041                "Test failed/inconclusive; cannot rule out a regression."
3042            )
3043        else:
3044            print_error(
3045                f"Control {resolved_control_image} failed while target "
3046                f"{resolved_test_image} succeeded. Test failed/inconclusive; "
3047                "nothing to compare the target against."
3048            )
3049
3050
3051def _repro_case_argument_error(
3052    *,
3053    command: Command,
3054    repro_case_dir: str | None,
3055    sandbox_scenario: str | None,
3056    skip_compare: bool,
3057    connection_id: str | None,
3058    config_path: str | None,
3059    catalog_path: str | None,
3060    state_path: str | None,
3061    selected_streams: str | None,
3062) -> str | None:
3063    """Why the reproduction-case options cannot be used as given, if they cannot."""
3064    if not (repro_case_dir and sandbox_scenario):
3065        return (
3066            "--repro-case-dir and --sandbox-scenario must be given together, "
3067            "and --repro-goal needs them"
3068        )
3069    if command not in (Command.READ, Command.READ_WITH_STATE, Command.CHECK):
3070        # Only a read or a CHECK is judged against a case.
3071        return "Reproduction cases run only `--command read` or `--command check`"
3072    if skip_compare:
3073        return "Reproduction cases need comparison mode; drop --skip-compare"
3074    if connection_id:
3075        return "Reproduction cases cannot use --connection-id"
3076    if not config_path:
3077        return "Reproduction cases need --config-path (from regression-case-setup)"
3078    if catalog_path or state_path:
3079        return "The reproduction case supplies the catalog and state"
3080    if selected_streams:
3081        # The filter rewrites the catalog file in place, i.e. the committed case.
3082        return "The reproduction case's catalog selects its own streams"
3083    return None
3084
3085
3086def _repro_case_command_error(command: Command, case_command: str) -> str | None:
3087    """Why `--command` is not the command the case is judged by, if it is not."""
3088    runs = "check" if command is Command.CHECK else "read"
3089    if runs != case_command:
3090        return f"This case runs only `--command {case_command}`"
3091    return None
3092
3093
3094def _restore_sandbox_baseline(
3095    sandbox: SandboxCli | None, scenario_id: str | None
3096) -> SandboxSide:
3097    """Restore the case's baseline before one side of a case read, else a no-op."""
3098    if sandbox is None or not scenario_id:
3099        return SandboxSide()
3100    side = restore_side(sandbox, scenario_id)
3101    if side.restore_error:
3102        print_error(f"Could not restore the sandbox baseline: {side.restore_error}")
3103    return side
3104
3105
3106def _check_sandbox_twin_gaps(
3107    sandbox: SandboxCli | None,
3108    scenario_id: str | None,
3109    side: SandboxSide,
3110    repro_case: ReproCase | None = None,
3111) -> SandboxSide:
3112    """Gate one side on the twin supporting every request it made."""
3113    if sandbox is None or not scenario_id:
3114        return side
3115    side = check_twin_gaps(
3116        sandbox,
3117        scenario_id,
3118        side,
3119        repro_case.injected_failures if repro_case else (),
3120    )
3121    for gap in side.twin_gaps:
3122        print_error(f"Twin gap: {gap}")
3123    if side.gap_check_error:
3124        print_error(f"Twin-gap check failed: {side.gap_check_error}")
3125    return side
3126
3127
3128@connector_app.command(name="regression-case-setup")
3129def regression_case_setup(
3130    case_dir: Annotated[
3131        str,
3132        Parameter(help="Reproduction case directory to set up."),
3133    ],
3134    config_output: Annotated[
3135        str | None,
3136        Parameter(
3137            help="Where to write the rendered connector config. Defaults to a "
3138            "private temporary directory outside the artifacts tree."
3139        ),
3140    ] = None,
3141) -> None:
3142    """Set up a reproduction case in the SaaS Sandbox.
3143
3144    Applies the case's sandbox/dataset.json together with its optional
3145    sandbox/behavior.json (failure rules), snapshots the result as the
3146    `baseline` that `regression-test --repro-case-dir` restores before each side
3147    of a read, and renders config.template.json with the scenario's
3148    synthetic credential. The sandbox connection comes from `SAAS_API_URL`,
3149    `SAAS_API_KEY`, `SAAS_WORKSPACE_ID` and `XDG_STATE_HOME`.
3150
3151    Writes `scenario_id`, `config_path` and the resolved `case_dir` to the
3152    GitHub step outputs.
3153    """
3154    try:
3155        case = ReproCase.load(Path(case_dir))
3156        sandbox = SandboxCli()
3157        name = f"{case.name}-{os.getenv('GITHUB_RUN_ID') or uuid.uuid4().hex[:8]}"
3158        destination = (
3159            Path(config_output)
3160            if config_output
3161            else Path(tempfile.mkdtemp(prefix=f"repro-case-{case.name}-"))
3162            / "config.json"
3163        )
3164        scenario_id, config_file, credential = set_up_scenario(
3165            case, sandbox, scenario_name=name, config_destination=destination
3166        )
3167    except (OSError, ValueError, SandboxError) as exc:
3168        exit_with_error(f"Could not set up reproduction case {case_dir}: {exc}")
3169
3170    if is_github_actions():
3171        # The workflow log must never show the credential, even a synthetic one.
3172        print(f"::add-mask::{credential}")
3173    outputs = {
3174        "scenario_id": scenario_id,
3175        "config_path": str(config_file),
3176        "case_dir": str(case.directory),
3177        # From a fixed set, so case content cannot inject an output line.
3178        "command": case.expectations.command,
3179    }
3180    write_github_outputs(outputs)
3181    print_json(outputs)
3182
3183
3184@connector_app.command(name="sandbox-support")
3185def sandbox_support(
3186    connector_name: Annotated[
3187        str,
3188        Parameter(help="Connector to check (e.g. 'source-slack')."),
3189    ],
3190    streams: Annotated[
3191        str | None,
3192        Parameter(
3193            help="Comma-separated streams the fix touches; their parent streams "
3194            "are included. Defaults to every stream in the manifest."
3195        ),
3196    ] = None,
3197    dataset_needs: Annotated[
3198        str | None,
3199        Parameter(
3200            help="Comma-separated `entity` or `entity.field` the reproduction "
3201            "needs in the twin's dataset (e.g. 'messages.thread_ts')."
3202        ),
3203    ] = None,
3204    connector_dir: Annotated[
3205        str | None,
3206        Parameter(
3207            help="Local connector directory. Without it, the files are read from "
3208            "airbytehq/airbyte at --ref."
3209        ),
3210    ] = None,
3211    ref: Annotated[
3212        str,
3213        Parameter(help="airbytehq/airbyte ref to read the connector from."),
3214    ] = "master",
3215) -> None:
3216    """Check statically whether a SaaS Sandbox twin covers a connector's needs.
3217
3218    The first gate before a reproduction case: provider host match, the HTTP
3219    operations, query parameters and page sizes the streams use, and optional
3220    dataset needs, against the sandbox's offline discovery data (`sandbox ...
3221    --offline`; needs the `sandbox` CLI, but no server). Prints a JSON report
3222    whose `result` is `supported`, `unsupported` or `unknown`. `unknown` means
3223    the requests could not be read statically; the case lane's twin-gap gate
3224    still decides during the run.
3225    """
3226    print_json(
3227        run_sandbox_support_check(
3228            connector_name=connector_name,
3229            streams=_split_csv(streams),
3230            dataset_needs=_split_csv(dataset_needs),
3231            connector_dir=Path(connector_dir) if connector_dir else None,
3232            ref=ref,
3233        )
3234    )
3235
3236
3237@connector_app.command(name="regression-case-gate")
3238def regression_case_gate(
3239    base_ref: Annotated[
3240        str | None,
3241        Parameter(
3242            help="The branch the PR merges into, e.g. `origin/main`. Required "
3243            "unless --case-dir is given."
3244        ),
3245    ] = None,
3246    head_ref: Annotated[
3247        str,
3248        Parameter(help="The PR's head commit; the case is read at this commit."),
3249    ] = "HEAD",
3250    repo_dir: Annotated[
3251        str,
3252        Parameter(
3253            help="airbyte-ops-mcp clone holding both --base-ref and --head-ref. "
3254            "Its working tree is not read."
3255        ),
3256    ] = ".",
3257    case_dir: Annotated[
3258        str | None,
3259        Parameter(
3260            help="Check this case directory on disk instead of a PR: no scope "
3261            "check, and the connector is the name of its parent directory."
3262        ),
3263    ] = None,
3264    require_case: Annotated[
3265        bool,
3266        Parameter(help="Fail, instead of answering `n/a`, when the PR adds no case."),
3267    ] = False,
3268    connector_dir: Annotated[
3269        str | None,
3270        Parameter(
3271            help="Local connector directory at the case's `affected_ref`. Without "
3272            "it, the files are read from airbytehq/airbyte at `affected_ref`."
3273        ),
3274    ] = None,
3275    json_out: Annotated[
3276        str | None,
3277        Parameter(help="Also write the JSON report to this file."),
3278    ] = None,
3279    sandbox_ref: Annotated[
3280        str | None,
3281        Parameter(
3282            help="The saas-sandbox commit the case runs use. Defaults to the pin "
3283            "in `.github/saas-sandbox-ref` of the current directory (the code "
3284            "checkout); the case's `origin.json` `sandbox_ref` must equal it."
3285        ),
3286    ] = None,
3287) -> None:
3288    """Check whether a reproduction-case PR may merge without human review.
3289
3290    Prints a JSON report whose `status` is `n/a` (the PR touches no case),
3291    `pass` or `fail`, and exits nonzero on `fail`. Checks that the PR only adds
3292    one new case under `regression_cases/<connector>/<case>/`, as regular files
3293    with known names; that the case loads with an `origin.json` naming a
3294    published airbytehq/airbyte release whose source is at `affected_ref`, a
3295    commit on `master`; that the sandbox validates `sandbox/` offline; and that
3296    the static support check says `supported`. The reproduction itself is a
3297    case run with goal `reproduced`. Writes `connector`, `case`,
3298    `affected_image`, `affected_ref`, `problems` (a JSON list) and `status` to
3299    the GitHub step outputs.
3300    """
3301    if sandbox_ref is None:
3302        try:
3303            sandbox_ref = read_sandbox_pin(Path(SANDBOX_PIN_FILE))
3304        except ValueError as exc:
3305            exit_with_error(f"{exc}; run from the code checkout or pass --sandbox-ref")
3306    elif not SHA_PATTERN.fullmatch(sandbox_ref):
3307        exit_with_error("--sandbox-ref must be a full 40-character commit SHA")
3308    token = resolve_default_github_token(allow_none=True)
3309    connector_path = Path(connector_dir) if connector_dir else None
3310    if case_dir is not None:
3311        report = run_case_dir_gate(
3312            Path(case_dir),
3313            sandbox_ref=sandbox_ref,
3314            github_token=token,
3315            connector_dir=connector_path,
3316        )
3317    elif base_ref is None:
3318        exit_with_error("Pass --base-ref, or --case-dir to check a case directory.")
3319    else:
3320        try:
3321            report = run_case_gate(
3322                Path(repo_dir),
3323                base_ref,
3324                head_ref,
3325                sandbox_ref=sandbox_ref,
3326                require_case=require_case,
3327                github_token=token,
3328                connector_dir=connector_path,
3329            )
3330        except subprocess.CalledProcessError as exc:
3331            exit_with_error(
3332                f"Could not compare {base_ref}...{head_ref}: {exc.stderr.strip()}"
3333            )
3334    if json_out:
3335        Path(json_out).write_text(json.dumps(report.to_dict(), indent=2))
3336    write_github_outputs(gate_step_outputs(report))
3337    print_json(report.to_dict())
3338    if report.status == "fail":
3339        exit_with_error("Case gate failed: " + "; ".join(report.problems))
3340
3341
3342@connector_app.command(name="regression-case-validate-decide")
3343def regression_case_validate_decide(
3344    pr: Annotated[int, Parameter(help="The case PR's number.")],
3345    needs: Annotated[
3346        str,
3347        Parameter(help="File holding the workflow's `toJSON(needs)`."),
3348    ],
3349    gate_report: Annotated[
3350        str,
3351        Parameter(help="The gate's `--json-out` report; may be missing."),
3352    ],
3353    merge_enabled: Annotated[
3354        bool,
3355        Parameter(help="The merge switch (`SANDBOX_CASE_AUTOMERGE_ENABLED`)."),
3356    ] = False,
3357    run_url: Annotated[str, Parameter(help="This run's URL, for the comment.")] = "",
3358    repo: Annotated[
3359        str | None,
3360        Parameter(help="`owner/name` of the repository; defaults to `GH_REPO`."),
3361    ] = None,
3362    comment_out: Annotated[
3363        str,
3364        Parameter(help="Where to write the failure comment, when there is one."),
3365    ] = "comment.md",
3366) -> None:
3367    """Decide what `sandbox-case-validate.yml` does after the gate and the runs.
3368
3369    Re-reads the PR through `gh` (set `GH_REPO` and `GH_TOKEN`) and prints a
3370    JSON decision: `action` (`fail`, `report` or `merge`), the commit status
3371    `state` and `description`, and on `fail` the PR comment, also written to
3372    --comment-out. Writes `action`, `state`, `description` and `validated_sha`
3373    to the GitHub step outputs.
3374    """
3375    needs_data = json.loads(Path(needs).read_text())
3376    gate_outputs = (needs_data.get("gate") or {}).get("outputs") or {}
3377    validated_sha = str(gate_outputs.get("head_sha") or "")
3378    # It is written back to the step outputs, so it must be a plain SHA.
3379    if not SHA_PATTERN.fullmatch(validated_sha):
3380        validated_sha = ""
3381    report_path = Path(gate_report)
3382    gate = json.loads(report_path.read_text()) if report_path.is_file() else None
3383    try:
3384        pr_now = gh_read_pr(repo or os.environ.get("GH_REPO", ""), pr)
3385    except (RuntimeError, ValueError, subprocess.SubprocessError):
3386        pr_now = None
3387    decision = decide_validation(
3388        gate=gate,
3389        gate_refusal=gate_outputs.get("refusal") or None,
3390        runs=runs_from_needs(needs_data),
3391        pr=pr_now,
3392        validated_sha=validated_sha,
3393        merge_enabled=merge_enabled,
3394        run_url=run_url,
3395    )
3396    if decision.comment is not None:
3397        Path(comment_out).write_text(decision.comment)
3398    write_github_outputs(
3399        {
3400            "action": decision.action,
3401            "state": decision.state,
3402            "description": decision.description,
3403            "validated_sha": validated_sha,
3404        }
3405    )
3406    print_json(decision.to_dict())
3407
3408
3409@connector_app.command(name="regression-case-merge")
3410def regression_case_merge(
3411    pr: Annotated[int, Parameter(help="The case PR's number.")],
3412    sha: Annotated[str, Parameter(help="The validated head commit.")],
3413    connector: Annotated[str, Parameter(help="The case's connector, for the title.")],
3414    case: Annotated[str, Parameter(help="The case's name, for the title.")],
3415    repo: Annotated[
3416        str | None,
3417        Parameter(help="`owner/name` of the repository; defaults to `GH_REPO`."),
3418    ] = None,
3419    run_url: Annotated[
3420        str, Parameter(help="The validating run, for the commit message.")
3421    ] = "",
3422    timeout_minutes: Annotated[
3423        float,
3424        Parameter(help="How long to wait for GitHub to report the PR mergeable."),
3425    ] = 10,
3426) -> None:
3427    """Merge a validated case PR at its validated head, waiting a bounded time.
3428
3429    Polls the PR through `gh` (set `GH_TOKEN` to a token that may merge) until
3430    GitHub reports it mergeable, then squash-merges it through the REST merge
3431    endpoint pinned to `sha`, with a fixed commit title. Never enables
3432    auto-merge. Prints the outcome and writes `outcome`, `merge_state`, `state`
3433    and `description` (the final commit status) to the GitHub step outputs.
3434    Exits nonzero unless the PR merged at `sha` or simply did not become
3435    mergeable in time.
3436    """
3437    repo = repo or os.environ.get("GH_REPO")
3438    if not repo:
3439        exit_with_error("Pass --repo or set GH_REPO.")
3440    if not (
3441        SHA_PATTERN.fullmatch(sha)
3442        and NAME_PATTERN.fullmatch(connector)
3443        and NAME_PATTERN.fullmatch(case)
3444    ):
3445        exit_with_error("--sha must be a full SHA; --connector and --case plain names.")
3446    title = f"test(regression-cases): add {connector} {case} reproduction case (#{pr})"
3447    message = (
3448        f"Validated by {run_url}" if run_url else "Validated by sandbox-case-validate"
3449    )
3450    outcome = wait_and_merge(
3451        sha,
3452        timeout_seconds=timeout_minutes * 60,
3453        read_pr=lambda: gh_read_pr(repo, pr),
3454        merge=lambda: gh_merge(repo, pr, sha, title, message),
3455    )
3456    result = outcome.to_dict()
3457    write_github_outputs(
3458        {
3459            "outcome": outcome.result,
3460            "merge_state": outcome.merge_state or "",
3461            "state": result["state"],
3462            "description": result["description"],
3463        }
3464    )
3465    print_json(result)
3466    if not outcome.succeeded:
3467        exit_with_error(f"Case PR #{pr} was not merged: {outcome.result}")
3468
3469
3470def _split_csv(value: str | None) -> list[str]:
3471    return [item.strip() for item in (value or "").split(",") if item.strip()]
3472
3473
3474def _configured_stream_names(
3475    command: Command, catalog_file: Path | None
3476) -> tuple[str, ...]:
3477    """Every stream the run was configured to read.
3478
3479    The report's per-stream tables are keyed on emitted records, so a stream
3480    that returned nothing has no entry and would be missing from the report
3481    entirely. Reading the configured catalog is what lets it be shown as zero.
3482
3483    Only for commands that read records. `--connection-id` fetches a catalog for
3484    every command, and handing it to `spec`, `check` or `discover` would give
3485    them an all-zero stream table and a coverage warning saying no stream
3486    returned anything -- true, meaningless, and enough noise to train a reviewer
3487    to ignore the warning on the one command where it means something.
3488    """
3489    if command not in (Command.READ, Command.READ_WITH_STATE):
3490        return ()
3491
3492    if catalog_file is None or not catalog_file.exists():
3493        return ()
3494
3495    try:
3496        catalog = ConfiguredAirbyteCatalog.parse_raw(catalog_file.read_text())
3497    except Exception as exc:
3498        print_warning(f"Could not read the configured catalog for the report: {exc}")
3499        return ()
3500
3501    return tuple(
3502        sorted(
3503            configured.stream.name
3504            for configured in catalog.streams
3505            if configured.stream
3506        )
3507    )
3508
3509
3510def _collect_connection_objects(
3511    catalog_file: Path | None,
3512    state_file: Path | None,
3513    messages_dir: Path | None,
3514) -> tuple[DiffBlock, ...]:
3515    """Gather the objects the run was given or discovered, for the report.
3516
3517    The connector config is deliberately absent: it is the one object that
3518    carries secrets, and the report is an artifact anyone with repo access can
3519    download.
3520
3521    Args:
3522        catalog_file: The configured catalog handed to the run.
3523        state_file: The state handed to the run, for a warm read.
3524        messages_dir: The target side's `airbyte_messages` directory, which
3525            holds the catalog the connector discovered.
3526
3527    Returns:
3528        One JSON block per object that exists, in the order a reader wants them.
3529    """
3530    sources: list[tuple[str, Path | None]] = [
3531        ("Configured catalog (input)", catalog_file),
3532        ("State (input)", state_file),
3533        (
3534            "Discovered catalog (from the target)",
3535            (messages_dir / "catalog.jsonl") if messages_dir else None,
3536        ),
3537    ]
3538
3539    blocks: list[DiffBlock] = []
3540    for title, path in sources:
3541        if path is None or not path.is_file():
3542            continue
3543
3544        try:
3545            payload = path.read_text()
3546        except OSError:
3547            continue
3548
3549        if payload.strip():
3550            blocks.append(build_json_block(title, payload))
3551
3552    return tuple(blocks)
3553
3554
3555# The order commands are presented in, matching how a sync runs them. Anything
3556# else found under the artifacts directory follows, sorted.
3557_REPORT_COMMAND_ORDER = ("spec", "check", "discover", "read")
3558
3559
3560@connector_app.command(name="consolidate-regression-reports")
3561def consolidate_regression_reports(
3562    artifacts_dir: Annotated[
3563        str,
3564        Parameter(
3565            help="Directory holding one subdirectory per command, each with the "
3566            "`report.html` that `regression-test --output-dir` wrote."
3567        ),
3568    ],
3569    output_path: Annotated[
3570        str | None,
3571        Parameter(
3572            name=["--output"],
3573            help="Where to write the consolidated report. "
3574            "Defaults to `report.html` in `--artifacts-dir`.",
3575        ),
3576    ] = None,
3577) -> None:
3578    """Fold one run's per-command regression reports into a single page.
3579
3580    The regression test runs the CLI once per Airbyte command, so each command
3581    writes its own report into its own artifact. This produces the run-level
3582    view -- a verdict table across commands plus every report inlined -- so a
3583    reviewer downloads one file instead of four.
3584
3585    Never fails the run: a command whose report is missing or unreadable is
3586    skipped, because this is a reporting convenience and the per-command
3587    artifacts remain the source of truth.
3588    """
3589    artifacts_path = Path(artifacts_dir)
3590    ordered = [artifacts_path / command for command in _REPORT_COMMAND_ORDER]
3591    extra = sorted(
3592        path
3593        for path in artifacts_path.glob("*")
3594        if path.is_dir() and path not in ordered
3595    )
3596    report_paths = [
3597        directory / HTML_REPORT_FILENAME for directory in [*ordered, *extra]
3598    ]
3599
3600    consolidated = consolidate_reports(report_paths)
3601    if consolidated is None:
3602        print_warning(
3603            f"No per-command reports found under {artifacts_path}; "
3604            "nothing to consolidate."
3605        )
3606        return
3607
3608    destination = Path(output_path) if output_path else artifacts_path / "report.html"
3609    destination.parent.mkdir(parents=True, exist_ok=True)
3610    destination.write_text(consolidated)
3611    print_success(f"Consolidated regression report: {destination}")
3612
3613
3614@connector_app.command(name="fetch-connection-config")
3615def fetch_connection_config_cmd(
3616    connection_id: Annotated[
3617        str,
3618        Parameter(help="The UUID of the Airbyte Cloud connection."),
3619    ],
3620    output_path: Annotated[
3621        str | None,
3622        Parameter(
3623            help="Path to output file or directory. "
3624            "If directory, writes connection-<id>-config.json inside it. "
3625            "Default: platform temp directory (e.g. /tmp/connection-<id>-config.json)"
3626        ),
3627    ] = None,
3628    with_secrets: Annotated[
3629        bool,
3630        Parameter(
3631            name="--with-secrets",
3632            negative="--no-secrets",
3633            help="If set, fetches unmasked secrets from the internal database. "
3634            "Requires GCP_PROD_DB_ACCESS_CREDENTIALS env var or `gcloud auth application-default login`. "
3635            "Must be used with --oc-issue-url.",
3636        ),
3637    ] = False,
3638    oc_issue_url: Annotated[
3639        str | None,
3640        Parameter(
3641            help="OC issue URL for audit logging. Required when using --with-secrets."
3642        ),
3643    ] = None,
3644) -> None:
3645    """Fetch connection configuration from Airbyte Cloud to a local file.
3646
3647    This command retrieves the source configuration for a given connection ID
3648    and writes it to a JSON file. When `--output-path` is omitted the file is
3649    written to the platform temp directory to avoid accidentally committing
3650    secrets to a git repository.
3651
3652    Requires authentication via AIRBYTE_CLOUD_CLIENT_ID and
3653    AIRBYTE_CLOUD_CLIENT_SECRET environment variables.
3654
3655    When --with-secrets is specified, the command fetches unmasked secrets from
3656    the internal database using the connection-retriever. This additionally requires:
3657    - An OC issue URL for audit logging (--oc-issue-url)
3658    - GCP credentials via `GCP_PROD_DB_ACCESS_CREDENTIALS` env var or `gcloud auth application-default login`
3659    - Cloud SQL Python Connector access to the Prod DB replica.
3660    """
3661    path = Path(output_path) if output_path else None
3662    result = fetch_connection_config(
3663        connection_id=connection_id,
3664        output_path=path,
3665        with_secrets=with_secrets,
3666        oc_issue_url=oc_issue_url,
3667    )
3668    if result.success:
3669        print_success(result.message)
3670    else:
3671        print_error(result.message)
3672    print_json(result.model_dump())
3673
3674
3675def _read_json_input(
3676    json_str: str | None,
3677    input_file: str | None,
3678) -> dict:
3679    """Read JSON from a positional arg, --input file, or STDIN (in that priority)."""
3680    if json_str is not None:
3681        return json.loads(json_str)
3682    if input_file is not None:
3683        return json.loads(Path(input_file).read_text())
3684    if not sys.stdin.isatty():
3685        return json.loads(sys.stdin.read())
3686    exit_with_error(
3687        "No JSON input provided. Pass it as a positional argument, "
3688        "via --input <file>, or pipe to STDIN."
3689    )
3690    raise SystemExit(1)  # unreachable, but satisfies type checker
3691
3692
3693@state_app.command(name="get")
3694def get_connection_state(
3695    connection_id: Annotated[
3696        str,
3697        Parameter(help="The connection ID (UUID) to fetch state for."),
3698    ],
3699    stream_name: Annotated[
3700        str | None,
3701        Parameter(
3702            help="Optional stream name to filter state for a single stream.",
3703            name="--stream",
3704        ),
3705    ] = None,
3706    stream_namespace: Annotated[
3707        str | None,
3708        Parameter(
3709            help="Optional stream namespace to narrow the stream filter.",
3710            name="--namespace",
3711        ),
3712    ] = None,
3713    output: Annotated[
3714        str | None,
3715        Parameter(
3716            help="Path to write the state JSON output to a file.",
3717            name="--output",
3718        ),
3719    ] = None,
3720) -> None:
3721    """Get the current state for an Airbyte Cloud connection."""
3722    workspace = CloudWorkspace.from_env()
3723    conn = workspace.get_connection(connection_id)
3724
3725    if stream_name is not None:
3726        stream_state = conn.get_stream_state(
3727            stream_name=stream_name,
3728            stream_namespace=stream_namespace,
3729        )
3730        result = {
3731            "connection_id": connection_id,
3732            "stream_name": stream_name,
3733            "stream_namespace": stream_namespace,
3734            "stream_state": stream_state,
3735        }
3736    else:
3737        result = conn.dump_raw_state()
3738
3739    json_output = json.dumps(result, indent=2, default=str)
3740    if output is not None:
3741        Path(output).write_text(json_output + "\n")
3742        print(f"State written to {output}", file=sys.stderr)
3743    else:
3744        print(json_output)
3745
3746
3747@state_app.command(name="set")
3748def set_connection_state(
3749    connection_id: Annotated[
3750        str,
3751        Parameter(help="The connection ID (UUID) to update state for."),
3752    ],
3753    state_json: Annotated[
3754        str | None,
3755        Parameter(
3756            help="The connection state as a JSON string. When --stream is used, "
3757            'this is just the stream\'s state blob (e.g., \'{"cursor": "2024-01-01"}\').'
3758            " Otherwise, must include 'stateType', 'connectionId', and the appropriate "
3759            "state field. Can also be provided via --input or STDIN."
3760        ),
3761    ] = None,
3762    stream_name: Annotated[
3763        str | None,
3764        Parameter(
3765            help="Optional stream name to update state for a single stream only.",
3766            name="--stream",
3767        ),
3768    ] = None,
3769    stream_namespace: Annotated[
3770        str | None,
3771        Parameter(
3772            help="Optional stream namespace to identify the stream.",
3773            name="--namespace",
3774        ),
3775    ] = None,
3776    input_file: Annotated[
3777        str | None,
3778        Parameter(
3779            help="Path to a JSON file containing the state to set.",
3780            name="--input",
3781        ),
3782    ] = None,
3783) -> None:
3784    """Set the state for an Airbyte Cloud connection.
3785
3786    State JSON can be provided as a positional argument, via --input <file>,
3787    or piped through STDIN.
3788
3789    Uses the safe variant that prevents updates while a sync is running.
3790    When --stream is provided, only that stream's state is updated within
3791    the existing connection state.
3792    """
3793    parsed_state = _read_json_input(state_json, input_file)
3794    workspace = CloudWorkspace.from_env()
3795    conn = workspace.get_connection(connection_id)
3796
3797    if stream_name is not None:
3798        conn.set_stream_state(
3799            stream_name=stream_name,
3800            state_blob_dict=parsed_state,
3801            stream_namespace=stream_namespace,
3802        )
3803    else:
3804        conn.import_raw_state(parsed_state)
3805
3806    result = conn.dump_raw_state()
3807    json_output = json.dumps(result, indent=2, default=str)
3808    print(json_output)
3809
3810
3811@state_app.command(name="reset")
3812def reset_stream_state(
3813    connection_id: Annotated[
3814        str,
3815        Parameter(help="The connection ID (UUID) to update state for."),
3816    ],
3817    stream_name: Annotated[
3818        str,
3819        Parameter(
3820            help="The configured stream name whose state should be reset.",
3821            name="--stream",
3822        ),
3823    ],
3824    stream_namespace: Annotated[
3825        str | None,
3826        Parameter(
3827            help="Optional stream namespace to identify the stream.",
3828            name="--namespace",
3829        ),
3830    ] = None,
3831) -> None:
3832    """Reset a configured stream's state so the next sync full-refreshes it.
3833
3834    Uses the safe variant that prevents updates while a sync is running.
3835    Returns a restorable `previous_state_backup` in raw Config API format.
3836    """
3837    workspace = CloudWorkspace.from_env()
3838    conn = workspace.get_connection(connection_id)
3839
3840    try:
3841        result = reset_cloud_stream_state(
3842            conn,
3843            stream_name=stream_name,
3844            stream_namespace=stream_namespace,
3845        )
3846    except PyAirbyteInputError as e:
3847        exit_with_error(str(e))
3848
3849    print(json.dumps(result.model_dump(), indent=2, default=str))
3850
3851
3852@catalog_app.command(name="get")
3853def get_connection_catalog(
3854    connection_id: Annotated[
3855        str,
3856        Parameter(help="The connection ID (UUID) to fetch catalog for."),
3857    ],
3858    output: Annotated[
3859        str | None,
3860        Parameter(
3861            help="Path to write the catalog JSON output to a file.",
3862            name="--output",
3863        ),
3864    ] = None,
3865) -> None:
3866    """Get the configured catalog for an Airbyte Cloud connection."""
3867    workspace = CloudWorkspace.from_env()
3868    conn = workspace.get_connection(connection_id)
3869    result = conn.dump_raw_catalog()
3870    if result is None:
3871        exit_with_error("No configured catalog found for this connection.")
3872        raise SystemExit(1)  # unreachable, but satisfies type checker
3873
3874    json_output = json.dumps(result, indent=2, default=str)
3875    if output is not None:
3876        Path(output).write_text(json_output + "\n")
3877        print(f"Catalog written to {output}", file=sys.stderr)
3878    else:
3879        print(json_output)
3880
3881
3882@catalog_app.command(name="set")
3883def set_connection_catalog(
3884    connection_id: Annotated[
3885        str,
3886        Parameter(help="The connection ID (UUID) to update catalog for."),
3887    ],
3888    catalog_json: Annotated[
3889        str | None,
3890        Parameter(
3891            help="The configured catalog as a JSON string. "
3892            "Can also be provided via --input or STDIN."
3893        ),
3894    ] = None,
3895    input_file: Annotated[
3896        str | None,
3897        Parameter(
3898            help="Path to a JSON file containing the catalog to set.",
3899            name="--input",
3900        ),
3901    ] = None,
3902) -> None:
3903    """Set the configured catalog for an Airbyte Cloud connection.
3904
3905    Catalog JSON can be provided as a positional argument, via --input <file>,
3906    or piped through STDIN.
3907
3908    WARNING: This replaces the entire configured catalog.
3909    """
3910    parsed_catalog = _read_json_input(catalog_json, input_file)
3911    workspace = CloudWorkspace.from_env()
3912    conn = workspace.get_connection(connection_id)
3913    conn.import_raw_catalog(parsed_catalog)
3914
3915    result = conn.dump_raw_catalog()
3916    if result is None:
3917        exit_with_error("Failed to retrieve catalog after import.")
3918        raise SystemExit(1)  # unreachable, but satisfies type checker
3919
3920    json_output = json.dumps(result, indent=2, default=str)
3921    print(json_output)
3922
3923
3924@logs_app.command(name="lookup-cloud-backend-error")
3925def lookup_cloud_backend_error(
3926    error_id: Annotated[
3927        str,
3928        Parameter(
3929            help=(
3930                "The error ID (UUID) to search for. This is typically returned "
3931                "in API error responses as {'errorId': '...'}"
3932            )
3933        ),
3934    ],
3935    lookback_days: Annotated[
3936        int,
3937        Parameter(help="Number of days to look back in logs."),
3938    ] = 7,
3939    min_severity_filter: Annotated[
3940        GCPSeverity | None,
3941        Parameter(
3942            help="Optional minimum severity level to filter logs.",
3943        ),
3944    ] = None,
3945    raw: Annotated[
3946        bool,
3947        Parameter(help="Output raw JSON instead of formatted text."),
3948    ] = False,
3949) -> None:
3950    """Look up error details from GCP Cloud Logging by error ID.
3951
3952    When an Airbyte Cloud API returns an error response with only an error ID
3953    (e.g., {"errorId": "3173452e-8f22-4286-a1ec-b0f16c1e078a"}), this command
3954    fetches the full stack trace and error details from GCP Cloud Logging.
3955
3956    Requires GCP credentials with Logs Viewer role on the target project.
3957    Set up credentials with: gcloud auth application-default login
3958    """
3959    print(f"Searching for error ID: {error_id}", file=sys.stderr)
3960    print(f"Lookback days: {lookback_days}", file=sys.stderr)
3961    if min_severity_filter:
3962        print(f"Severity filter: {min_severity_filter}", file=sys.stderr)
3963    print(file=sys.stderr)
3964
3965    result = fetch_error_logs(
3966        error_id=error_id,
3967        lookback_days=lookback_days,
3968        min_severity_filter=min_severity_filter,
3969    )
3970
3971    if raw:
3972        print_json(result.model_dump())
3973        return
3974
3975    print(f"Found {result.total_entries_found} log entries", file=sys.stderr)
3976    print(file=sys.stderr)
3977
3978    if result.payloads:
3979        for i, payload in enumerate(result.payloads):
3980            print(f"=== Log Group {i + 1} ===")
3981            print(f"Timestamp: {payload.timestamp}")
3982            print(f"Severity: {payload.severity}")
3983            if payload.resource.labels.pod_name:
3984                print(f"Pod: {payload.resource.labels.pod_name}")
3985            print(f"Lines: {payload.num_log_lines}")
3986            print()
3987            print(payload.message)
3988            print()
3989    elif result.entries:
3990        print("No grouped payloads, showing raw entries:", file=sys.stderr)
3991        for entry in result.entries:
3992            print(f"[{entry.timestamp}] {entry.severity}: {entry.payload}")
3993    else:
3994        print_error("No log entries found for this error ID.")
3995
3996
3997# =============================================================================
3998# Organization commands
3999# =============================================================================
4000
4001
4002@organization_app.command(name="search")
4003def organization_search(
4004    name_contains: Annotated[
4005        str,
4006        Parameter(
4007            name="--name-contains",
4008            help="Case-insensitive substring to match against organization name or email.",
4009        ),
4010    ],
4011    limit: Annotated[
4012        int,
4013        Parameter(
4014            name="--limit",
4015            help="Maximum number of results to return.",
4016        ),
4017    ] = 20,
4018) -> None:
4019    """Search organizations by name or email substring."""
4020    rows = _search_organizations(name_contains=name_contains, limit=limit)
4021    if not rows:
4022        print_error(f"No organizations found matching '{name_contains}'.")
4023        return
4024
4025    print_json([dict(row) for row in rows])
4026
4027
4028@organization_app.command(name="info")
4029def organization_info(
4030    organization_id: Annotated[
4031        str,
4032        Parameter(help="The organization UUID."),
4033    ],
4034) -> None:
4035    """Get information about an organization (stub — not yet implemented)."""
4036    exit_with_error(
4037        "organization info is not yet implemented. "
4038        "Use 'airbyte-ops cloud organization search' to find organizations by name."
4039    )
4040
4041
4042# =============================================================================
4043# Workspace commands
4044# =============================================================================
4045
4046
4047@workspace_app.command(name="search")
4048def workspace_search(
4049    name_contains: Annotated[
4050        str | None,
4051        Parameter(
4052            name="--name-contains",
4053            help="Case-insensitive substring to match against workspace name or slug.",
4054        ),
4055    ] = None,
4056    email_domain: Annotated[
4057        str | None,
4058        Parameter(
4059            name="--email-domain",
4060            help="Email domain to search for (e.g. 'motherduck.com').",
4061        ),
4062    ] = None,
4063    limit: Annotated[
4064        int,
4065        Parameter(
4066            name="--limit",
4067            help="Maximum number of results to return.",
4068        ),
4069    ] = 100,
4070) -> None:
4071    """Search workspaces by name substring or email domain."""
4072    if not name_contains and not email_domain:
4073        exit_with_error(
4074            "At least one of --name-contains or --email-domain must be provided."
4075        )
4076
4077    if name_contains:
4078        rows = _search_workspaces(name_contains=name_contains, limit=limit)
4079    else:
4080        rows = query_workspaces_by_email_domain(
4081            email_domain=email_domain.lstrip("@"),  # type: ignore[union-attr]
4082            limit=limit,
4083        )
4084
4085    if not rows:
4086        search_term = name_contains or email_domain
4087        print_error(f"No workspaces found matching '{search_term}'.")
4088        return
4089
4090    print_json([dict(row) for row in rows])
4091
4092
4093@workspace_app.command(name="info")
4094def workspace_info_cmd(
4095    workspace_id: Annotated[
4096        str,
4097        Parameter(help="The workspace UUID."),
4098    ],
4099) -> None:
4100    """Get information about a workspace."""
4101    result = query_workspace_info(workspace_id=workspace_id)
4102    if not result:
4103        print_error(f"No workspace found with ID '{workspace_id}'.")
4104        return
4105
4106    print_json(dict(result))