fix(mcp): polish Connector Builder base PR with additional fixes and improvements

airbyte/cloud/connector_builder/__init__.py

@@ -22,6 +22,9 @@
 )
 from airbyte.cloud.connector_builder.project import (
     CONNECTOR_BUILDER_TEST_DESTINATION_NAME,
+    TEST_READ_MAX_PAGES,
+    TEST_READ_MAX_RECORDS,
+    TEST_READ_MAX_SLICES,
     CloudConnectorBuilderProject,
 )
 
@@ -39,4 +42,7 @@
     "ConnectorBuilderStreamReadResult",
     "ManifestTextEditMode",
     "ReadinessCheckResult",
+    "TEST_READ_MAX_PAGES",
+    "TEST_READ_MAX_RECORDS",
+    "TEST_READ_MAX_SLICES",
 ]

airbyte/cloud/connector_builder/_util.py

@@ -141,8 +141,12 @@ def readiness_next_steps(
     not_attempted_streams: list[str],
 ) -> list[str]:
     steps = [check.description for check in checks[:4] if check.result != "passed"]
-    failed = [stream.stream_name for stream in streams if not stream.passed]
-    capped = [stream.stream_name for stream in streams if stream.passed and not stream.read_to_end]
+    failed = [stream.stream_name for stream in streams if stream.failed]
+    capped = [
+        stream.stream_name
+        for stream in streams
+        if stream.error is None and not stream.http_errors and not stream.read_to_end
+    ]
     if failed:
         steps.append(f"Fix the failed streams and test them again: {', '.join(failed)}")
     if pending_streams:
@@ -156,16 +160,14 @@ def readiness_next_steps(
             "Run the report again with the not-attempted streams, skipping any already "
             f"validated or unchanged since: {', '.join(not_attempted_streams)}"
         )
-    if not (failed or pending_streams or not_attempted_streams):
-        steps.append(
-            "All named streams passed. Once every stream is validated, publish the custom "
-            "source definition (if not yet) and run a full sync of every stream to a no-op "
-            "destination."
-        )
     if capped:
         steps.append(
-            f"These streams hit a test-read cap; check them with the full sync: {', '.join(capped)}"
+            "These streams hit a test-read cap, so later pages weren't read; "
+            f"check them with a full sync: {', '.join(capped)}"
         )
+    warning_streams = [stream.stream_name for stream in streams if stream.warnings]
+    if warning_streams:
+        steps.append(f"Report these streams' `warnings` to the user: {', '.join(warning_streams)}")
     return steps
 
 
