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
  41from pathlib import Path
  42from typing import Annotated, Any, Literal
  43
  44import requests
  45import yaml
  46from airbyte.cloud.workspaces import CloudWorkspace
  47from airbyte.exceptions import PyAirbyteInputError
  48from airbyte_cdk.models.connector_metadata import MetadataFile
  49from airbyte_cdk.utils.connector_paths import find_connector_root_from_name
  50from airbyte_cdk.utils.docker import build_connector_image, verify_docker_installation
  51from airbyte_protocol.models import ConfiguredAirbyteCatalog
  52from cyclopts import Parameter
  53from fastmcp_extensions.cli import (
  54    exit_with_error,
  55    print_error,
  56    print_json,
  57    print_success,
  58    print_warning,
  59)
  60
  61from airbyte_ops_mcp.cli._base import App, app
  62from airbyte_ops_mcp.cloud_admin.auth import CloudAuthError
  63from airbyte_ops_mcp.cloud_admin.connection_config import fetch_connection_config
  64from airbyte_ops_mcp.cloud_admin.connection_state import (
  65    reset_stream_state as reset_cloud_stream_state,
  66)
  67from airbyte_ops_mcp.cloud_admin.registry_lookup import (
  68    resolve_definition_id_to_canonical_info,
  69)
  70from airbyte_ops_mcp.cloud_admin.version_overrides import (
  71    ResolvedCloudAuth,
  72    VersionOverrideTarget,
  73    get_connector_version_info,
  74)
  75from airbyte_ops_mcp.cloud_admin.version_overrides import (
  76    set_version_override as set_cloud_version_override,
  77)
  78from airbyte_ops_mcp.constants import (
  79    CLOUD_SQL_INSTANCE,
  80    CLOUD_SQL_PROXY_PID_FILE,
  81    DEFAULT_CLOUD_SQL_PROXY_PORT,
  82    ENV_GCP_PROD_DB_ACCESS_CREDENTIALS,
  83)
  84from airbyte_ops_mcp.gcp_logs import GCPSeverity, fetch_error_logs
  85from airbyte_ops_mcp.prod_db_access.queries import (
  86    query_workspace_info,
  87    query_workspaces_by_email_domain,
  88)
  89from airbyte_ops_mcp.prod_db_access.queries import (
  90    search_organizations as _search_organizations,
  91)
  92from airbyte_ops_mcp.prod_db_access.queries import (
  93    search_workspaces as _search_workspaces,
  94)
  95from airbyte_ops_mcp.regression_tests.cdk_secrets import get_first_config_from_secrets
  96from airbyte_ops_mcp.regression_tests.ci_output import (
  97    ComparisonOutcome,
  98    command_result_succeeded,
  99    evaluate_comparison_outcome,
 100    generate_regression_report,
 101    generate_single_version_report,
 102    write_github_output,
 103    write_github_outputs,
 104    write_github_summary,
 105    write_json_output,
 106)
 107from airbyte_ops_mcp.regression_tests.config_overrides import (
 108    filter_configured_catalog_file,
 109)
 110from airbyte_ops_mcp.regression_tests.connection_fetcher import (
 111    enrich_catalog_schemas_from_cloud,
 112    fetch_connection_data,
 113    fetch_connection_state,
 114    save_connection_data_to_files,
 115    streams_missing_schemas,
 116)
 117from airbyte_ops_mcp.regression_tests.connection_secret_retriever import (
 118    SecretRetrievalError,
 119    enrich_catalog_schemas_from_platform_db,
 120    enrich_config_with_secrets,
 121    should_use_secret_retriever,
 122)
 123from airbyte_ops_mcp.regression_tests.connector_runner import (
 124    ConnectorRunner,
 125    ensure_image_available,
 126)
 127from airbyte_ops_mcp.regression_tests.http_metrics import (
 128    MitmproxyManager,
 129    parse_http_dump,
 130)
 131from airbyte_ops_mcp.regression_tests.models import (
 132    Command,
 133    ComparableOutputs,
 134    ConnectorUnderTest,
 135    ExecutionInputs,
 136    TargetOrControl,
 137    duplicate_stream_names,
 138    get_primary_keys_per_stream,
 139    split_records_per_stream,
 140)
 141from airbyte_ops_mcp.regression_tests.regression import (
 142    ComparisonResult,
 143    RecordComparisonSummary,
 144    compare_catalog_schemas,
 145    compare_final_states,
 146    compare_specs,
 147    run_record_comparisons_from_files,
 148)
 149from airbyte_ops_mcp.regression_tests.report import (
 150    HTML_REPORT_FILENAME,
 151    CheckResult,
 152    DiffBlock,
 153    DiffLine,
 154    build_comparison_report_model,
 155    build_diff_block,
 156    build_json_block,
 157    build_single_version_report_model,
 158    consolidate_reports,
 159    failed_checks,
 160    render_inline_summary,
 161    write_html_report,
 162)
 163from airbyte_ops_mcp.telemetry import track_regression_test
 164from airbyte_ops_mcp.tier_cache import resolve_workspace
 165
 166# Path to connectors directory within the airbyte repo
 167CONNECTORS_SUBDIR = Path("airbyte-integrations") / "connectors"
 168
 169# Create the cloud sub-app
 170cloud_app = App(name="cloud", help="Airbyte Cloud operations.")
 171app.command(cloud_app)
 172
 173# Create the connector sub-app under cloud
 174connector_app = App(
 175    name="connector", help="Deployed connector operations in Airbyte Cloud."
 176)
 177cloud_app.command(connector_app)
 178
 179# Create the db sub-app under cloud
 180db_app = App(name="db", help="Database operations for Airbyte Cloud Prod DB Replica.")
 181cloud_app.command(db_app)
 182
 183# Create the connection sub-app under cloud
 184connection_app = App(name="connection", help="Connection operations in Airbyte Cloud.")
 185cloud_app.command(connection_app)
 186
 187# Create the state sub-app under connection
 188state_app = App(name="state", help="Connection state operations in Airbyte Cloud.")
 189connection_app.command(state_app)
 190
 191# Create the catalog sub-app under connection
 192catalog_app = App(
 193    name="catalog", help="Connection catalog operations in Airbyte Cloud."
 194)
 195connection_app.command(catalog_app)
 196
 197# Create the logs sub-app under cloud
 198logs_app = App(name="logs", help="GCP Cloud Logging operations for Airbyte Cloud.")
 199cloud_app.command(logs_app)
 200
 201# Create the organization sub-app under cloud
 202organization_app = App(
 203    name="organization", help="Organization operations in Airbyte Cloud."
 204)
 205cloud_app.command(organization_app)
 206
 207# Create the workspace sub-app under cloud
 208workspace_app = App(name="workspace", help="Workspace operations in Airbyte Cloud.")
 209cloud_app.command(workspace_app)
 210
 211
 212@db_app.command(name="start-proxy")
 213def start_proxy(
 214    port: Annotated[
 215        int,
 216        Parameter(help="Port for the Cloud SQL Proxy to listen on."),
 217    ] = DEFAULT_CLOUD_SQL_PROXY_PORT,
 218    daemon: Annotated[
 219        bool,
 220        Parameter(
 221            help="Run as daemon in background (default). Use --no-daemon for foreground."
 222        ),
 223    ] = True,
 224) -> None:
 225    """Start the Cloud SQL Proxy for database access.
 226
 227    This command starts the Cloud SQL Auth Proxy to enable connections to the
 228    Airbyte Cloud Prod DB Replica. The proxy is required for database query tools.
 229
 230    By default, runs as a daemon (background process). Use --no-daemon to run in
 231    foreground mode where you can see logs and stop with Ctrl+C.
 232
 233    Credentials are read from the GCP_PROD_DB_ACCESS_CREDENTIALS environment variable,
 234    which should contain the service account JSON credentials.
 235
 236    After starting the proxy, set these environment variables to use database tools:
 237        export USE_CLOUD_SQL_PROXY=1
 238        export DB_PORT={port}
 239
 240    Example:
 241        airbyte-ops cloud db start-proxy
 242        airbyte-ops cloud db start-proxy --port 15432
 243        airbyte-ops cloud db start-proxy --no-daemon
 244    """
 245    # Check if proxy is already running on the requested port (idempotency)
 246    try:
 247        with socket.create_connection(("127.0.0.1", port), timeout=0.5):
 248            # Something is already listening on this port
 249            pid_file = Path(CLOUD_SQL_PROXY_PID_FILE)
 250            pid_info = ""
 251            if pid_file.exists():
 252                pid_info = f" (PID: {pid_file.read_text().strip()})"
 253            print_success(
 254                f"Cloud SQL Proxy is already running on port {port}{pid_info}"
 255            )
 256            print_success("")
 257            print_success("To use database tools, set these environment variables:")
 258            print_success("  export USE_CLOUD_SQL_PROXY=1")
 259            print_success(f"  export DB_PORT={port}")
 260            return
 261    except (OSError, TimeoutError, ConnectionRefusedError):
 262        pass  # Port not in use, proceed with starting proxy
 263
 264    # Check if cloud-sql-proxy is installed
 265    proxy_path = shutil.which("cloud-sql-proxy")
 266    if not proxy_path:
 267        exit_with_error(
 268            "cloud-sql-proxy not found in PATH. "
 269            "Install it from: https://cloud.google.com/sql/docs/mysql/sql-proxy"
 270        )
 271
 272    # Get credentials from environment
 273    creds_json = os.getenv(ENV_GCP_PROD_DB_ACCESS_CREDENTIALS)
 274    if not creds_json:
 275        exit_with_error(
 276            f"{ENV_GCP_PROD_DB_ACCESS_CREDENTIALS} environment variable is not set. "
 277            "This should contain the GCP service account JSON credentials."
 278        )
 279
 280    # Build the command using --json-credentials to avoid writing to disk
 281    cmd = [
 282        proxy_path,
 283        CLOUD_SQL_INSTANCE,
 284        f"--port={port}",
 285        f"--json-credentials={creds_json}",
 286    ]
 287
 288    print_success(f"Starting Cloud SQL Proxy on port {port}...")
 289    print_success(f"Instance: {CLOUD_SQL_INSTANCE}")
 290    print_success("")
 291    print_success("To use database tools, set these environment variables:")
 292    print_success("  export USE_CLOUD_SQL_PROXY=1")
 293    print_success(f"  export DB_PORT={port}")
 294    print_success("")
 295
 296    if daemon:
 297        # Run in background (daemon mode) with log file for diagnostics
 298        log_file_path = Path("/tmp/airbyte-cloud-sql-proxy.log")
 299        log_file = log_file_path.open("ab")
 300        process = subprocess.Popen(
 301            cmd,
 302            stdout=subprocess.DEVNULL,
 303            stderr=log_file,
 304            start_new_session=True,
 305        )
 306
 307        # Brief wait to verify the process started successfully
 308        time.sleep(0.5)
 309        if process.poll() is not None:
 310            # Process exited immediately - read any error output
 311            log_file.close()
 312            error_output = ""
 313            if log_file_path.exists():
 314                error_output = log_file_path.read_text()[-1000:]  # Last 1000 chars
 315            exit_with_error(
 316                f"Cloud SQL Proxy failed to start (exit code: {process.returncode}).\n"
 317                f"Check logs at {log_file_path}\n"
 318                f"Recent output: {error_output}"
 319            )
 320
 321        # Write PID to file for stop-proxy command
 322        pid_file = Path(CLOUD_SQL_PROXY_PID_FILE)
 323        pid_file.write_text(str(process.pid))
 324        print_success(f"Cloud SQL Proxy started as daemon (PID: {process.pid})")
 325        print_success(f"Logs: {log_file_path}")
 326        print_success("To stop: airbyte-ops cloud db stop-proxy")
 327    else:
 328        # Run in foreground - replace current process
 329        # Signals (Ctrl+C) will be handled directly by the cloud-sql-proxy process
 330        print_success("Running in foreground. Press Ctrl+C to stop the proxy.")
 331        print_success("")
 332        os.execv(proxy_path, cmd)
 333
 334
 335@db_app.command(name="stop-proxy")
 336def stop_proxy() -> None:
 337    """Stop the Cloud SQL Proxy daemon.
 338
 339    This command stops a Cloud SQL Proxy that was started with 'start-proxy'.
 340    It reads the PID from the PID file and sends a SIGTERM signal to stop the process.
 341
 342    Example:
 343        airbyte-ops cloud db stop-proxy
 344    """
 345    pid_file = Path(CLOUD_SQL_PROXY_PID_FILE)
 346
 347    if not pid_file.exists():
 348        exit_with_error(
 349            f"PID file not found at {CLOUD_SQL_PROXY_PID_FILE}. "
 350            "No Cloud SQL Proxy daemon appears to be running."
 351        )
 352
 353    pid_str = pid_file.read_text().strip()
 354    if not pid_str.isdigit():
 355        pid_file.unlink()
 356        exit_with_error(f"Invalid PID in {CLOUD_SQL_PROXY_PID_FILE}: {pid_str}")
 357
 358    pid = int(pid_str)
 359
 360    # Check if process is still running
 361    try:
 362        os.kill(pid, 0)  # Signal 0 just checks if process exists
 363    except ProcessLookupError:
 364        pid_file.unlink()
 365        print_success(
 366            f"Cloud SQL Proxy (PID: {pid}) is not running. Cleaned up PID file."
 367        )
 368        return
 369    except PermissionError:
 370        exit_with_error(f"Permission denied to check process {pid}.")
 371
 372    # Send SIGTERM to stop the process
 373    try:
 374        os.kill(pid, signal.SIGTERM)
 375        print_success(f"Sent SIGTERM to Cloud SQL Proxy (PID: {pid}).")
 376    except ProcessLookupError:
 377        print_success(f"Cloud SQL Proxy (PID: {pid}) already stopped.")
 378    except PermissionError:
 379        exit_with_error(f"Permission denied to stop process {pid}.")
 380
 381    # Clean up PID file
 382    pid_file.unlink(missing_ok=True)
 383    print_success("Cloud SQL Proxy stopped.")
 384
 385
 386@connector_app.command(name="get-version-info")
 387def get_version_info(
 388    workspace_id: Annotated[
 389        str,
 390        Parameter(help="The Airbyte Cloud workspace ID."),
 391    ],
 392    connector_id: Annotated[
 393        str,
 394        Parameter(help="The ID of the deployed connector (source or destination)."),
 395    ],
 396    connector_type: Annotated[
 397        Literal["source", "destination"],
 398        Parameter(help="The type of connector."),
 399    ],
 400) -> None:
 401    """Get the current version information for a deployed connector."""
 402    result = get_connector_version_info(
 403        auth=_resolve_cli_cloud_auth(),
 404        workspace_id=workspace_id,
 405        actor_id=connector_id,
 406        actor_type=connector_type,
 407    )
 408    print_json(result.model_dump())
 409
 410
 411_OverrideLevel = Literal["actor", "workspace", "organization"]
 412_TierFilter = Literal["TIER_0", "TIER_1", "TIER_2", "UNKNOWN", "ALL"]
 413
 414
 415def _resolve_cli_cloud_auth() -> ResolvedCloudAuth:
 416    """Resolve Airbyte Cloud auth from environment variables for CLI use.
 417
 418    Mirrors the priority order the MCP server uses:
 419
 420    1. `AIRBYTE_CLOUD_BEARER_TOKEN`
 421    2. `AIRBYTE_CLOUD_CLIENT_ID` + `AIRBYTE_CLOUD_CLIENT_SECRET`
 422
 423    Raises a CLI error via `exit_with_error` if no usable credentials are present.
 424    """
 425    bearer_token = os.environ.get("AIRBYTE_CLOUD_BEARER_TOKEN")
 426    if bearer_token:
 427        return ResolvedCloudAuth(bearer_token=bearer_token)
 428
 429    client_id = os.environ.get("AIRBYTE_CLOUD_CLIENT_ID")
 430    client_secret = os.environ.get("AIRBYTE_CLOUD_CLIENT_SECRET")
 431    if client_id and client_secret:
 432        return ResolvedCloudAuth(client_id=client_id, client_secret=client_secret)
 433
 434    exit_with_error(
 435        "Missing Airbyte Cloud credentials. Set AIRBYTE_CLOUD_BEARER_TOKEN, or "
 436        "both AIRBYTE_CLOUD_CLIENT_ID and AIRBYTE_CLOUD_CLIENT_SECRET."
 437    )
 438
 439
 440def _validate_override_scope_args(
 441    *,
 442    override_level: _OverrideLevel,
 443    workspace_id: str | None,
 444    organization_id: str | None,
 445    connector_id: str | None,
 446    connector_type: Literal["source", "destination"] | None,
 447    actor_definition_id: str | None,
 448) -> None:
 449    """Validate scope-specific arguments for `set/clear-version-override`.
 450
 451    Calls `exit_with_error` (which terminates the process) if the supplied
 452    arguments are inconsistent with the requested `override_level`.
 453    """
 454    if override_level == "actor":
 455        missing = [
 456            name
 457            for name, value in (
 458                ("--workspace-id", workspace_id),
 459                ("--connector-id", connector_id),
 460                ("--connector-type", connector_type),
 461            )
 462            if not value
 463        ]
 464        if missing:
 465            exit_with_error(
 466                "Override level 'actor' requires " + ", ".join(missing) + "."
 467            )
 468        if actor_definition_id is not None:
 469            exit_with_error(
 470                "Override level 'actor' must not be combined with "
 471                "--actor-definition-id."
 472            )
 473        if organization_id is not None:
 474            exit_with_error(
 475                "Override level 'actor' must not be combined with --organization-id."
 476            )
 477        return
 478
 479    if override_level == "workspace":
 480        missing = [
 481            name
 482            for name, value in (
 483                ("--workspace-id", workspace_id),
 484                ("--actor-definition-id", actor_definition_id),
 485            )
 486            if not value
 487        ]
 488        if missing:
 489            exit_with_error(
 490                "Override level 'workspace' requires " + ", ".join(missing) + "."
 491            )
 492        if connector_id is not None or connector_type is not None:
 493            exit_with_error(
 494                "Override level 'workspace' must not be combined with "
 495                "--connector-id or --connector-type."
 496            )
 497        if organization_id is not None:
 498            exit_with_error(
 499                "Override level 'workspace' must not be combined with "
 500                "--organization-id."
 501            )
 502        return
 503
 504    # override_level == "organization"
 505    missing = [
 506        name
 507        for name, value in (
 508            ("--organization-id", organization_id),
 509            ("--actor-definition-id", actor_definition_id),
 510        )
 511        if not value
 512    ]
 513    if missing:
 514        exit_with_error(
 515            "Override level 'organization' requires " + ", ".join(missing) + "."
 516        )
 517    if connector_id is not None or connector_type is not None:
 518        exit_with_error(
 519            "Override level 'organization' must not be combined with "
 520            "--connector-id or --connector-type."
 521        )
 522    if workspace_id is not None:
 523        exit_with_error(
 524            "Override level 'organization' must not be combined with --workspace-id."
 525        )
 526
 527
 528def _dispatch_version_override(
 529    *,
 530    override_level: _OverrideLevel,
 531    workspace_id: str | None,
 532    organization_id: str | None,
 533    connector_id: str | None,
 534    connector_type: Literal["source", "destination"] | None,
 535    actor_definition_id: str | None,
 536    version: str | None,
 537    unset: bool,
 538    reason: str | None,
 539    reason_url: str | None,
 540    issue_url: str,
 541    approval_comment_url: str,
 542    ai_agent_session_url: str | None,
 543    customer_tier_filter: _TierFilter,
 544) -> None:
 545    """Route a version-override request through the normalized core helper.
 546
 547    Resolves `--actor-definition-id` to its canonical name and connector type
 548    via the cloud registry for `workspace`/`organization` scopes. Prints the
 549    operation result as JSON.
 550    """
 551    _validate_override_scope_args(
 552        override_level=override_level,
 553        workspace_id=workspace_id,
 554        organization_id=organization_id,
 555        connector_id=connector_id,
 556        connector_type=connector_type,
 557        actor_definition_id=actor_definition_id,
 558    )
 559
 560    auth = _resolve_cli_cloud_auth()
 561
 562    if override_level == "actor":
 563        assert workspace_id is not None
 564        assert connector_id is not None
 565        assert connector_type is not None
 566        assert organization_id is None
 567        try:
 568            ws_resolution = resolve_workspace(workspace_id)
 569            if not ws_resolution.organization_id:
 570                exit_with_error("Could not resolve organization for workspace.")
 571            result = set_cloud_version_override(
 572                auth=auth,
 573                target=VersionOverrideTarget(
 574                    scope="actor",
 575                    organization_id=ws_resolution.organization_id,
 576                    workspace_id=workspace_id,
 577                    actor_id=connector_id,
 578                    connector_type=connector_type,
 579                ),
 580                approval_comment_url=approval_comment_url,
 581                version=version,
 582                unset=unset,
 583                override_reason=reason,
 584                override_reason_reference_url=reason_url,
 585                issue_url=issue_url,
 586                ai_agent_session_url=ai_agent_session_url,
 587                customer_tier_filter=customer_tier_filter,
 588            )
 589        except (PyAirbyteInputError, CloudAuthError) as e:
 590            exit_with_error(str(e))
 591        _print_override_result(result.success, result.message)
 592        print_json(result.model_dump())
 593        return
 594
 595    try:
 596        assert actor_definition_id is not None
 597        canonical_name, resolved_connector_type = (
 598            resolve_definition_id_to_canonical_info(actor_definition_id)
 599        )
 600    except (PyAirbyteInputError, requests.RequestException) as e:
 601        exit_with_error(f"Failed to resolve --actor-definition-id: {e}")
 602    typed_connector_type: Literal["source", "destination"] = (
 603        "source" if resolved_connector_type == "source" else "destination"
 604    )
 605
 606    if override_level == "workspace":
 607        assert workspace_id is not None
 608        try:
 609            ws_resolution = resolve_workspace(workspace_id)
 610            if not ws_resolution.organization_id:
 611                exit_with_error("Could not resolve organization for workspace.")
 612            target = VersionOverrideTarget(
 613                scope="workspace",
 614                organization_id=ws_resolution.organization_id,
 615                workspace_id=workspace_id,
 616                connector_name=canonical_name,
 617                connector_type=typed_connector_type,
 618            )
 619        except PyAirbyteInputError as e:
 620            exit_with_error(str(e))
 621    else:
 622        assert organization_id is not None
 623        target = VersionOverrideTarget(
 624            scope="organization",
 625            organization_id=organization_id,
 626            connector_name=canonical_name,
 627            connector_type=typed_connector_type,
 628        )
 629
 630    try:
 631        result = set_cloud_version_override(
 632            auth=auth,
 633            target=target,
 634            approval_comment_url=approval_comment_url,
 635            version=version,
 636            unset=unset,
 637            override_reason=reason,
 638            override_reason_reference_url=reason_url,
 639            issue_url=issue_url,
 640            ai_agent_session_url=ai_agent_session_url,
 641            customer_tier_filter=customer_tier_filter,
 642        )
 643    except (PyAirbyteInputError, CloudAuthError) as e:
 644        exit_with_error(str(e))
 645    _print_override_result(result.success, result.message)
 646    print_json(result.model_dump())
 647
 648
 649def _print_override_result(success: bool, message: str) -> None:
 650    """Emit an override-operation status message via the shared CLI helpers."""
 651    if success:
 652        print_success(message)
 653    else:
 654        print_error(message)
 655
 656
 657@connector_app.command(name="set-version-override")
 658def set_version_override(
 659    version: Annotated[
 660        str,
 661        Parameter(
 662            help="The semver version string to pin to (e.g., '2.1.5-preview.abc1234')."
 663        ),
 664    ],
 665    reason: Annotated[
 666        str,
 667        Parameter(help="Explanation for the override (min 10 characters)."),
 668    ],
 669    issue_url: Annotated[
 670        str,
 671        Parameter(help="GitHub issue URL providing context for this operation."),
 672    ],
 673    approval_comment_url: Annotated[
 674        str,
 675        Parameter(
 676            help="Slack approval record URL where admin authorized this deployment."
 677        ),
 678    ],
 679    override_level: Annotated[
 680        _OverrideLevel,
 681        Parameter(
 682            help=(
 683                "Scope at which to apply the version override: 'actor' (single "
 684                "deployed connector instance), 'workspace' (all instances of a "
 685                "connector type within a workspace), or 'organization' (all "
 686                "instances across an organization). Defaults to 'actor'."
 687            ),
 688        ),
 689    ] = "actor",
 690    workspace_id: Annotated[
 691        str | None,
 692        Parameter(
 693            help=(
 694                "The Airbyte Cloud workspace ID. Required for 'actor' and "
 695                "'workspace' override levels."
 696            )
 697        ),
 698    ] = None,
 699    organization_id: Annotated[
 700        str | None,
 701        Parameter(
 702            help=(
 703                "The Airbyte Cloud organization ID. Required for "
 704                "'organization' override level."
 705            )
 706        ),
 707    ] = None,
 708    connector_id: Annotated[
 709        str | None,
 710        Parameter(
 711            help=(
 712                "The ID of the deployed connector (source or destination). "
 713                "Required for 'actor' override level."
 714            )
 715        ),
 716    ] = None,
 717    connector_type: Annotated[
 718        Literal["source", "destination"] | None,
 719        Parameter(help="The type of connector. Required for 'actor' override level."),
 720    ] = None,
 721    actor_definition_id: Annotated[
 722        str | None,
 723        Parameter(
 724            help=(
 725                "The connector definition UUID. Required for 'workspace' and "
 726                "'organization' override levels."
 727            )
 728        ),
 729    ] = None,
 730    ai_agent_session_url: Annotated[
 731        str | None,
 732        Parameter(
 733            help="URL to AI agent session driving this operation (for auditability)."
 734        ),
 735    ] = None,
 736    reason_url: Annotated[
 737        str | None,
 738        Parameter(help="Optional URL with more context (e.g., issue link)."),
 739    ] = None,
 740    customer_tier_filter: Annotated[
 741        _TierFilter,
 742        Parameter(
 743            help=(
 744                "Tier filter: 'TIER_0', 'TIER_1', 'TIER_2', 'UNKNOWN', or 'ALL'. "
 745                "The operation will be rejected if the actual customer tier does not match. "
 746                "Defaults to 'TIER_2' (non-sensitive customers)."
 747            ),
 748        ),
 749    ] = "TIER_2",
 750) -> None:
 751    """Set a version override for a deployed connector.
 752
 753    Requires admin authentication via `AIRBYTE_INTERNAL_ADMIN_FLAG` and
 754    `AIRBYTE_INTERNAL_ADMIN_USER` environment variables.
 755
 756    The `--override-level` flag selects the scope at which the pin is applied:
 757
 758    - `actor` (default): pins a single deployed connector instance. Requires
 759      `--workspace-id`, `--connector-id`, and `--connector-type`.
 760    - `workspace`: pins ALL instances of a connector type within a workspace.
 761      Requires `--workspace-id` and `--actor-definition-id`.
 762    - `organization`: pins ALL instances of a connector type across an
 763      organization. Requires `--organization-id` and `--actor-definition-id`.
 764
 765    The `customer_tier_filter` gates the operation: the call fails if the
 766    actual tier of the target organization does not match. Use `ALL` to
 767    proceed regardless of tier (a warning is shown for sensitive tiers).
 768    """
 769    _dispatch_version_override(
 770        override_level=override_level,
 771        workspace_id=workspace_id,
 772        organization_id=organization_id,
 773        connector_id=connector_id,
 774        connector_type=connector_type,
 775        actor_definition_id=actor_definition_id,
 776        version=version,
 777        unset=False,
 778        reason=reason,
 779        reason_url=reason_url,
 780        issue_url=issue_url,
 781        approval_comment_url=approval_comment_url,
 782        ai_agent_session_url=ai_agent_session_url,
 783        customer_tier_filter=customer_tier_filter,
 784    )
 785
 786
 787@connector_app.command(name="clear-version-override")
 788def clear_version_override(
 789    issue_url: Annotated[
 790        str,
 791        Parameter(help="GitHub issue URL providing context for this operation."),
 792    ],
 793    approval_comment_url: Annotated[
 794        str,
 795        Parameter(
 796            help="Slack approval record URL where admin authorized this deployment."
 797        ),
 798    ],
 799    override_level: Annotated[
 800        _OverrideLevel,
 801        Parameter(
 802            help=(
 803                "Scope at which to clear the version override: 'actor', "
 804                "'workspace', or 'organization'. Defaults to 'actor'."
 805            ),
 806        ),
 807    ] = "actor",
 808    workspace_id: Annotated[
 809        str | None,
 810        Parameter(
 811            help=(
 812                "The Airbyte Cloud workspace ID. Required for 'actor' and "
 813                "'workspace' override levels."
 814            )
 815        ),
 816    ] = None,
 817    organization_id: Annotated[
 818        str | None,
 819        Parameter(
 820            help=(
 821                "The Airbyte Cloud organization ID. Required for "
 822                "'organization' override level."
 823            )
 824        ),
 825    ] = None,
 826    connector_id: Annotated[
 827        str | None,
 828        Parameter(
 829            help=(
 830                "The ID of the deployed connector (source or destination). "
 831                "Required for 'actor' override level."
 832            )
 833        ),
 834    ] = None,
 835    connector_type: Annotated[
 836        Literal["source", "destination"] | None,
 837        Parameter(help="The type of connector. Required for 'actor' override level."),
 838    ] = None,
 839    actor_definition_id: Annotated[
 840        str | None,
 841        Parameter(
 842            help=(
 843                "The connector definition UUID. Required for 'workspace' and "
 844                "'organization' override levels."
 845            )
 846        ),
 847    ] = None,
 848    ai_agent_session_url: Annotated[
 849        str | None,
 850        Parameter(
 851            help="URL to AI agent session driving this operation (for auditability)."
 852        ),
 853    ] = None,
 854    customer_tier_filter: Annotated[
 855        _TierFilter,
 856        Parameter(
 857            help=(
 858                "Tier filter: 'TIER_0', 'TIER_1', 'TIER_2', 'UNKNOWN', or 'ALL'. "
 859                "The operation will be rejected if the actual customer tier does not match. "
 860                "Defaults to 'TIER_2' (non-sensitive customers)."
 861            ),
 862        ),
 863    ] = "TIER_2",
 864) -> None:
 865    """Clear a version override from a deployed connector.
 866
 867    Requires admin authentication via `AIRBYTE_INTERNAL_ADMIN_FLAG` and
 868    `AIRBYTE_INTERNAL_ADMIN_USER` environment variables.
 869
 870    The `--override-level` flag selects the scope at which the pin is removed:
 871
 872    - `actor` (default): clears a single deployed connector instance pin.
 873      Requires `--workspace-id`, `--connector-id`, and `--connector-type`.
 874    - `workspace`: clears the workspace-level pin for a connector type.
 875      Requires `--workspace-id` and `--actor-definition-id`.
 876    - `organization`: clears the organization-level pin for a connector type.
 877      Requires `--organization-id` and `--actor-definition-id`.
 878    """
 879    _dispatch_version_override(
 880        override_level=override_level,
 881        workspace_id=workspace_id,
 882        organization_id=organization_id,
 883        connector_id=connector_id,
 884        connector_type=connector_type,
 885        actor_definition_id=actor_definition_id,
 886        version=None,
 887        unset=True,
 888        reason=None,
 889        reason_url=None,
 890        issue_url=issue_url,
 891        approval_comment_url=approval_comment_url,
 892        ai_agent_session_url=ai_agent_session_url,
 893        customer_tier_filter=customer_tier_filter,
 894    )
 895
 896
 897def _load_json_file(file_path: Path) -> dict | None:
 898    """Load a JSON file and return its contents.
 899
 900    Returns None if the file doesn't exist or contains invalid JSON.
 901    """
 902    if not file_path.exists():
 903        return None
 904    try:
 905        return json.loads(file_path.read_text())
 906    except json.JSONDecodeError as e:
 907        print_error(f"Failed to parse JSON in file: {file_path}\nError: {e}")
 908        return None
 909
 910
 911def _run_connector_command(
 912    connector_image: str,
 913    command: Command,
 914    output_dir: Path,
 915    target_or_control: TargetOrControl,
 916    config_path: Path | None = None,
 917    catalog_path: Path | None = None,
 918    state_path: Path | None = None,
 919    proxy_url: str | None = None,
 920    enable_debug_logs: bool = False,
 921) -> tuple[dict, ComparableOutputs]:
 922    """Run a connector command and return its results and its declared objects.
 923
 924    The dict is a flattened view of `ExecutionResult`: counts, exit status and
 925    artifact paths, which is all the reports need. What a comparison needs
 926    instead -- the spec, the discovered catalog -- travels beside it as
 927    `ComparableOutputs` rather than growing a dict that several consumers read.
 928    The `ExecutionResult` itself is still dropped here: it holds every parsed
 929    message, and a comparison keeps both sides alive at once.
 930
 931    Args:
 932        connector_image: Full connector image name with tag.
 933        command: The Airbyte command to run.
 934        output_dir: Directory to store output files.
 935        target_or_control: Whether this is target or control version.
 936        config_path: Path to connector config JSON file.
 937        catalog_path: Path to configured catalog JSON file.
 938        state_path: Path to state JSON file.
 939        proxy_url: Optional HTTP proxy URL for traffic capture.
 940        enable_debug_logs: If `True`, pass `LOG_LEVEL=DEBUG` to the connector container.
 941
 942    Returns:
 943        The flattened results, and the objects the connector declared.
 944    """
 945    connector = ConnectorUnderTest.from_image_name(connector_image, target_or_control)
 946
 947    config = _load_json_file(config_path) if config_path else None
 948    state = _load_json_file(state_path) if state_path else None
 949
 950    configured_catalog = None
 951    if catalog_path and catalog_path.exists():
 952        catalog_json = catalog_path.read_text()
 953        configured_catalog = ConfiguredAirbyteCatalog.parse_raw(catalog_json)
 954
 955    # Pass log level as an environment variable to the connector container
 956    container_env: dict[str, str] = {}
 957    if enable_debug_logs:
 958        container_env["LOG_LEVEL"] = "DEBUG"
 959
 960    execution_inputs = ExecutionInputs(
 961        connector_under_test=connector,
 962        command=command,
 963        output_dir=output_dir,
 964        config=config,
 965        configured_catalog=configured_catalog,
 966        state=state,
 967        environment_variables=container_env or None,
 968    )
 969
 970    runner = ConnectorRunner(execution_inputs, proxy_url=proxy_url)
 971    result = runner.run()
 972
 973    result.save_artifacts(output_dir)
 974
 975    result_dict: dict[str, object] = {
 976        "connector": connector_image,
 977        "command": command.value,
 978        "success": result.success,
 979        "exit_code": result.exit_code,
 980        "stdout_file": str(result.stdout_file_path),
 981        "stderr_file": str(result.stderr_file_path),
 982        "message_counts": {
 983            k.value: v for k, v in result.get_message_count_per_type().items()
 984        },
 985        "record_counts_per_stream": result.get_record_count_per_stream(),
 986    }
 987
 988    # For CHECK, include the parsed connectionStatus from the Airbyte protocol.
 989    # The CDK exits 0 regardless of check outcome; the real pass/fail signal
 990    # lives in the CONNECTION_STATUS message (status=SUCCEEDED or FAILED).
 991    if command == Command.CHECK:
 992        conn_status = result.get_connection_status()
 993        if conn_status is not None:
 994            result_dict["connection_status"] = conn_status.status.value
 995            result_dict["connection_status_message"] = conn_status.message or ""
 996        else:
 997            result_dict["connection_status"] = None
 998            result_dict["connection_status_message"] = ""
 999
1000    return result_dict, ComparableOutputs.from_execution_result(result)
1001
1002
1003def _annotate_run_verdict(command: str, result: dict[str, Any]) -> dict[str, Any]:
1004    """Copy a run result with `success` as the verdict, exit status kept separately.
1005
1006    `ExecutionResult.success` is the container exit status, which for `check` says
1007    nothing about whether the connector could connect -- the CDK exits 0 even when a
1008    check fails. Any payload read as the verdict must therefore not expose the exit
1009    status under `success`, at any nesting level. The raw value stays available as
1010    `exited_cleanly`; for every command but `check` the two are equal.
1011
1012    Returns a copy: callers pass the unannotated originals to the report generator,
1013    which reads `success` as the exit status.
1014    """
1015    return {
1016        **result,
1017        "success": command_result_succeeded(command, result),
1018        "exited_cleanly": result["success"],
1019    }
1020
1021
1022# The checks the output comparisons contribute, named for the row a reviewer
1023# reads in the report and in the step summary.
1024_SPEC_CHECK = "Spec compatibility"
1025_CATALOG_SCHEMA_CHECK = "Catalog schema"
1026_FINAL_STATE_CHECK = "Final state per stream"
1027
1028# How many individual changes a check's one-line summary names before it says
1029# how many more there are. The full list is in the report and in the run's JSON
1030# payload, and the comparators order their findings worst-first, so the ones
1031# named here are the ones worth naming.
1032_MAX_LISTED_CHANGES = 3
1033
1034# How many changes the JSON payload lists per comparison. Same reason as
1035# `_MAX_DIFF_PAYLOAD_CHARS`: the payload is written to `GITHUB_OUTPUT`, and a
1036# reordered `oneOf` on a wide spec produces hundreds of sentences.
1037_MAX_PAYLOAD_CHANGES = 100
1038
1039# How many per-stream entries it lists. A wide source has thousands of streams,
1040# and one line each is enough to exceed the same output budget without a single
1041# diff being attached.
1042_MAX_PAYLOAD_STREAMS = 100
1043
1044# How much of a comparison's detail the run's JSON payload carries. That payload
1045# is written to `GITHUB_OUTPUT`, which is size-limited, and a wide connector's
1046# catalog diff has no natural bound -- so the payload keeps as many stream diffs
1047# as fit and says how many it dropped. The catalogs themselves are saved
1048# artifacts, which is where an unbounded diff belongs.
1049_MAX_DIFF_PAYLOAD_CHARS = 50_000
1050
1051
1052def _compare_outputs(
1053    *,
1054    command: str,
1055    outcome: ComparisonOutcome,
1056    control_outputs: ComparableOutputs,
1057    target_outputs: ComparableOutputs,
1058) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1059    """Compare what the two versions produced, for the commands that produce it.
1060
1061    This is the content check for `spec`, `discover` and `read`: without it their
1062    verdict is the exit code, and a connector that drops a config field, changes
1063    a stream's schema or corrupts the state the next sync resumes from exits 0
1064    all the way to a release. `check` reports its own status, so it is the one
1065    command with nothing to diff here.
1066
1067    Only runs when both commands succeeded. A failed run has nothing to compare,
1068    and reporting "no spec to compare" beside the command failure that caused it
1069    would state the same problem twice, in weaker terms the second time.
1070
1071    A comparator that raises is left to propagate. The comparators normalise what
1072    they are given, so the only ways here are a programming error in a future
1073    caller or a schema nested past the recursion limit -- and the workflow
1074    already has the right vocabulary for both: a step that dies is an
1075    `internal_failure`, reported as infrastructure rather than as a regression,
1076    with the connector's artifacts uploaded either way. Catching it here would
1077    only let a tooling bug read as "the new version broke its spec".
1078
1079    Args:
1080        command: The Airbyte command that was run, as the CLI names it. This is
1081            always `"read"` for a read, including one given a state file: the
1082            `READ` -> `READ_WITH_STATE` upgrade only changes the `Command` the
1083            container is run with.
1084        outcome: The already-evaluated command outcome.
1085        control_outputs: What the control version produced.
1086        target_outputs: What the target version produced.
1087
1088    Returns:
1089        The checks to fold into the verdict, the diff blocks that show what they
1090        found, and the comparison detail for the run's JSON payload -- all empty
1091        when there was nothing to compare.
1092    """
1093    if not outcome.success:
1094        return (), (), {}
1095
1096    if command == "spec":
1097        return _spec_comparison(control_outputs.spec, target_outputs.spec)
1098
1099    if command == "discover":
1100        return _catalog_comparison(control_outputs.catalog, target_outputs.catalog)
1101
1102    if command == "read":
1103        return _final_state_comparison(
1104            control_outputs.final_states, target_outputs.final_states
1105        )
1106
1107    return (), (), {}
1108
1109
1110def _spec_comparison(
1111    control_spec: Any, target_spec: Any
1112) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1113    """Run the spec comparison and shape it for every surface."""
1114    result = compare_specs(control_spec, target_spec)
1115
1116    check = CheckResult(
1117        name=_SPEC_CHECK,
1118        passed=result.passed,
1119        summary=_change_summary(
1120            result.message,
1121            # A passing spec check still says what was added: an additive change
1122            # is the thing a reviewer most wants to see confirmed.
1123            result.errors if result.errors else result.warnings,
1124        ),
1125        details=tuple(result.errors),
1126    )
1127    # The incompatible changes are the check's `details`, which the report renders
1128    # under the check row and the step summary lists -- a block would repeat them.
1129    # The compatible ones have nowhere else to go, and a reviewer asking "what
1130    # else moved?" is asking about exactly those.
1131    blocks = tuple(
1132        block
1133        for block in (
1134            _change_block("Spec — compatible changes", result.warnings, "add"),
1135        )
1136        if block is not None
1137    )
1138
1139    return (check,), blocks, {"spec": _spec_detail(result)}
1140
1141
1142def _catalog_comparison(
1143    control_catalog: Any, target_catalog: Any
1144) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1145    """Run the discovered-catalog comparison and shape it for every surface."""
1146    result = compare_catalog_schemas(control_catalog, target_catalog)
1147
1148    check = CheckResult(
1149        name=_CATALOG_SCHEMA_CHECK,
1150        passed=result.passed,
1151        # A catalog that only grew -- new streams, new fields, nothing removed
1152        # or retyped -- is reported and does not gate the release. Connectors
1153        # add streams constantly, and a check that goes red on every one of them
1154        # is a check people learn to click past.
1155        severity="diagnostic" if result.additive_only else "strict",
1156        summary=_change_summary(result.message, result.errors),
1157        details=tuple(result.errors),
1158    )
1159
1160    # The findings themselves are the check's `details`, which the report renders
1161    # under the check row: a stream that is missing or new has no schema diff, so
1162    # that list is the only place it appears.
1163    blocks = _stream_diff_blocks(
1164        result,
1165        block_title="Schema diff",
1166        omission_title="Schema diffs omitted for size",
1167        artifact_hint=(
1168            "Both discovered catalogs are in the artifact, under "
1169            "airbyte_messages/catalog.jsonl."
1170        ),
1171    )
1172
1173    return (check,), blocks, {"catalog_schema": _catalog_schema_detail(result)}
1174
1175
1176def _final_state_comparison(
1177    control_states: dict[tuple[str | None, str], Any] | None,
1178    target_states: dict[tuple[str | None, str], Any] | None,
1179) -> tuple[tuple[CheckResult, ...], tuple[DiffBlock, ...], dict[str, Any]]:
1180    """Run the final-state comparison and shape it for every surface."""
1181    result = compare_final_states(control_states, target_states)
1182
1183    # Structural findings first, then the value-only ones: the check lists
1184    # everything it found, and the summary quotes the front of that list.
1185    findings = [*result.errors, *result.warnings]
1186
1187    check = CheckResult(
1188        name=_FINAL_STATE_CHECK,
1189        passed=result.passed,
1190        # Two ways a non-passing state check does not gate. A cursor that only
1191        # advanced: until HTTP replay lands, the two versions call the live API
1192        # in separate runs and a timestamp cursor moves between them on its own.
1193        # And a read with no state on either side, where there was nothing to
1194        # compare rather than something wrong.
1195        severity=(
1196            "diagnostic" if result.value_only or result.inconclusive else "strict"
1197        ),
1198        summary=_change_summary(result.message, findings),
1199        details=tuple(findings),
1200    )
1201
1202    blocks = _stream_diff_blocks(
1203        result,
1204        block_title="Final state diff",
1205        omission_title="Final state diffs omitted for size",
1206        artifact_hint=(
1207            "Both versions' state messages are in the artifact, under "
1208            "airbyte_messages/state.jsonl."
1209        ),
1210    )
1211
1212    return (check,), blocks, {"final_state": _final_state_detail(result)}
1213
1214
1215def _stream_diff_blocks(
1216    result: ComparisonResult,
1217    *,
1218    block_title: str,
1219    omission_title: str,
1220    artifact_hint: str,
1221) -> tuple[DiffBlock, ...]:
1222    """One block per changed stream, within the report's diff budget.
1223
1224    In comparison order -- only a *changed* stream has a diff, so there is no
1225    severity left to sort by here -- and within the same budget the payload uses:
1226    a report that inlines every schema of a 200-stream source is not a report
1227    anyone reads. What the budget drops is named rather than skipped, since a
1228    block that is simply absent reads as "there was nothing to see".
1229    """
1230    blocks: list[DiffBlock] = []
1231    budget = _MAX_DIFF_PAYLOAD_CHARS
1232    dropped: list[str] = []
1233
1234    for name in result.failed_streams:
1235        stream = result.stream_results[name]
1236        if not stream.schema_diff:
1237            continue
1238
1239        payload = json.dumps(stream.schema_diff, indent=2, sort_keys=True)
1240        if len(payload) > budget:
1241            dropped.append(name)
1242            continue
1243
1244        budget -= len(payload)
1245        blocks.append(build_json_block(f"{block_title} — {name}", payload))
1246
1247    if dropped:
1248        blocks.append(
1249            build_diff_block(
1250                omission_title,
1251                [
1252                    DiffLine(
1253                        kind="meta",
1254                        text=(
1255                            "These streams changed, and their diffs were too "
1256                            f"large for this report. {artifact_hint}"
1257                        ),
1258                    ),
1259                    *(DiffLine(kind="del", text=name) for name in dropped),
1260                ],
1261                language="text",
1262            )
1263        )
1264
1265    return tuple(blocks)
1266
1267
1268def _change_block(title: str, changes: list[str], kind: str) -> DiffBlock | None:
1269    """One diff block listing a comparison's findings, or `None` if there are none."""
1270    if not changes:
1271        return None
1272
1273    return build_diff_block(
1274        title,
1275        [DiffLine(kind=kind, text=change) for change in changes],
1276        language="text",
1277    )
1278
1279
1280def _change_summary(message: str, changes: list[str]) -> str:
1281    """One line: what the comparison concluded, and the first few reasons."""
1282    if not changes:
1283        return message
1284
1285    listed = "; ".join(changes[:_MAX_LISTED_CHANGES])
1286    remaining = len(changes) - _MAX_LISTED_CHANGES
1287    if remaining > 0:
1288        listed = f"{listed}; and {remaining} more"
1289
1290    return f"{message}: {listed}"
1291
1292
1293def _bounded_changes(changes: list[str]) -> list[str]:
1294    """A change list the JSON payload can carry, saying what it left out."""
1295    if len(changes) <= _MAX_PAYLOAD_CHANGES:
1296        return changes
1297
1298    dropped = len(changes) - _MAX_PAYLOAD_CHANGES
1299
1300    return [
1301        *changes[:_MAX_PAYLOAD_CHANGES],
1302        f"... and {dropped} more, in the report and the console output",
1303    ]
1304
1305
1306def _spec_detail(result: ComparisonResult) -> dict[str, Any]:
1307    """The spec comparison, for the run's JSON payload."""
1308    return {
1309        "passed": result.passed,
1310        "message": result.message,
1311        "breaking_changes": _bounded_changes(result.errors),
1312        "compatible_changes": _bounded_changes(result.warnings),
1313    }
1314
1315
1316def _catalog_schema_detail(result: ComparisonResult) -> dict[str, Any]:
1317    """The catalog comparison, for the run's JSON payload."""
1318    streams, omitted, dropped = _bounded_stream_diffs(
1319        result,
1320        omission_note="Too large for this payload; diff the catalogs",
1321        artifact_hint="Both discovered catalogs are in the saved artifacts.",
1322    )
1323
1324    detail: dict[str, Any] = {
1325        "passed": result.passed,
1326        # Why a `passed: false` did not fail the run: everything found was an
1327        # addition, which is reported as a warning rather than a regression.
1328        "additive_only": result.additive_only,
1329        "message": result.message,
1330        "changed_streams": _bounded_changes(result.failed_streams),
1331        "streams": streams,
1332    }
1333    if omitted:
1334        detail["streams_with_omitted_diffs"] = omitted
1335    if dropped:
1336        detail["streams_omitted"] = dropped
1337
1338    return detail
1339
1340
1341def _final_state_detail(result: ComparisonResult) -> dict[str, Any]:
1342    """The final-state comparison, for the run's JSON payload."""
1343    streams, omitted, dropped = _bounded_stream_diffs(
1344        result,
1345        omission_note="Too large for this payload; diff the state messages",
1346        artifact_hint="Both versions' state messages are in the saved artifacts.",
1347    )
1348
1349    detail: dict[str, Any] = {
1350        "passed": result.passed,
1351        # The two reasons a `passed: false` did not fail the run, kept apart so
1352        # the payload never says "every state that moved moved in value only"
1353        # about a read where nothing moved because nothing was compared.
1354        "value_only": result.value_only,
1355        "inconclusive": result.inconclusive,
1356        "message": result.message,
1357        "changed_streams": _bounded_changes(result.failed_streams),
1358        "streams": streams,
1359    }
1360    if omitted:
1361        detail["streams_with_omitted_diffs"] = omitted
1362    if dropped:
1363        detail["streams_omitted"] = dropped
1364
1365    return detail
1366
1367
1368def _bounded_stream_diffs(
1369    result: ComparisonResult,
1370    *,
1371    omission_note: str,
1372    artifact_hint: str,
1373) -> tuple[dict[str, Any], list[str], int]:
1374    """A comparison's per-stream verdicts and diffs, sized for the JSON payload.
1375
1376    Every stream that changed is named with its verdict -- so the payload says
1377    what moved, not just that something did -- while the diffs themselves are
1378    kept only while they fit in `_MAX_DIFF_PAYLOAD_CHARS`, and the number of
1379    entries is capped in turn: a thousand-stream source would otherwise put a
1380    thousand of them into `GITHUB_OUTPUT`. Both bounds say what they dropped,
1381    on the console and in the payload: a bound nobody mentions reads as "there
1382    was nothing more to see".
1383    """
1384    streams: dict[str, Any] = {}
1385    omitted: list[str] = []
1386    budget = _MAX_DIFF_PAYLOAD_CHARS
1387
1388    # Non-passing first, so the cap falls on the streams with nothing to say.
1389    ordered = sorted(result.stream_results.items(), key=lambda item: item[1].passed)
1390    kept, dropped_entries = (
1391        ordered[:_MAX_PAYLOAD_STREAMS],
1392        ordered[_MAX_PAYLOAD_STREAMS:],
1393    )
1394
1395    for name, stream in kept:
1396        entry: dict[str, Any] = {"passed": stream.passed, "message": stream.message}
1397        streams[name] = entry
1398
1399        if not stream.schema_diff:
1400            continue
1401
1402        rendered = len(json.dumps(stream.schema_diff, separators=(",", ":")))
1403        if rendered > budget:
1404            omitted.append(name)
1405            entry["diff_omitted"] = omission_note
1406            continue
1407
1408        budget -= rendered
1409        entry["diff"] = stream.schema_diff
1410
1411    if omitted:
1412        print_warning(
1413            f"Comparison detail for {len(omitted)} streams was too large for the "
1414            f"run's JSON output ({', '.join(omitted)}); their diffs are omitted "
1415            f"there. {artifact_hint}"
1416        )
1417
1418    if dropped_entries:
1419        print_warning(
1420            f"The run's JSON output lists {_MAX_PAYLOAD_STREAMS} of "
1421            f"{len(ordered)} streams, changed ones first. The report carries "
1422            f"them all. {artifact_hint}"
1423        )
1424
1425    return streams, omitted, len(dropped_entries)
1426
1427
1428def _build_connector_image_from_source(
1429    connector_name: str,
1430    repo_root: Path | None = None,
1431    tag: str = "dev",
1432) -> str | None:
1433    """Build a connector image from source code.
1434
1435    Args:
1436        connector_name: Name of the connector (e.g., 'source-pokeapi').
1437        repo_root: Optional path to the airbyte repo root. If not provided,
1438            will attempt to auto-detect from current directory.
1439        tag: Tag to apply to the built image (default: 'dev').
1440
1441    Returns:
1442        The full image name with tag if successful, None if build fails.
1443    """
1444    if not verify_docker_installation():
1445        print_error("Docker is not installed or not running")
1446        return None
1447
1448    try:
1449        connector_directory = find_connector_root_from_name(connector_name)
1450    except FileNotFoundError:
1451        if repo_root:
1452            connector_directory = repo_root / CONNECTORS_SUBDIR / connector_name
1453            if not connector_directory.exists():
1454                print_error(f"Connector directory not found: {connector_directory}")
1455                return None
1456        else:
1457            print_error(
1458                f"Could not find connector '{connector_name}'. "
1459                "Try providing --repo-root to specify the airbyte repo location."
1460            )
1461            return None
1462
1463    metadata_file_path = connector_directory / "metadata.yaml"
1464    if not metadata_file_path.exists():
1465        print_error(f"metadata.yaml not found at {metadata_file_path}")
1466        return None
1467
1468    metadata = MetadataFile.from_file(metadata_file_path)
1469    print_success(f"Building image for connector: {connector_name}")
1470
1471    built_image = build_connector_image(
1472        connector_name=connector_name,
1473        connector_directory=connector_directory,
1474        metadata=metadata,
1475        tag=tag,
1476        no_verify=False,
1477    )
1478    print_success(f"Successfully built image: {built_image}")
1479    return built_image
1480
1481
1482def _fetch_control_image_from_metadata(connector_name: str) -> str | None:
1483    """Fetch the current released connector image from metadata.yaml on main branch.
1484
1485    This fetches the connector's metadata.yaml from the airbyte monorepo's master branch
1486    and extracts the dockerRepository and dockerImageTag to construct the control image.
1487
1488    Args:
1489        connector_name: The connector name (e.g., 'source-github').
1490
1491    Returns:
1492        The full connector image with tag (e.g., 'airbyte/source-github:1.0.0'),
1493        or None if the metadata could not be fetched or parsed.
1494    """
1495    metadata_url = (
1496        f"https://raw.githubusercontent.com/airbytehq/airbyte/master/"
1497        f"airbyte-integrations/connectors/{connector_name}/metadata.yaml"
1498    )
1499    response = requests.get(metadata_url, timeout=30)
1500    if not response.ok:
1501        print_error(
1502            f"Failed to fetch metadata for {connector_name}: "
1503            f"HTTP {response.status_code} from {metadata_url}"
1504        )
1505        return None
1506
1507    metadata = yaml.safe_load(response.text)
1508    if not isinstance(metadata, dict):
1509        print_error(f"Invalid metadata format for {connector_name}: expected dict")
1510        return None
1511
1512    data = metadata.get("data", {})
1513    docker_repository = data.get("dockerRepository")
1514    docker_image_tag = data.get("dockerImageTag")
1515
1516    if not docker_repository or not docker_image_tag:
1517        print_error(
1518            f"Could not find dockerRepository/dockerImageTag in metadata for {connector_name}"
1519        )
1520        return None
1521
1522    return f"{docker_repository}:{docker_image_tag}"
1523
1524
1525def _run_with_optional_http_metrics(
1526    connector_image: str,
1527    command: Command,
1528    output_dir: Path,
1529    target_or_control: TargetOrControl,
1530    enable_http_metrics: bool,
1531    config_path: Path | None,
1532    catalog_path: Path | None,
1533    state_path: Path | None,
1534    enable_debug_logs: bool = False,
1535) -> tuple[dict, ComparableOutputs]:
1536    """Run a connector command with optional HTTP metrics capture.
1537
1538    When enable_http_metrics is True, starts mitmproxy to capture HTTP traffic.
1539    If mitmproxy fails to start, falls back to running without metrics.
1540
1541    Args:
1542        connector_image: Full connector image name with tag.
1543        command: The Airbyte command to run.
1544        output_dir: Directory to store output files.
1545        target_or_control: Whether this is target or control version.
1546        enable_http_metrics: Whether to capture HTTP metrics via mitmproxy.
1547        config_path: Path to connector config JSON file.
1548        catalog_path: Path to configured catalog JSON file.
1549        state_path: Path to state JSON file.
1550        enable_debug_logs: If `True`, pass `LOG_LEVEL=DEBUG` to the connector container.
1551
1552    Returns:
1553        What `_run_connector_command` returns, with `http_metrics` added to the
1554        results dict when they were captured.
1555    """
1556    if not enable_http_metrics:
1557        return _run_connector_command(
1558            connector_image=connector_image,
1559            command=command,
1560            output_dir=output_dir,
1561            target_or_control=target_or_control,
1562            config_path=config_path,
1563            catalog_path=catalog_path,
1564            state_path=state_path,
1565            enable_debug_logs=enable_debug_logs,
1566        )
1567
1568    with MitmproxyManager.start(output_dir) as session:
1569        if session is None:
1570            print_error("Mitmproxy unavailable, running without HTTP metrics")
1571            return _run_connector_command(
1572                connector_image=connector_image,
1573                command=command,
1574                output_dir=output_dir,
1575                target_or_control=target_or_control,
1576                config_path=config_path,
1577                catalog_path=catalog_path,
1578                state_path=state_path,
1579                enable_debug_logs=enable_debug_logs,
1580            )
1581
1582        print_success(f"Started mitmproxy on {session.proxy_url}")
1583        result, comparable = _run_connector_command(
1584            connector_image=connector_image,
1585            command=command,
1586            output_dir=output_dir,
1587            target_or_control=target_or_control,
1588            config_path=config_path,
1589            catalog_path=catalog_path,
1590            state_path=state_path,
1591            proxy_url=session.proxy_url,
1592            enable_debug_logs=enable_debug_logs,
1593        )
1594
1595        http_metrics = parse_http_dump(session.dump_file_path)
1596        result["http_metrics"] = {
1597            "flow_count": http_metrics.flow_count,
1598            "duplicate_flow_count": http_metrics.duplicate_flow_count,
1599            # Carried into the report because it is the number that tells a
1600            # reviewer whether a record difference is a real regression or
1601            # upstream drift. Today it is a duplicate-URL heuristic; it only
1602            # becomes exact once HTTP replay lands.
1603            "cache_hits_count": http_metrics.cache_hits_count,
1604            "cache_hit_ratio": http_metrics.cache_hit_ratio,
1605        }
1606        print_success(
1607            f"Captured {http_metrics.flow_count} HTTP flows "
1608            f"({http_metrics.duplicate_flow_count} duplicates)"
1609        )
1610        return result, comparable
1611
1612
1613def _compare_read_records(
1614    target_result: dict,
1615    control_result: dict,
1616    catalog_file: Path | None,
1617) -> tuple[RecordComparisonSummary | None, str | None]:
1618    """Run record-level comparisons between control and target read outputs.
1619
1620    Uses primary keys from the configured catalog (source-defined primary
1621    keys come from the discovered catalog) to compare record counts, PK
1622    presence, PK integrity, and field values. Streams without a primary key
1623    fall back to a directional record count comparison.
1624
1625    Each run's stdout is first split into per-stream record files (a single
1626    streaming pass), and the comparison then loads one stream at a time, so
1627    memory stays bounded by the largest stream instead of the full output
1628    of both runs. The per-stream files land next to each run's stdout and
1629    double as debugging artifacts. Within one stream, both sides are
1630    materialized as parsed messages plus dict copies plus PK indexes --
1631    measured at roughly 13 bytes resident per byte of record JSON, so a
1632    single stream of ~500 MB of records is the practical ceiling on a
1633    standard 7 GB runner. Start here when debugging an OOM.
1634
1635    Returns:
1636        `(summary, skip_reason)`. The summary is None when the comparison
1637        cannot run at all -- then `skip_reason` says why, so the caller can
1638        surface a diagnostic check instead of letting the run read as a pass
1639        that compared records. An unexpected error during the comparison
1640        itself returns a FAILED summary instead of raising: by this point
1641        both (expensive, live-API) runs have completed, and a comparison bug
1642        must fail the verdict, not destroy the run's outputs and reports.
1643    """
1644    if catalog_file is None or not catalog_file.exists():
1645        reason = "no configured catalog available"
1646        print_warning(f"Skipping record-level comparison: {reason}")
1647        return None, reason
1648
1649    try:
1650        configured_catalog = ConfiguredAirbyteCatalog.model_validate_json(
1651            catalog_file.read_text()
1652        )
1653    except Exception as exc:
1654        reason = f"configured catalog could not be parsed ({exc})"
1655        print_warning(f"Skipping record-level comparison: {reason}")
1656        return None, reason
1657
1658    try:
1659        primary_keys = get_primary_keys_per_stream(configured_catalog)
1660        # A stream name that repeats across namespaces merges two streams in
1661        # this name-keyed pipeline; those names degraded to count-only above,
1662        # and the reader must hear why from the summary, not just stdout.
1663        collision_warnings = [
1664            f"Stream name {name!r} appears in more than one namespace; the "
1665            "record comparison keys streams by name, so its records are "
1666            "compared by count only"
1667            for name in duplicate_stream_names(configured_catalog)
1668        ]
1669        control_record_files = split_records_per_stream(
1670            Path(control_result["stdout_file"])
1671        )
1672        target_record_files = split_records_per_stream(
1673            Path(target_result["stdout_file"])
1674        )
1675
1676        summary = run_record_comparisons_from_files(
1677            control_record_files=control_record_files,
1678            target_record_files=target_record_files,
1679            primary_keys_per_stream=primary_keys,
1680        )
1681        summary.warnings[:0] = collision_warnings
1682    except Exception as exc:
1683        summary = RecordComparisonSummary(
1684            passed=False,
1685            errored=True,
1686            count_comparison=ComparisonResult(
1687                passed=False, message="Record comparison errored"
1688            ),
1689            errors=[f"Record comparison errored: {exc!r}"],
1690        )
1691
1692    if summary.passed:
1693        print_success("Record comparison passed (counts, PKs, field values)")
1694    else:
1695        print_error("Record comparison failed:")
1696        for error in summary.errors[:20]:
1697            print_error(f"  - {error}")
1698    for warning in summary.warnings[:20]:
1699        print_warning(warning)
1700
1701    return summary, None
1702
1703
1704@connector_app.command(name="regression-test")
1705def regression_test(
1706    skip_compare: Annotated[
1707        bool,
1708        Parameter(
1709            help="If True, skip comparison and run single-version tests only. "
1710            "If False (default), run comparison tests (target vs control)."
1711        ),
1712    ] = False,
1713    skip_record_comparison: Annotated[
1714        bool,
1715        Parameter(
1716            help="If True, skip the record-level comparison for read commands "
1717            "(counts, PK presence, PK integrity, field values) and gate the "
1718            "verdict on command outcomes only. Escape hatch for sources whose "
1719            "data legitimately changes between the two live runs (rolling "
1720            "windows, feeds that age out records) until HTTP request caching "
1721            "makes strict comparison reliable."
1722        ),
1723    ] = False,
1724    test_image: Annotated[
1725        str | None,
1726        Parameter(
1727            help="Test connector image with tag (e.g., airbyte/source-github:1.0.0). "
1728            "This is the image under test - in comparison mode, it's compared against control_image."
1729        ),
1730    ] = None,
1731    control_image: Annotated[
1732        str | None,
1733        Parameter(
1734            help="Control connector image (baseline version) with tag (e.g., airbyte/source-github:1.0.0). "
1735            "Ignored if `skip_compare=True`."
1736        ),
1737    ] = None,
1738    connector_name: Annotated[
1739        str | None,
1740        Parameter(
1741            help="Connector name to build image from source (e.g., 'source-pokeapi'). "
1742            "If provided, builds the image locally with tag 'dev'. "
1743            "For comparison tests (default), this builds the target image. "
1744            "For single-version tests (skip_compare=True), this builds the test image."
1745        ),
1746    ] = None,
1747    repo_root: Annotated[
1748        str | None,
1749        Parameter(
1750            help="Path to the airbyte repo root. Required if connector_name is provided "
1751            "and the repo cannot be auto-detected."
1752        ),
1753    ] = None,
1754    command: Annotated[
1755        Literal["spec", "check", "discover", "read"],
1756        Parameter(help="The Airbyte command to run."),
1757    ] = "check",
1758    connection_id: Annotated[
1759        str | None,
1760        Parameter(
1761            help="Airbyte Cloud connection ID to fetch config/catalog from. "
1762            "Mutually exclusive with config-path/catalog-path. "
1763            "If provided, test_image/control_image can be auto-detected."
1764        ),
1765    ] = None,
1766    config_path: Annotated[
1767        str | None,
1768        Parameter(help="Path to the connector config JSON file."),
1769    ] = None,
1770    catalog_path: Annotated[
1771        str | None,
1772        Parameter(help="Path to the configured catalog JSON file (required for read)."),
1773    ] = None,
1774    state_path: Annotated[
1775        str | None,
1776        Parameter(help="Path to the state JSON file (optional for read)."),
1777    ] = None,
1778    output_dir: Annotated[
1779        str,
1780        Parameter(help="Directory to store test artifacts."),
1781    ] = "/tmp/regression_test_artifacts",
1782    enable_http_metrics: Annotated[
1783        bool,
1784        Parameter(
1785            help="Capture HTTP traffic metrics via mitmproxy (experimental). "
1786            "Requires mitmdump to be installed. Only used in comparison mode."
1787        ),
1788    ] = False,
1789    selected_streams: Annotated[
1790        str | None,
1791        Parameter(
1792            help="Comma-separated list of stream names to include in the read. "
1793            "Only these streams will be included in the configured catalog. "
1794            "This is useful to limit data volume by testing only specific streams."
1795        ),
1796    ] = None,
1797    enable_debug_logs: Annotated[
1798        bool,
1799        Parameter(
1800            help="Enable debug-level logging for regression test output. "
1801            "Also passed as `LOG_LEVEL=DEBUG` to the connector Docker container."
1802        ),
1803    ] = False,
1804    with_state: Annotated[
1805        bool | None,
1806        Parameter(
1807            negative="--no-state",
1808            help="Fetch and pass the connection's current state to the read command, "
1809            "producing a warm read instead of a cold read. Defaults to `True` when "
1810            "`--connection-id` is provided, `False` otherwise. Has no effect unless "
1811            "the command is `read`. Ignored when `--state-path` is explicitly provided.",
1812        ),
1813    ] = None,
1814) -> None:
1815    """Run regression tests on connectors.
1816
1817    This command supports two modes:
1818
1819    Comparison mode (skip_compare=False, default):
1820        Runs the specified Airbyte protocol command against both the target (new)
1821        and control (baseline) connector versions, then compares the results.
1822        This helps identify regressions between versions.
1823
1824    Single-version mode (skip_compare=True):
1825        Runs the specified Airbyte protocol command against a single connector
1826        and validates the output. No comparison is performed.
1827
1828    Results are written to the output directory and to GitHub Actions outputs
1829    if running in CI.
1830
1831    You can provide the test image in three ways:
1832    1. --test-image: Use a pre-built image from Docker registry
1833    2. --connector-name: Build the image locally from source code
1834    3. --connection-id: Auto-detect from an Airbyte Cloud connection
1835
1836    You can provide config/catalog either via file paths OR via a connection_id
1837    that fetches them from Airbyte Cloud.
1838    """
1839    # Configure debug logging for the regression test harness when requested
1840    if enable_debug_logs:
1841        logging.basicConfig(
1842            level=logging.DEBUG,
1843            format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
1844            force=True,
1845        )
1846
1847    output_path = Path(output_dir)
1848    output_path.mkdir(parents=True, exist_ok=True)
1849
1850    cmd = Command(command)
1851
1852    config_file: Path | None = None
1853    catalog_file: Path | None = None
1854    state_file = Path(state_path) if state_path else None
1855
1856    selected_stream_names: set[str] = (
1857        {s.strip() for s in selected_streams.split(",") if s.strip()}
1858        if selected_streams
1859        else set()
1860    )
1861
1862    # Resolve the test image (used in both single-version and comparison modes)
1863    resolved_test_image: str | None = test_image
1864    resolved_control_image: str | None = control_image
1865
1866    # Validate conflicting parameters
1867    # Single-version mode: reject comparison-specific parameters
1868    if skip_compare and control_image:
1869        write_github_output("success", False)
1870        write_github_output(
1871            "error", "Cannot specify control_image with skip_compare=True"
1872        )
1873        exit_with_error(
1874            "Cannot specify --control-image with --skip-compare. "
1875            "Control image is only used in comparison mode."
1876        )
1877
1878    # If connector_name is provided, build the image from source
1879    if connector_name:
1880        if resolved_test_image:
1881            write_github_output("success", False)
1882            write_github_output(
1883                "error", "Cannot specify both test_image and connector_name"
1884            )
1885            exit_with_error("Cannot specify both --test-image and --connector-name")
1886
1887        repo_root_path = Path(repo_root) if repo_root else None
1888        built_image = _build_connector_image_from_source(
1889            connector_name=connector_name,
1890            repo_root=repo_root_path,
1891            tag="dev",
1892        )
1893        if not built_image:
1894            write_github_output("success", False)
1895            write_github_output("error", f"Failed to build image for {connector_name}")
1896            exit_with_error(f"Failed to build image for {connector_name}")
1897        resolved_test_image = built_image
1898
1899    if connection_id:
1900        if config_path or catalog_path:
1901            write_github_output("success", False)
1902            write_github_output(
1903                "error", "Cannot specify both connection_id and file paths"
1904            )
1905            exit_with_error(
1906                "Cannot specify both connection_id and config_path/catalog_path"
1907            )
1908
1909        print_success(f"Fetching config/catalog from connection: {connection_id}")
1910        connection_data = fetch_connection_data(connection_id)
1911
1912        # Check if we should retrieve unmasked secrets
1913        if should_use_secret_retriever():
1914            print_success(
1915                "USE_CONNECTION_SECRET_RETRIEVER enabled - enriching config with unmasked secrets..."
1916            )
1917            try:
1918                connection_data = enrich_config_with_secrets(
1919                    connection_data,
1920                    retrieval_reason="Regression test with USE_CONNECTION_SECRET_RETRIEVER=true",
1921                )
1922                print_success("Successfully retrieved unmasked secrets from database")
1923            except SecretRetrievalError as e:
1924                write_github_output("success", False)
1925                write_github_output("error", str(e))
1926                exit_with_error(
1927                    f"{e}\n\n"
1928                    f"This connection cannot be used for regression testing. "
1929                    f"Please use a connection from a non-EU workspace, or use GSM-based "
1930                    f"integration test credentials instead (by omitting --connection-id)."
1931                )
1932
1933        # Fill in stream schemas: the minimal catalog from the public API has
1934        # empty json_schema values, which silently yields zero records for
1935        # connectors that derive request fields from the catalog schema.
1936        # Only the read command consumes the catalog, so spec/check/discover
1937        # skip the extra Cloud API round trip.
1938        enrichment_state: list[dict[str, Any]] | None = None
1939        enrichment_state_fetched = False
1940        if cmd.needs_catalog():
1941            merged_count = 0
1942            try:
1943                enrichment = enrich_catalog_schemas_from_cloud(connection_data)
1944                connection_data = enrichment.connection_data
1945                merged_count = enrichment.merged_count
1946                enrichment_state = enrichment.state
1947                enrichment_state_fetched = True
1948            except Exception as exc:
1949                print_error(f"Failed to fetch stream schemas from Cloud API: {exc}")
1950            missing_schemas = streams_missing_schemas(connection_data.catalog)
1951            if selected_stream_names:
1952                missing_schemas = [
1953                    s for s in missing_schemas if s in selected_stream_names
1954                ]
1955            if missing_schemas and should_use_secret_retriever():
1956                try:
1957                    connection_data, db_merged_count = (
1958                        enrich_catalog_schemas_from_platform_db(
1959                            connection_data,
1960                            retrieval_reason="Regression test catalog schema enrichment",
1961                        )
1962                    )
1963                    merged_count += db_merged_count
1964                except Exception as exc:
1965                    print_error(f"Failed to merge stored stream schemas: {exc}")
1966                missing_schemas = streams_missing_schemas(connection_data.catalog)
1967                if selected_stream_names:
1968                    missing_schemas = [
1969                        s for s in missing_schemas if s in selected_stream_names
1970                    ]
1971            total_streams = len(connection_data.catalog.get("streams", []))
1972            checked_streams = (
1973                f"{len(missing_schemas)} of {len(selected_stream_names)} "
1974                "selected stream(s)"
1975                if selected_stream_names
1976                else f"{len(missing_schemas)} of {total_streams} stream(s)"
1977            )
1978            if not total_streams:
1979                error_msg = (
1980                    "No configured streams found in the connection catalog. "
1981                    "A read with zero streams would produce zero records on "
1982                    "both versions, so failing instead of producing a false pass."
1983                )
1984                write_github_output("success", False)
1985                write_github_output("error", error_msg)
1986                exit_with_error(error_msg)
1987            if missing_schemas:
1988                error_msg = (
1989                    f"Stream schemas missing for {checked_streams}: "
1990                    f"{', '.join(missing_schemas)}. "
1991                    "Connectors that derive request fields from the catalog "
1992                    "schema would silently return zero records, so failing "
1993                    "instead of producing a false pass."
1994                )
1995                write_github_output("success", False)
1996                write_github_output("error", error_msg)
1997                exit_with_error(error_msg)
1998            print_success(
1999                f"Stream schemas ready: merged {merged_count} of {total_streams} "
2000                "stream(s)"
2001                + (
2002                    f"; checked {len(selected_stream_names)} selected stream(s)"
2003                    if selected_stream_names
2004                    else ""
2005                )
2006            )
2007
2008        # Resolve with_state default: True when connection_id is provided
2009        resolved_with_state = with_state if with_state is not None else True
2010
2011        if not (resolved_with_state and not state_file and command == "read"):
2012            connection_data.state = None
2013        elif enrichment_state_fetched:
2014            connection_data.state = enrichment_state
2015            if connection_data.state:
2016                print_success(
2017                    f"Using state fetched during schema enrichment "
2018                    f"({len(connection_data.state)} stream(s))"
2019                )
2020            else:
2021                print_success("No state available for this connection (initial sync)")
2022        else:
2023            print_success("Fetching connection state for warm read...")
2024            try:
2025                connection_data.state = fetch_connection_state(connection_data)
2026                if connection_data.state:
2027                    print_success(
2028                        f"Fetched state with {len(connection_data.state)} stream(s)"
2029                    )
2030                else:
2031                    connection_data.state = None
2032                    print_success(
2033                        "No state available for this connection (initial sync)"
2034                    )
2035            except Exception as exc:
2036                print_error(f"Failed to fetch connection state: {exc}")
2037                print_success("Falling back to cold read (no state)")
2038
2039        config_file, catalog_file, state_file_from_conn = save_connection_data_to_files(
2040            connection_data, output_path / "connection_data"
2041        )
2042
2043        if state_file_from_conn and not state_file:
2044            state_file = state_file_from_conn
2045
2046        print_success(
2047            f"Fetched config for source: {connection_data.source_name} "
2048            f"with {len(connection_data.stream_names)} streams"
2049        )
2050
2051        # Auto-detect test/control image from connection if not provided
2052        if not resolved_test_image and connection_data.connector_image:
2053            resolved_test_image = connection_data.connector_image
2054            print_success(f"Auto-detected test image: {resolved_test_image}")
2055
2056        if (
2057            not skip_compare
2058            and not resolved_control_image
2059            and connection_data.connector_image
2060        ):
2061            resolved_control_image = connection_data.connector_image
2062            print_success(f"Auto-detected control image: {resolved_control_image}")
2063    elif config_path:
2064        config_file = Path(config_path)
2065        catalog_file = Path(catalog_path) if catalog_path else None
2066    elif connector_name:
2067        # Fallback: fetch integration test secrets from GSM using PyAirbyte API
2068        print_success(
2069            f"No connection_id or config_path provided. "
2070            f"Attempting to fetch integration test config from GSM for {connector_name}..."
2071        )
2072        gsm_config = get_first_config_from_secrets(connector_name)
2073        if gsm_config:
2074            # Write config to a temp file (not in output_path to avoid artifact upload)
2075            gsm_config_dir = Path(
2076                tempfile.mkdtemp(prefix=f"gsm-config-{connector_name}-")
2077            )
2078            gsm_config_dir.chmod(0o700)
2079            gsm_config_file = gsm_config_dir / "config.json"
2080            gsm_config_file.write_text(json.dumps(gsm_config, indent=2))
2081            gsm_config_file.chmod(0o600)
2082            config_file = gsm_config_file
2083            # Use catalog_path if provided (e.g., generated from discover output)
2084            catalog_file = Path(catalog_path) if catalog_path else None
2085            print_success(
2086                f"Fetched integration test config from GSM for {connector_name}"
2087            )
2088        else:
2089            print_error(
2090                f"Failed to fetch integration test config from GSM for {connector_name}."
2091            )
2092            config_file = None
2093            # Use catalog_path if provided (e.g., generated from discover output)
2094            catalog_file = Path(catalog_path) if catalog_path else None
2095    else:
2096        config_file = None
2097        catalog_file = Path(catalog_path) if catalog_path else None
2098
2099    # Auto-detect control_image from metadata.yaml if connector_name is provided (comparison mode only)
2100    if not skip_compare and not resolved_control_image and connector_name:
2101        resolved_control_image = _fetch_control_image_from_metadata(connector_name)
2102        if resolved_control_image:
2103            print_success(
2104                f"Auto-detected control image from metadata.yaml: {resolved_control_image}"
2105            )
2106
2107    # Validate that we have the required images
2108    if not resolved_test_image:
2109        write_github_output("success", False)
2110        write_github_output("error", "No test image specified")
2111        exit_with_error(
2112            "You must provide one of the following: a test_image, a connector_name "
2113            "to build the image from source, or a connection_id to auto-detect the image."
2114        )
2115
2116    if not skip_compare and not resolved_control_image:
2117        write_github_output("success", False)
2118        write_github_output("error", "No control image specified")
2119        exit_with_error(
2120            "You must provide one of the following: a control_image, a connection_id "
2121            "for a connection that has an associated connector image, or a connector_name "
2122            "to auto-detect the control image from the airbyte repo's metadata.yaml."
2123        )
2124
2125    # Pull images if they weren't just built locally
2126    # If connector_name was provided, we just built the test image locally
2127    if not connector_name and not ensure_image_available(resolved_test_image):
2128        write_github_output("success", False)
2129        write_github_output("error", f"Failed to pull image: {resolved_test_image}")
2130        exit_with_error(f"Failed to pull test image: {resolved_test_image}")
2131
2132    if (
2133        not skip_compare
2134        and resolved_control_image
2135        and not ensure_image_available(resolved_control_image)
2136    ):
2137        write_github_output("success", False)
2138        write_github_output("error", f"Failed to pull image: {resolved_control_image}")
2139        exit_with_error(
2140            f"Failed to pull control connector image: {resolved_control_image}"
2141        )
2142
2143    # Apply selected_streams filter to catalog if requested
2144    if selected_stream_names and catalog_file:
2145        print_success(
2146            f"Filtering catalog to {len(selected_stream_names)} selected streams: "
2147            f"{', '.join(sorted(selected_stream_names))}"
2148        )
2149        filter_configured_catalog_file(catalog_file, selected_stream_names)
2150
2151    # Upgrade READ → READ_WITH_STATE when a state file is available
2152    if cmd == Command.READ and state_file:
2153        cmd = Command.READ_WITH_STATE
2154        print_success("State available — using read-with-state (warm read)")
2155
2156    # Track telemetry for the regression test
2157    # Extract version from image tag (e.g., "airbyte/source-github:1.0.0" -> "1.0.0")
2158    target_version = (
2159        resolved_test_image.rsplit(":", 1)[-1]
2160        if ":" in resolved_test_image
2161        else "unknown"
2162    )
2163    control_version = None
2164    if resolved_control_image and ":" in resolved_control_image:
2165        control_version = resolved_control_image.rsplit(":", 1)[-1]
2166
2167    # Get tester identity from environment (GitHub Actions sets GITHUB_ACTOR)
2168    tester = os.getenv("GITHUB_ACTOR") or os.getenv("USER")
2169
2170    track_regression_test(
2171        user_id=tester,
2172        connector_image=resolved_test_image,
2173        command=command,
2174        target_version=target_version,
2175        control_version=control_version,
2176        additional_properties={
2177            "connection_id": connection_id,
2178            "skip_compare": skip_compare,
2179            "with_state": cmd == Command.READ_WITH_STATE,
2180        },
2181    )
2182
2183    # Execute the appropriate mode
2184    if skip_compare:
2185        # Single-version mode: run only the connector image. There is no control
2186        # to compare its spec or catalog against, so the declared objects go
2187        # unread here.
2188        result, _ = _run_connector_command(
2189            connector_image=resolved_test_image,
2190            command=cmd,
2191            output_dir=output_path,
2192            target_or_control=TargetOrControl.TARGET,
2193            config_path=config_file,
2194            catalog_path=catalog_file,
2195            state_path=state_file,
2196            enable_debug_logs=enable_debug_logs,
2197        )
2198
2199        # Same per-version rule as comparison mode: for CHECK the exit code alone
2200        # is not the verdict, connectionStatus is.
2201        annotated_result = _annotate_run_verdict(command, result)
2202        single_success = bool(annotated_result["success"])
2203
2204        print_json(annotated_result)
2205
2206        single_outputs: dict[str, object] = {
2207            "success": single_success,
2208            "connector": resolved_test_image,
2209            "command": command,
2210            "exit_code": result["exit_code"],
2211        }
2212        if command == "check":
2213            # Named to match the comparison-mode output the workflow summary reads.
2214            single_outputs["target_connection_status"] = result.get("connection_status")
2215
2216        write_github_outputs(single_outputs)
2217
2218        # Generate report.md with detailed metrics
2219        report_path = generate_single_version_report(
2220            connector_image=resolved_test_image,
2221            command=command,
2222            result=result,
2223            output_dir=output_path,
2224        )
2225        print_success(f"Generated report: {report_path}")
2226
2227        # The detailed surface is report.html in the uploaded artifact; the step
2228        # summary gets the concise block instead of the whole report. Both are
2229        # built from the unannotated result: the model derives the verdict itself.
2230        report_model = build_single_version_report_model(
2231            connector_image=resolved_test_image,
2232            command=command,
2233            result=result,
2234            configured_streams=_configured_stream_names(cmd, catalog_file),
2235            connection_objects=_collect_connection_objects(
2236                catalog_file, state_file, output_path / "airbyte_messages"
2237            ),
2238        )
2239        html_report_path = write_html_report(report_model, output_path)
2240        print_success(f"Generated HTML report: {html_report_path}")
2241
2242        # Write the concise summary to GITHUB_STEP_SUMMARY (if env var exists)
2243        write_github_summary(render_inline_summary(report_model))
2244
2245        if single_success:
2246            print_success(
2247                f"Single-version regression test passed for {resolved_test_image}"
2248            )
2249        elif command == "check" and result["success"]:
2250            print_error(
2251                f"Single-version regression test failed for {resolved_test_image}: "
2252                f"connectionStatus {result.get('connection_status') or 'missing'} "
2253                "(the connector itself exited 0)."
2254            )
2255        else:
2256            print_error(
2257                f"Single-version regression test failed for {resolved_test_image}"
2258            )
2259    else:
2260        # Comparison mode: run both control and target images. The control
2261        # version runs FIRST on purpose: the record comparison is
2262        # directional (extra records in the target are allowed, missing
2263        # ones fail), so data created upstream between the two live runs
2264        # must land in the target output — where it is tolerated — rather
2265        # than in the control output, where it would read as records
2266        # missing from the target and fail the comparison. Accepted side
2267        # effect (applies to every command, not just read): a hanging or
2268        # timing-out control run means the target never runs, so such a
2269        # run says nothing about the version under test.
2270        target_output = output_path / "target"
2271        control_output = output_path / "control"
2272
2273        control_result, control_outputs = _run_with_optional_http_metrics(
2274            connector_image=resolved_control_image,  # type: ignore[arg-type]
2275            command=cmd,
2276            output_dir=control_output,
2277            target_or_control=TargetOrControl.CONTROL,
2278            enable_http_metrics=enable_http_metrics,
2279            config_path=config_file,
2280            catalog_path=catalog_file,
2281            state_path=state_file,
2282            enable_debug_logs=enable_debug_logs,
2283        )
2284
2285        target_result, target_outputs = _run_with_optional_http_metrics(
2286            connector_image=resolved_test_image,
2287            command=cmd,
2288            output_dir=target_output,
2289            target_or_control=TargetOrControl.TARGET,
2290            enable_http_metrics=enable_http_metrics,
2291            config_path=config_file,
2292            catalog_path=catalog_file,
2293            state_path=state_file,
2294            enable_debug_logs=enable_debug_logs,
2295        )
2296
2297        outcome = evaluate_comparison_outcome(command, target_result, control_result)
2298
2299        # Record-level comparison for READ commands (counts, PK presence,
2300        # PK integrity, field values). Only meaningful when both runs
2301        # succeeded; a comparison failure is a regression and fails the
2302        # verdict even though both containers exited cleanly. `outcome`
2303        # itself stays the pure exit-status/connectionStatus layer — the
2304        # folded verdict lives in `success`/`regression_detected` below.
2305        record_comparison: RecordComparisonSummary | None = None
2306        # A skipped comparison must be visible on the report surfaces, not
2307        # only on stdout: without this row the run reads as a pass that
2308        # compared records. Diagnostic, because the skip is not a regression.
2309        record_comparison_skipped: CheckResult | None = None
2310        if (
2311            cmd in (Command.READ, Command.READ_WITH_STATE)
2312            and outcome.both_succeeded
2313            and not skip_record_comparison
2314        ):
2315            record_comparison, skip_reason = _compare_read_records(
2316                target_result=target_result,
2317                control_result=control_result,
2318                catalog_file=catalog_file,
2319            )
2320            if record_comparison is None and skip_reason is not None:
2321                record_comparison_skipped = CheckResult(
2322                    name="Record comparison",
2323                    passed=False,
2324                    severity="diagnostic",
2325                    summary=f"Skipped — {skip_reason}; records were not compared",
2326                )
2327        elif skip_record_comparison and cmd in (Command.READ, Command.READ_WITH_STATE):
2328            print_warning(
2329                "Record-level comparison skipped (--skip-record-comparison); "
2330                "the verdict gates on command outcomes only."
2331            )
2332            record_comparison_skipped = CheckResult(
2333                name="Record comparison",
2334                passed=False,
2335                severity="diagnostic",
2336                summary="Skipped (--skip-record-comparison); records were not compared",
2337            )
2338        record_comparison_failed = (
2339            record_comparison is not None and not record_comparison.passed
2340        )
2341        # An errored comparison fails the run (fail closed) but as
2342        # INCONCLUSIVE, like both_failed: a comparator crash must not tell a
2343        # connector developer their change caused a regression.
2344        comparison_errored = record_comparison is not None and record_comparison.errored
2345
2346        # Spec/catalog-schema comparison of what the two versions declared
2347        # about themselves; a failed check here is a regression as much as
2348        # a dropped record.
2349        (
2350            comparison_checks,
2351            comparison_diff_blocks,
2352            comparison_detail,
2353        ) = _compare_outputs(
2354            command=command,
2355            outcome=outcome,
2356            control_outputs=control_outputs,
2357            target_outputs=target_outputs,
2358        )
2359        failed_comparisons = failed_checks(comparison_checks)
2360
2361        success = (
2362            outcome.success and not record_comparison_failed and not failed_comparisons
2363        )
2364        regression_detected = (
2365            outcome.regression_detected
2366            or (record_comparison_failed and not comparison_errored)
2367            or bool(failed_comparisons)
2368        )
2369
2370        # Annotate copies of each side, never the originals: `generate_regression_report`
2371        # below reads `result["success"]` as the exit status, so handing it a dict whose
2372        # `success` is the verdict would corrupt the report's per-version wording.
2373        combined_result = {
2374            "target": _annotate_run_verdict(command, target_result),
2375            "control": _annotate_run_verdict(command, control_result),
2376            "both_succeeded": outcome.both_succeeded,
2377            "regression_detected": regression_detected,
2378            # The authoritative verdict, so consumers of this payload (stdout and
2379            # the `regression_report` output) never have to re-derive it from
2380            # `regression_detected` -- which is false for an inconclusive run.
2381            "success": success,
2382            "both_failed": outcome.both_failed,
2383        }
2384        if record_comparison is not None:
2385            combined_result["record_comparison"] = record_comparison.to_dict()
2386        if comparison_detail:
2387            combined_result["comparison"] = comparison_detail
2388
2389        print_json(combined_result)
2390
2391        gh_outputs: dict[str, object] = {
2392            "success": success,
2393            "target_image": resolved_test_image,
2394            "control_image": resolved_control_image,
2395            "command": command,
2396            "target_exit_code": target_result["exit_code"],
2397            "control_exit_code": control_result["exit_code"],
2398            "regression_detected": regression_detected,
2399            "both_failed": outcome.both_failed,
2400        }
2401
2402        if command == "check":
2403            gh_outputs["target_connection_status"] = target_result.get(
2404                "connection_status"
2405            )
2406            gh_outputs["control_connection_status"] = control_result.get(
2407                "connection_status"
2408            )
2409
2410        if record_comparison is not None:
2411            gh_outputs["record_comparison_passed"] = record_comparison.passed
2412
2413        write_github_outputs(gh_outputs)
2414
2415        write_json_output("regression_report", combined_result)
2416
2417        report_path = generate_regression_report(
2418            target_image=resolved_test_image,
2419            control_image=resolved_control_image,  # type: ignore[arg-type]
2420            command=command,
2421            target_result=target_result,
2422            control_result=control_result,
2423            output_dir=output_path,
2424            record_comparison=(
2425                record_comparison.to_dict() if record_comparison is not None else None
2426            ),
2427            # Without these the markdown report would announce "no regression"
2428            # over the same run the HTML report and the CLI call a failure --
2429            # and, for a warning, would show no finding at all where the other
2430            # two surfaces show one.
2431            comparison_findings=[
2432                (check.status, f"{check.name}: {check.summary}")
2433                for check in comparison_checks
2434                if check.status in {"fail", "warn"}
2435            ],
2436        )
2437        print_success(f"Generated regression report: {report_path}")
2438
2439        # Same rule as `generate_regression_report` above: the model reads
2440        # `result["success"]` as the exit status and derives the per-version
2441        # verdict itself, so it gets the unannotated originals. `outcome` is
2442        # passed through so the run is evaluated exactly once.
2443        report_model = build_comparison_report_model(
2444            target_image=resolved_test_image,
2445            control_image=resolved_control_image,  # type: ignore[arg-type]
2446            command=command,
2447            target_result=target_result,
2448            control_result=control_result,
2449            outcome=outcome,
2450            record_comparison=record_comparison,
2451            extra_checks=(
2452                *comparison_checks,
2453                *([record_comparison_skipped] if record_comparison_skipped else []),
2454            ),
2455            extra_diff_blocks=comparison_diff_blocks,
2456            configured_streams=_configured_stream_names(cmd, catalog_file),
2457            connection_objects=_collect_connection_objects(
2458                catalog_file, state_file, target_output / "airbyte_messages"
2459            ),
2460        )
2461        html_report_path = write_html_report(report_model, output_path)
2462        print_success(f"Generated regression HTML report: {html_report_path}")
2463
2464        # Write the concise summary to GITHUB_STEP_SUMMARY (if env var exists)
2465        write_github_summary(render_inline_summary(report_model))
2466
2467        if failed_comparisons:
2468            # Only reachable when both commands ran cleanly -- a comparison is
2469            # skipped otherwise -- so this says what the exit codes could not.
2470            print_error(
2471                f"Regression detected between {resolved_test_image} and "
2472                f"{resolved_control_image}: "
2473                + "; ".join(
2474                    f"{check.name.lower()} — {check.summary}"
2475                    for check in failed_comparisons
2476                )
2477            )
2478        elif outcome.check_improvement:
2479            print_success(
2480                f"Regression test passed for {resolved_test_image} vs "
2481                f"{resolved_control_image}: target connectionStatus SUCCEEDED "
2482                "while control FAILED (improvement)."
2483            )
2484        elif outcome.both_succeeded and comparison_errored:
2485            print_error(
2486                f"Record comparison errored for {resolved_test_image} vs "
2487                f"{resolved_control_image}. Test failed/inconclusive; this "
2488                "is a tooling failure, not evidence of a regression."
2489            )
2490        elif outcome.both_succeeded and record_comparison_failed:
2491            print_error(
2492                f"Regression detected between {resolved_test_image} and "
2493                f"{resolved_control_image}: both versions ran, but the "
2494                "record comparison failed."
2495            )
2496        elif outcome.both_succeeded:
2497            print_success(
2498                f"Regression test passed for {resolved_test_image} vs {resolved_control_image}"
2499            )
2500        elif outcome.regression_detected:
2501            print_error(
2502                f"Regression detected between {resolved_test_image} and {resolved_control_image}"
2503            )
2504        elif outcome.both_failed:
2505            print_error(
2506                f"Both versions failed for {resolved_test_image} vs {resolved_control_image}. "
2507                "Test failed/inconclusive; cannot rule out a regression."
2508            )
2509        else:
2510            print_error(
2511                f"Control {resolved_control_image} failed while target "
2512                f"{resolved_test_image} succeeded. Test failed/inconclusive; "
2513                "nothing to compare the target against."
2514            )
2515
2516
2517def _configured_stream_names(
2518    command: Command, catalog_file: Path | None
2519) -> tuple[str, ...]:
2520    """Every stream the run was configured to read.
2521
2522    The report's per-stream tables are keyed on emitted records, so a stream
2523    that returned nothing has no entry and would be missing from the report
2524    entirely. Reading the configured catalog is what lets it be shown as zero.
2525
2526    Only for commands that read records. `--connection-id` fetches a catalog for
2527    every command, and handing it to `spec`, `check` or `discover` would give
2528    them an all-zero stream table and a coverage warning saying no stream
2529    returned anything -- true, meaningless, and enough noise to train a reviewer
2530    to ignore the warning on the one command where it means something.
2531    """
2532    if command not in (Command.READ, Command.READ_WITH_STATE):
2533        return ()
2534
2535    if catalog_file is None or not catalog_file.exists():
2536        return ()
2537
2538    try:
2539        catalog = ConfiguredAirbyteCatalog.parse_raw(catalog_file.read_text())
2540    except Exception as exc:
2541        print_warning(f"Could not read the configured catalog for the report: {exc}")
2542        return ()
2543
2544    return tuple(
2545        sorted(
2546            configured.stream.name
2547            for configured in catalog.streams
2548            if configured.stream
2549        )
2550    )
2551
2552
2553def _collect_connection_objects(
2554    catalog_file: Path | None,
2555    state_file: Path | None,
2556    messages_dir: Path | None,
2557) -> tuple[DiffBlock, ...]:
2558    """Gather the objects the run was given or discovered, for the report.
2559
2560    The connector config is deliberately absent: it is the one object that
2561    carries secrets, and the report is an artifact anyone with repo access can
2562    download.
2563
2564    Args:
2565        catalog_file: The configured catalog handed to the run.
2566        state_file: The state handed to the run, for a warm read.
2567        messages_dir: The target side's `airbyte_messages` directory, which
2568            holds the catalog the connector discovered.
2569
2570    Returns:
2571        One JSON block per object that exists, in the order a reader wants them.
2572    """
2573    sources: list[tuple[str, Path | None]] = [
2574        ("Configured catalog (input)", catalog_file),
2575        ("State (input)", state_file),
2576        (
2577            "Discovered catalog (from the target)",
2578            (messages_dir / "catalog.jsonl") if messages_dir else None,
2579        ),
2580    ]
2581
2582    blocks: list[DiffBlock] = []
2583    for title, path in sources:
2584        if path is None or not path.is_file():
2585            continue
2586
2587        try:
2588            payload = path.read_text()
2589        except OSError:
2590            continue
2591
2592        if payload.strip():
2593            blocks.append(build_json_block(title, payload))
2594
2595    return tuple(blocks)
2596
2597
2598# The order commands are presented in, matching how a sync runs them. Anything
2599# else found under the artifacts directory follows, sorted.
2600_REPORT_COMMAND_ORDER = ("spec", "check", "discover", "read")
2601
2602
2603@connector_app.command(name="consolidate-regression-reports")
2604def consolidate_regression_reports(
2605    artifacts_dir: Annotated[
2606        str,
2607        Parameter(
2608            help="Directory holding one subdirectory per command, each with the "
2609            "`report.html` that `regression-test --output-dir` wrote."
2610        ),
2611    ],
2612    output_path: Annotated[
2613        str | None,
2614        Parameter(
2615            name=["--output"],
2616            help="Where to write the consolidated report. "
2617            "Defaults to `report.html` in `--artifacts-dir`.",
2618        ),
2619    ] = None,
2620) -> None:
2621    """Fold one run's per-command regression reports into a single page.
2622
2623    The regression test runs the CLI once per Airbyte command, so each command
2624    writes its own report into its own artifact. This produces the run-level
2625    view -- a verdict table across commands plus every report inlined -- so a
2626    reviewer downloads one file instead of four.
2627
2628    Never fails the run: a command whose report is missing or unreadable is
2629    skipped, because this is a reporting convenience and the per-command
2630    artifacts remain the source of truth.
2631    """
2632    artifacts_path = Path(artifacts_dir)
2633    ordered = [artifacts_path / command for command in _REPORT_COMMAND_ORDER]
2634    extra = sorted(
2635        path
2636        for path in artifacts_path.glob("*")
2637        if path.is_dir() and path not in ordered
2638    )
2639    report_paths = [
2640        directory / HTML_REPORT_FILENAME for directory in [*ordered, *extra]
2641    ]
2642
2643    consolidated = consolidate_reports(report_paths)
2644    if consolidated is None:
2645        print_warning(
2646            f"No per-command reports found under {artifacts_path}; "
2647            "nothing to consolidate."
2648        )
2649        return
2650
2651    destination = Path(output_path) if output_path else artifacts_path / "report.html"
2652    destination.parent.mkdir(parents=True, exist_ok=True)
2653    destination.write_text(consolidated)
2654    print_success(f"Consolidated regression report: {destination}")
2655
2656
2657@connector_app.command(name="fetch-connection-config")
2658def fetch_connection_config_cmd(
2659    connection_id: Annotated[
2660        str,
2661        Parameter(help="The UUID of the Airbyte Cloud connection."),
2662    ],
2663    output_path: Annotated[
2664        str | None,
2665        Parameter(
2666            help="Path to output file or directory. "
2667            "If directory, writes connection-<id>-config.json inside it. "
2668            "Default: platform temp directory (e.g. /tmp/connection-<id>-config.json)"
2669        ),
2670    ] = None,
2671    with_secrets: Annotated[
2672        bool,
2673        Parameter(
2674            name="--with-secrets",
2675            negative="--no-secrets",
2676            help="If set, fetches unmasked secrets from the internal database. "
2677            "Requires GCP_PROD_DB_ACCESS_CREDENTIALS env var or `gcloud auth application-default login`. "
2678            "Must be used with --oc-issue-url.",
2679        ),
2680    ] = False,
2681    oc_issue_url: Annotated[
2682        str | None,
2683        Parameter(
2684            help="OC issue URL for audit logging. Required when using --with-secrets."
2685        ),
2686    ] = None,
2687) -> None:
2688    """Fetch connection configuration from Airbyte Cloud to a local file.
2689
2690    This command retrieves the source configuration for a given connection ID
2691    and writes it to a JSON file. When `--output-path` is omitted the file is
2692    written to the platform temp directory to avoid accidentally committing
2693    secrets to a git repository.
2694
2695    Requires authentication via AIRBYTE_CLOUD_CLIENT_ID and
2696    AIRBYTE_CLOUD_CLIENT_SECRET environment variables.
2697
2698    When --with-secrets is specified, the command fetches unmasked secrets from
2699    the internal database using the connection-retriever. This additionally requires:
2700    - An OC issue URL for audit logging (--oc-issue-url)
2701    - GCP credentials via `GCP_PROD_DB_ACCESS_CREDENTIALS` env var or `gcloud auth application-default login`
2702    - Cloud SQL Python Connector access to the Prod DB replica.
2703    """
2704    path = Path(output_path) if output_path else None
2705    result = fetch_connection_config(
2706        connection_id=connection_id,
2707        output_path=path,
2708        with_secrets=with_secrets,
2709        oc_issue_url=oc_issue_url,
2710    )
2711    if result.success:
2712        print_success(result.message)
2713    else:
2714        print_error(result.message)
2715    print_json(result.model_dump())
2716
2717
2718def _read_json_input(
2719    json_str: str | None,
2720    input_file: str | None,
2721) -> dict:
2722    """Read JSON from a positional arg, --input file, or STDIN (in that priority)."""
2723    if json_str is not None:
2724        return json.loads(json_str)
2725    if input_file is not None:
2726        return json.loads(Path(input_file).read_text())
2727    if not sys.stdin.isatty():
2728        return json.loads(sys.stdin.read())
2729    exit_with_error(
2730        "No JSON input provided. Pass it as a positional argument, "
2731        "via --input <file>, or pipe to STDIN."
2732    )
2733    raise SystemExit(1)  # unreachable, but satisfies type checker
2734
2735
2736@state_app.command(name="get")
2737def get_connection_state(
2738    connection_id: Annotated[
2739        str,
2740        Parameter(help="The connection ID (UUID) to fetch state for."),
2741    ],
2742    stream_name: Annotated[
2743        str | None,
2744        Parameter(
2745            help="Optional stream name to filter state for a single stream.",
2746            name="--stream",
2747        ),
2748    ] = None,
2749    stream_namespace: Annotated[
2750        str | None,
2751        Parameter(
2752            help="Optional stream namespace to narrow the stream filter.",
2753            name="--namespace",
2754        ),
2755    ] = None,
2756    output: Annotated[
2757        str | None,
2758        Parameter(
2759            help="Path to write the state JSON output to a file.",
2760            name="--output",
2761        ),
2762    ] = None,
2763) -> None:
2764    """Get the current state for an Airbyte Cloud connection."""
2765    workspace = CloudWorkspace.from_env()
2766    conn = workspace.get_connection(connection_id)
2767
2768    if stream_name is not None:
2769        stream_state = conn.get_stream_state(
2770            stream_name=stream_name,
2771            stream_namespace=stream_namespace,
2772        )
2773        result = {
2774            "connection_id": connection_id,
2775            "stream_name": stream_name,
2776            "stream_namespace": stream_namespace,
2777            "stream_state": stream_state,
2778        }
2779    else:
2780        result = conn.dump_raw_state()
2781
2782    json_output = json.dumps(result, indent=2, default=str)
2783    if output is not None:
2784        Path(output).write_text(json_output + "\n")
2785        print(f"State written to {output}", file=sys.stderr)
2786    else:
2787        print(json_output)
2788
2789
2790@state_app.command(name="set")
2791def set_connection_state(
2792    connection_id: Annotated[
2793        str,
2794        Parameter(help="The connection ID (UUID) to update state for."),
2795    ],
2796    state_json: Annotated[
2797        str | None,
2798        Parameter(
2799            help="The connection state as a JSON string. When --stream is used, "
2800            'this is just the stream\'s state blob (e.g., \'{"cursor": "2024-01-01"}\').'
2801            " Otherwise, must include 'stateType', 'connectionId', and the appropriate "
2802            "state field. Can also be provided via --input or STDIN."
2803        ),
2804    ] = None,
2805    stream_name: Annotated[
2806        str | None,
2807        Parameter(
2808            help="Optional stream name to update state for a single stream only.",
2809            name="--stream",
2810        ),
2811    ] = None,
2812    stream_namespace: Annotated[
2813        str | None,
2814        Parameter(
2815            help="Optional stream namespace to identify the stream.",
2816            name="--namespace",
2817        ),
2818    ] = None,
2819    input_file: Annotated[
2820        str | None,
2821        Parameter(
2822            help="Path to a JSON file containing the state to set.",
2823            name="--input",
2824        ),
2825    ] = None,
2826) -> None:
2827    """Set the state for an Airbyte Cloud connection.
2828
2829    State JSON can be provided as a positional argument, via --input <file>,
2830    or piped through STDIN.
2831
2832    Uses the safe variant that prevents updates while a sync is running.
2833    When --stream is provided, only that stream's state is updated within
2834    the existing connection state.
2835    """
2836    parsed_state = _read_json_input(state_json, input_file)
2837    workspace = CloudWorkspace.from_env()
2838    conn = workspace.get_connection(connection_id)
2839
2840    if stream_name is not None:
2841        conn.set_stream_state(
2842            stream_name=stream_name,
2843            state_blob_dict=parsed_state,
2844            stream_namespace=stream_namespace,
2845        )
2846    else:
2847        conn.import_raw_state(parsed_state)
2848
2849    result = conn.dump_raw_state()
2850    json_output = json.dumps(result, indent=2, default=str)
2851    print(json_output)
2852
2853
2854@state_app.command(name="reset")
2855def reset_stream_state(
2856    connection_id: Annotated[
2857        str,
2858        Parameter(help="The connection ID (UUID) to update state for."),
2859    ],
2860    stream_name: Annotated[
2861        str,
2862        Parameter(
2863            help="The configured stream name whose state should be reset.",
2864            name="--stream",
2865        ),
2866    ],
2867    stream_namespace: Annotated[
2868        str | None,
2869        Parameter(
2870            help="Optional stream namespace to identify the stream.",
2871            name="--namespace",
2872        ),
2873    ] = None,
2874) -> None:
2875    """Reset a configured stream's state so the next sync full-refreshes it.
2876
2877    Uses the safe variant that prevents updates while a sync is running.
2878    Returns a restorable `previous_state_backup` in raw Config API format.
2879    """
2880    workspace = CloudWorkspace.from_env()
2881    conn = workspace.get_connection(connection_id)
2882
2883    try:
2884        result = reset_cloud_stream_state(
2885            conn,
2886            stream_name=stream_name,
2887            stream_namespace=stream_namespace,
2888        )
2889    except PyAirbyteInputError as e:
2890        exit_with_error(str(e))
2891
2892    print(json.dumps(result.model_dump(), indent=2, default=str))
2893
2894
2895@catalog_app.command(name="get")
2896def get_connection_catalog(
2897    connection_id: Annotated[
2898        str,
2899        Parameter(help="The connection ID (UUID) to fetch catalog for."),
2900    ],
2901    output: Annotated[
2902        str | None,
2903        Parameter(
2904            help="Path to write the catalog JSON output to a file.",
2905            name="--output",
2906        ),
2907    ] = None,
2908) -> None:
2909    """Get the configured catalog for an Airbyte Cloud connection."""
2910    workspace = CloudWorkspace.from_env()
2911    conn = workspace.get_connection(connection_id)
2912    result = conn.dump_raw_catalog()
2913    if result is None:
2914        exit_with_error("No configured catalog found for this connection.")
2915        raise SystemExit(1)  # unreachable, but satisfies type checker
2916
2917    json_output = json.dumps(result, indent=2, default=str)
2918    if output is not None:
2919        Path(output).write_text(json_output + "\n")
2920        print(f"Catalog written to {output}", file=sys.stderr)
2921    else:
2922        print(json_output)
2923
2924
2925@catalog_app.command(name="set")
2926def set_connection_catalog(
2927    connection_id: Annotated[
2928        str,
2929        Parameter(help="The connection ID (UUID) to update catalog for."),
2930    ],
2931    catalog_json: Annotated[
2932        str | None,
2933        Parameter(
2934            help="The configured catalog as a JSON string. "
2935            "Can also be provided via --input or STDIN."
2936        ),
2937    ] = None,
2938    input_file: Annotated[
2939        str | None,
2940        Parameter(
2941            help="Path to a JSON file containing the catalog to set.",
2942            name="--input",
2943        ),
2944    ] = None,
2945) -> None:
2946    """Set the configured catalog for an Airbyte Cloud connection.
2947
2948    Catalog JSON can be provided as a positional argument, via --input <file>,
2949    or piped through STDIN.
2950
2951    WARNING: This replaces the entire configured catalog.
2952    """
2953    parsed_catalog = _read_json_input(catalog_json, input_file)
2954    workspace = CloudWorkspace.from_env()
2955    conn = workspace.get_connection(connection_id)
2956    conn.import_raw_catalog(parsed_catalog)
2957
2958    result = conn.dump_raw_catalog()
2959    if result is None:
2960        exit_with_error("Failed to retrieve catalog after import.")
2961        raise SystemExit(1)  # unreachable, but satisfies type checker
2962
2963    json_output = json.dumps(result, indent=2, default=str)
2964    print(json_output)
2965
2966
2967@logs_app.command(name="lookup-cloud-backend-error")
2968def lookup_cloud_backend_error(
2969    error_id: Annotated[
2970        str,
2971        Parameter(
2972            help=(
2973                "The error ID (UUID) to search for. This is typically returned "
2974                "in API error responses as {'errorId': '...'}"
2975            )
2976        ),
2977    ],
2978    lookback_days: Annotated[
2979        int,
2980        Parameter(help="Number of days to look back in logs."),
2981    ] = 7,
2982    min_severity_filter: Annotated[
2983        GCPSeverity | None,
2984        Parameter(
2985            help="Optional minimum severity level to filter logs.",
2986        ),
2987    ] = None,
2988    raw: Annotated[
2989        bool,
2990        Parameter(help="Output raw JSON instead of formatted text."),
2991    ] = False,
2992) -> None:
2993    """Look up error details from GCP Cloud Logging by error ID.
2994
2995    When an Airbyte Cloud API returns an error response with only an error ID
2996    (e.g., {"errorId": "3173452e-8f22-4286-a1ec-b0f16c1e078a"}), this command
2997    fetches the full stack trace and error details from GCP Cloud Logging.
2998
2999    Requires GCP credentials with Logs Viewer role on the target project.
3000    Set up credentials with: gcloud auth application-default login
3001    """
3002    print(f"Searching for error ID: {error_id}", file=sys.stderr)
3003    print(f"Lookback days: {lookback_days}", file=sys.stderr)
3004    if min_severity_filter:
3005        print(f"Severity filter: {min_severity_filter}", file=sys.stderr)
3006    print(file=sys.stderr)
3007
3008    result = fetch_error_logs(
3009        error_id=error_id,
3010        lookback_days=lookback_days,
3011        min_severity_filter=min_severity_filter,
3012    )
3013
3014    if raw:
3015        print_json(result.model_dump())
3016        return
3017
3018    print(f"Found {result.total_entries_found} log entries", file=sys.stderr)
3019    print(file=sys.stderr)
3020
3021    if result.payloads:
3022        for i, payload in enumerate(result.payloads):
3023            print(f"=== Log Group {i + 1} ===")
3024            print(f"Timestamp: {payload.timestamp}")
3025            print(f"Severity: {payload.severity}")
3026            if payload.resource.labels.pod_name:
3027                print(f"Pod: {payload.resource.labels.pod_name}")
3028            print(f"Lines: {payload.num_log_lines}")
3029            print()
3030            print(payload.message)
3031            print()
3032    elif result.entries:
3033        print("No grouped payloads, showing raw entries:", file=sys.stderr)
3034        for entry in result.entries:
3035            print(f"[{entry.timestamp}] {entry.severity}: {entry.payload}")
3036    else:
3037        print_error("No log entries found for this error ID.")
3038
3039
3040# =============================================================================
3041# Organization commands
3042# =============================================================================
3043
3044
3045@organization_app.command(name="search")
3046def organization_search(
3047    name_contains: Annotated[
3048        str,
3049        Parameter(
3050            name="--name-contains",
3051            help="Case-insensitive substring to match against organization name or email.",
3052        ),
3053    ],
3054    limit: Annotated[
3055        int,
3056        Parameter(
3057            name="--limit",
3058            help="Maximum number of results to return.",
3059        ),
3060    ] = 20,
3061) -> None:
3062    """Search organizations by name or email substring."""
3063    rows = _search_organizations(name_contains=name_contains, limit=limit)
3064    if not rows:
3065        print_error(f"No organizations found matching '{name_contains}'.")
3066        return
3067
3068    print_json([dict(row) for row in rows])
3069
3070
3071@organization_app.command(name="info")
3072def organization_info(
3073    organization_id: Annotated[
3074        str,
3075        Parameter(help="The organization UUID."),
3076    ],
3077) -> None:
3078    """Get information about an organization (stub — not yet implemented)."""
3079    exit_with_error(
3080        "organization info is not yet implemented. "
3081        "Use 'airbyte-ops cloud organization search' to find organizations by name."
3082    )
3083
3084
3085# =============================================================================
3086# Workspace commands
3087# =============================================================================
3088
3089
3090@workspace_app.command(name="search")
3091def workspace_search(
3092    name_contains: Annotated[
3093        str | None,
3094        Parameter(
3095            name="--name-contains",
3096            help="Case-insensitive substring to match against workspace name or slug.",
3097        ),
3098    ] = None,
3099    email_domain: Annotated[
3100        str | None,
3101        Parameter(
3102            name="--email-domain",
3103            help="Email domain to search for (e.g. 'motherduck.com').",
3104        ),
3105    ] = None,
3106    limit: Annotated[
3107        int,
3108        Parameter(
3109            name="--limit",
3110            help="Maximum number of results to return.",
3111        ),
3112    ] = 100,
3113) -> None:
3114    """Search workspaces by name substring or email domain."""
3115    if not name_contains and not email_domain:
3116        exit_with_error(
3117            "At least one of --name-contains or --email-domain must be provided."
3118        )
3119
3120    if name_contains:
3121        rows = _search_workspaces(name_contains=name_contains, limit=limit)
3122    else:
3123        rows = query_workspaces_by_email_domain(
3124            email_domain=email_domain.lstrip("@"),  # type: ignore[union-attr]
3125            limit=limit,
3126        )
3127
3128    if not rows:
3129        search_term = name_contains or email_domain
3130        print_error(f"No workspaces found matching '{search_term}'.")
3131        return
3132
3133    print_json([dict(row) for row in rows])
3134
3135
3136@workspace_app.command(name="info")
3137def workspace_info_cmd(
3138    workspace_id: Annotated[
3139        str,
3140        Parameter(help="The workspace UUID."),
3141    ],
3142) -> None:
3143    """Get information about a workspace."""
3144    result = query_workspace_info(workspace_id=workspace_id)
3145    if not result:
3146        print_error(f"No workspace found with ID '{workspace_id}'.")
3147        return
3148
3149    print_json(dict(result))