@@ -180,8 +182,15 @@ def readiness_checks(
     takes_no_inputs = isinstance(spec, dict) and not spec.get("properties")
     testing_values = project_data.get("testingValues") or {}
     source_definition_id = project_data["builderProject"].get("sourceDefinitionId")
-    failed = [stream.stream_name for stream in streams if not stream.passed]
-    not_read_to_end = [stream.stream_name for stream in streams if not stream.read_to_end]
+    failed = [stream.stream_name for stream in streams if stream.failed]
+    capped = [
+        stream.stream_name
+        for stream in streams
+        if stream.error is None and not stream.http_errors and not stream.read_to_end
+    ]
+    not_read = [
+        stream.stream_name for stream in streams if stream.error is not None or stream.http_errors
+    ]
     return [
         ConnectorBuilderReadinessCheck(
             description="Save a draft manifest",
@@ -201,7 +210,7 @@ def readiness_checks(
             detail=url,
         ),
         ConnectorBuilderReadinessCheck(
-            description="Pass a test read on the named streams",
+            description="Test-read the named streams without errors",
             result=_result(not failed and not pending_streams),
             detail="; ".join(
                 part
@@ -215,13 +224,21 @@ def readiness_checks(
         ),
         ConnectorBuilderReadinessCheck(
             description="Read the named streams to the end within the test-read caps",
-            result=_result(not not_read_to_end and not pending_streams),
-            detail=(
-                f"hit a cap: {', '.join(not_read_to_end)}. Check these with a full sync "
-                "after publishing."
-                if not_read_to_end
-                else None
-            ),
+            result=_result(not pending_streams and all(stream.read_to_end for stream in streams)),
+            detail="; ".join(
+                part
+                for part in (
+                    (
+                        f"hit a cap: {', '.join(capped)}. Check these with a full sync "
+                        "after publishing."
+                    )
+                    if capped
+                    else "",
+                    f"not read: {', '.join(not_read)} (test read failed)" if not_read else "",
+                )
+                if part
+            )
+            or None,
         ),
         ConnectorBuilderReadinessCheck(
             description="Publish the custom source definition",

airbyte/cloud/connector_builder/models.py

@@ -33,6 +33,40 @@ class ConnectorBuilderPageSummary(BaseModel):
     )
 
 
+def _record_count_warnings(
+    record_count: int,
+    pages: list[ConnectorBuilderPageSummary],
+    *,
+    test_read_limit_reached: bool,
+) -> list[str]:
+    """Return informational hints for record counts that may indicate a partial read."""
+    if test_read_limit_reached:
+        return []
+    if record_count == 0:
+        return ["Read 0 records. Check the request, record selector, and testing values."]
+
+    warnings = []
+    if record_count % 10 == 0:
+        warnings.append(
+            f"Read {record_count:,} records, a multiple of 10. A page size or API limit may "
+            "have cut the read short; check the paginator."
+        )
+
+    record_pages = [page for page in pages if page.record_count > 0]
+    if (
+        record_pages
+        and record_pages[0].record_count > 1
+        and pages[-1].record_count > 0
+        and pages[-1].record_count == record_pages[0].record_count
+    ):
+        record_count_per_page = record_pages[0].record_count
+        warnings.append(
+            f"The last page returned a full page of {record_count_per_page:,} records, the "
+            "same as the first page, so more pages may exist; check the paginator."
+        )
+    return warnings
+
+
 class ConnectorBuilderStreamReadResult(BaseModel):
     """Result of a Connector Builder test read of one stream."""
 
@@ -45,6 +79,13 @@ class ConnectorBuilderStreamReadResult(BaseModel):
     pages: list[ConnectorBuilderPageSummary]
     logs: list[dict[str, Any]]
     inferred_schema: dict[str, Any] | None = None
+    warnings: list[str] = Field(
+        default_factory=list,
+        description=(
+            "Hints that the read may be incomplete, such as round record counts. "
+            "They don't fail the read."
+        ),
+    )
 
     @classmethod
     def from_api_response(
@@ -77,19 +118,25 @@ def from_api_response(
                     )
                 )
 
+        test_read_limit_reached = bool(response.get("test_read_limit_reached"))
         return cls(
             stream_name=stream_name,
             record_count=len(records),
             slice_count=len(slices),
             page_count=len(pages),
-            test_read_limit_reached=bool(response.get("test_read_limit_reached")),
+            test_read_limit_reached=test_read_limit_reached,
             records=records,
             pages=pages,
             logs=[
                 {"level": log.get("level"), "message": log.get("message")}
                 for log in response.get("logs") or []
             ],
             inferred_schema=response.get("inferred_schema"),
+            warnings=_record_count_warnings(
+                len(records),
+                pages,
+                test_read_limit_reached=test_read_limit_reached,
+            ),
         )
 
 
@@ -129,19 +176,31 @@ class ConnectorBuilderStreamReadiness(BaseModel):
     """Readiness of one stream, from a Connector Builder test read."""
 
     stream_name: str
-    passed: bool
+    failed: bool = Field(
+        description=(
+            "True when the test read raised an error, returned HTTP errors, or logged errors."
+        ),
+    )
     record_count: int = 0
     page_count: int = 0
     read_to_end: bool = Field(
         default=False,
         description=(
-            "True when the stream ended within the test-read caps. False means the read hit a "
-            "record, page, or slice cap, so later pages were never read."
+            "True when the stream ended within the test-read caps without HTTP errors. False "
+            "means a record, page, or slice cap was hit, an HTTP error occurred, or the read "
+            "did not run (see `error`)."
         ),
     )
     http_errors: list[ConnectorBuilderPageSummary] = Field(default_factory=list)
     problem_logs: list[dict[str, Any]] = Field(default_factory=list)
     inferred_schema: dict[str, Any] | None = None
+    warnings: list[str] = Field(
+        default_factory=list,
+        description=(
+            "Hints that the read may be incomplete, such as round record counts. "
+            "They don't fail the read."
+        ),
+    )
     error: str | None = Field(
         default=None,
         description="The error raised by the test read itself, if it failed to run.",
@@ -151,7 +210,7 @@ class ConnectorBuilderStreamReadiness(BaseModel):
     def from_stream_read(
         cls, result: ConnectorBuilderStreamReadResult
     ) -> ConnectorBuilderStreamReadiness:
-        """Judge a stream from its test read: records returned, no HTTP or error logs."""
+        """Summarize errors and results from a stream test read."""
         http_errors = [
             page
             for page in result.pages
@@ -163,13 +222,14 @@ def from_stream_read(
         )
         return cls(
             stream_name=result.stream_name,
-            passed=result.record_count > 0 and not has_errors,
+            failed=has_errors,
             record_count=result.record_count,
             page_count=result.page_count,
-            read_to_end=not result.test_read_limit_reached,
+            read_to_end=not result.test_read_limit_reached and not http_errors,
             http_errors=http_errors,
             problem_logs=problem_logs,
             inferred_schema=result.inferred_schema,
+            warnings=result.warnings,
         )
 
 
@@ -200,13 +260,6 @@ class ConnectorBuilderFullSync(BaseModel):
 class ConnectorBuilderReadinessReport(BaseModel):
     """Readiness checks and per-stream test reads for a Connector Builder draft."""
 
-    passed: bool = Field(
-        description=(
-            "True when every named stream passed and was read to the end. Covers only the "
-            "named streams; a full sync validates the whole connector."
-        ),
-    )
-    passed_streams: list[str] = Field(default_factory=list)
     failed_streams: list[str] = Field(default_factory=list)
     pending_streams: list[str] = Field(
         default_factory=list,

airbyte/cloud/connector_builder/project.py

@@ -51,6 +51,33 @@
 """Shortest HTTP timeout for a test read started just before the readiness deadline."""
 CONNECTOR_BUILDER_TEST_DESTINATION_NAME = "Connector Builder Test (No-op)"
 """Name of the no-op destination that full syncs find or create by default."""
+TEST_READ_MAX_RECORDS = 5_000
+"""Most records read per stream by `run_readiness_checks()`."""
+TEST_READ_MAX_PAGES = 20
+TEST_READ_MAX_SLICES = 20
+
+
+def _check_test_read_limits(
+    record_limit: int,
+    page_limit: int,
+    slice_limit: int,
+) -> None:
+    """Reject test-read limits outside Airbyte Cloud's supported range."""
+    limits = (
+        ("record_limit", record_limit, TEST_READ_MAX_RECORDS),
+        ("page_limit", page_limit, TEST_READ_MAX_PAGES),
+        ("slice_limit", slice_limit, TEST_READ_MAX_SLICES),
+    )
+    if any(not 1 <= value <= maximum for _, value, maximum in limits):
+        raise exc.AirbyteLibInputError(
+            message="Test-read limits are out of range.",
+            guidance=(
+                f"Airbyte Cloud accepts record_limit 1 to {TEST_READ_MAX_RECORDS:,}, "
+                f"page_limit 1 to {TEST_READ_MAX_PAGES}, and "
+                f"slice_limit 1 to {TEST_READ_MAX_SLICES}."
+            ),
+            context={name: value for name, value, _ in limits},
+        )
 
 
 class CloudConnectorBuilderProject:  # noqa: PLR0904  # Builder project API facade.
@@ -271,10 +298,11 @@ def test_read_stream(
 
         Args:
             stream_name: The stream to read.
-            record_limit: Maximum records to return (1 to 10,000).
+            record_limit: Maximum records to return (1 to 5,000).
             page_limit: Maximum pages to fetch per slice (1 to 20).
             slice_limit: Maximum slices to read (1 to 20).
         """
+        _check_test_read_limits(record_limit, page_limit, slice_limit)
         return self._test_read_stream(
             self._require_draft_manifest(),
             stream_name,
@@ -339,36 +367,37 @@ def run_readiness_checks(
         self,
         stream_names: list[str],
         *,
-        record_limit: int = 10_000,
-        page_limit: int = 20,
-        slice_limit: int = 20,
+        record_limit: int = TEST_READ_MAX_RECORDS,
+        page_limit: int = TEST_READ_MAX_PAGES,
+        slice_limit: int = TEST_READ_MAX_SLICES,
         timeout_seconds: float | None = None,
         run_full_sync: bool = False,
         test_source_id: str | None = None,
         on_resource_created: Callable[[str], None] | None = None,
     ) -> ConnectorBuilderReadinessReport:
         """Check the draft's progress and test-read the given streams on Airbyte Cloud.
 
-        Reads run in parallel. A stream passes when its test read returns records with no
-        HTTP errors or error logs, and is `read_to_end` when it ended within the test-read
-        caps (which default to Cloud's maximums). The readiness checks are derived from the project,
-        so nothing is stored. `passed` covers only the named streams: each must pass and be
-        read to the end. A full sync after publishing validates the whole connector.
+        Reads run in parallel. A stream is `failed` when its read errored, returned HTTP
+        errors, or logged errors. The report lists failures and warnings without judging
+        readiness; `read_to_end` indicates whether the read ended within the test-read caps.
+        The readiness checks are derived from the project, so nothing is stored. A full sync
+        after publishing validates the whole connector.
 
         Args:
             stream_names: Streams to test-read (at least 1). For
                 dynamic streams, get the generated names from `resolve_manifest()`.
-            record_limit: Maximum records to read per stream (1 to 10,000).
+            record_limit: Maximum records to read per stream (1 to 5,000).
             page_limit: Maximum pages to fetch per slice (1 to 20).
             slice_limit: Maximum slices to read (1 to 20).
             timeout_seconds: Return after this many seconds with the streams finished so far;
                 the rest are listed in `pending_streams` and `not_attempted_streams`. No limit
                 by default.
-            run_full_sync: If every named stream finished and passed (or none are named),
+            run_full_sync: If no named stream failed or is still pending (or none are named),
                 publish the draft and start a full sync with `run_full_sync()`.
             test_source_id: The source for the full sync. See `run_full_sync()`.
             on_resource_created: Passed to `run_full_sync()`.
         """
+        _check_test_read_limits(record_limit, page_limit, slice_limit)
         stream_names = list(dict.fromkeys(stream_names))
         if not stream_names and not run_full_sync:
             raise exc.AirbyteLibInputError(
@@ -414,12 +443,12 @@ def read(stream_name: str) -> ConnectorBuilderStreamReadiness:
                 if ex.guidance:
                     error += f" {ex.guidance}"
                 return ConnectorBuilderStreamReadiness(
-                    stream_name=stream_name, passed=False, error=error
+                    stream_name=stream_name, failed=True, error=error
                 )
             except requests.RequestException as ex:
                 return ConnectorBuilderStreamReadiness(
                     stream_name=stream_name,
-                    passed=False,
+                    failed=True,
                     error=f"Request to Airbyte Cloud failed: {type(ex).__name__}",
                 )
             return ConnectorBuilderStreamReadiness.from_stream_read(result)
@@ -461,10 +490,7 @@ def _readiness_report(
         remaining = pending_streams + not_attempted_streams
         checks = readiness_checks(project_data, draft, streams, remaining, self.url)
         return ConnectorBuilderReadinessReport(
-            passed=not remaining
-            and all(stream.passed and stream.read_to_end for stream in streams),
-            passed_streams=[stream.stream_name for stream in streams if stream.passed],
-            failed_streams=[stream.stream_name for stream in streams if not stream.passed],
+            failed_streams=[stream.stream_name for stream in streams if stream.failed],
             pending_streams=pending_streams,
             not_attempted_streams=not_attempted_streams,
             checks=checks,