diff --git a/airbyte/cloud/connections.py b/airbyte/cloud/connections.py index 15d29f401..31d27488e 100644 --- a/airbyte/cloud/connections.py +++ b/airbyte/cloud/connections.py @@ -9,6 +9,7 @@ from airbyte._direct_connectors import api_util as agents_api_util from airbyte._util import api_util +from airbyte._util.cloud_errors import parse_cloud_error from airbyte.cloud._connection_catalog import ( _denormalize_catalog_to_api, _is_protocol_catalog_format, @@ -35,6 +36,7 @@ from airbyte.cloud.sync_results import SyncResult from airbyte.exceptions import ( AirbyteCloudError, + AirbyteConnectionSyncActiveError, AirbyteConnectionSyncError, AirbyteLibInputError, AirbyteMissingResourceError, @@ -375,6 +377,10 @@ def run_sync( ), context={"connection_id": self.connection_id}, ) from ex + if ex.context and ex.context.get("status_code") == HTTPStatus.CONFLICT: + active_sync_error = self._already_running_sync_error(ex) + if active_sync_error is not None: + raise active_sync_error from ex raise sync_result = SyncResult( workspace=self.workspace, @@ -391,6 +397,55 @@ def run_sync( return sync_result + def _already_running_sync_error( + self, + error: AirbyteConnectionSyncError, + ) -> AirbyteConnectionSyncActiveError | None: + """Return actionable guidance when Cloud rejects a duplicate running sync.""" + sdk_error = error.__cause__ + if not isinstance(sdk_error, api_util.SDKError): + return None + + problem = parse_cloud_error( + sdk_error.status_code, + api_util.error_response_body(sdk_error), + ) + if problem.slug != "try-again-later": + return None + + sync_results = self.get_previous_sync_logs( + limit=1, + job_type=JobTypeEnum.SYNC, + ) + sync_result = sync_results[0] if sync_results else None + if sync_result is None or sync_result.is_job_complete(): + return None + + job_id = sync_result.job_id + return AirbyteConnectionSyncActiveError( + connection_id=self.connection_id, + message=( + f"A sync is already running for connection '{self.connection_id}' " + f"(job {job_id})." + ), + guidance=( + "Don't start another sync. Monitor the running job with " + f"`connection.get_sync_result(job_id={job_id})` or the " + "`get_cloud_sync_status` MCP tool " + f"(`connection_id='{self.connection_id}', job_id={job_id}`). " + "Only if the user wants to restart it, cancel it first with " + f"`connection.cancel_sync(job_id={job_id})` or the " + "`cancel_cloud_sync` MCP tool " + f"(`connection_id='{self.connection_id}', job_id={job_id}`), " + f"then run the sync again. Job URL: {sync_result.job_url}." + ), + context={ + "connection_id": self.connection_id, + "job_id": job_id, + "job_url": sync_result.job_url, + }, + ) + def _get_latest_cancellable_sync_job_id(self) -> int: """Get the latest cancellable sync job ID.""" sync_results = self.get_previous_sync_logs( diff --git a/airbyte/exceptions.py b/airbyte/exceptions.py index 94f413cd6..d2456f668 100644 --- a/airbyte/exceptions.py +++ b/airbyte/exceptions.py @@ -698,7 +698,7 @@ class AirbyteConnectionSyncError(AirbyteCloudConnectionError): @dataclass class AirbyteConnectionSyncActiveError(AirbyteCloudConnectionError): - """State update rejected because a sync is currently running (HTTP 423).""" + """Rejected because a sync is currently running on this connection.""" @dataclass diff --git a/airbyte/mcp/_error_handling.py b/airbyte/mcp/_error_handling.py index 0360fece0..8f900fcf1 100644 --- a/airbyte/mcp/_error_handling.py +++ b/airbyte/mcp/_error_handling.py @@ -1,5 +1,5 @@ # Copyright (c) 2026 Airbyte, Inc., all rights reserved. -"""PyAirbyte-specific handling and trace classification of MCP tool errors.""" +"""Format MCP tool errors with Cloud details and local recovery guidance.""" from __future__ import annotations @@ -28,6 +28,8 @@ from airbyte.exceptions import ( AirbyteAgentsUnavailableError, AirbyteCloudApiError, + AirbyteCloudError, + AirbyteConnectionSyncActiveError, AirbyteConnectionSyncError, AirbyteConnectionSyncTimeoutError, AirbyteConnectorInUseError, @@ -90,6 +92,7 @@ AirbyteDestructiveToolsDisabledError, AirbyteExternalAccessDisabledError, AirbyteAgentsUnavailableError, + AirbyteConnectionSyncActiveError, AirbyteConnectorNotReadyError, AirbyteConnectorInUseError, AirbyteMissingResourceError, @@ -194,10 +197,38 @@ def _cloud_problem_text(problem: CloudErrorInfo) -> str: return f"{problem.message} {guidance}".rstrip() if guidance else problem.message -def format_user_facing_error(error: BaseException) -> str: - """Return the error message followed by its guidance, when present.""" - if (problem := _cloud_error_info(error)) is not None: +def _format_cloud_error_with_local_guidance(error: BaseException) -> str | None: + """Keep local recovery guidance alongside a selected Cloud problem message.""" + problem = _cloud_error_info(error) + if problem is None: + return None + + local_error = next( + ( + cause + for cause in _chain(error) + if isinstance(cause, AirbyteLibError) + and not isinstance(cause, AirbyteCloudApiError) + and not ( + isinstance(cause, AirbyteCloudError) and cause.get_message() == problem.message + ) + ), + None, + ) + if local_error is None or not local_error.guidance: return _cloud_problem_text(problem) + + local_message = local_error.get_message() + local_text = f"{local_message} {local_error.guidance}" + if problem.message == local_message or problem.message in local_text: + return local_text + return f"{local_text} Airbyte Cloud response: {problem.message}" + + +def format_user_facing_error(error: BaseException) -> str: + """Format local error guidance and Cloud problem details without debug context.""" + if text := _format_cloud_error_with_local_guidance(error): + return text if not isinstance(error, AirbyteLibError): return str(error) text = error.get_message() @@ -241,9 +272,9 @@ def mcp_tool_error_reason(error: BaseException) -> str | None: def agent_error_text(error: ToolError) -> str: - """Return the text an agent sees for a tool error before redaction.""" - if (problem := _cloud_error_info(error)) is not None: - return _cloud_problem_text(problem) + """Return local guidance and Cloud details visible to an agent before redaction.""" + if text := _format_cloud_error_with_local_guidance(error): + return text cause = error.__cause__ if isinstance(cause, AirbyteLibError): return format_user_facing_error(cause) diff --git a/airbyte/mcp/cloud.py b/airbyte/mcp/cloud.py index 39c1d41b0..6c6aa69e1 100644 --- a/airbyte/mcp/cloud.py +++ b/airbyte/mcp/cloud.py @@ -1006,7 +1006,14 @@ def run_cloud_sync( ), ], ) -> str: - """Run a sync job on Airbyte Cloud.""" + """Run a sync job on Airbyte Cloud. + + A disabled connection returns 409; re-enable it with `update_cloud_connection` + (`enabled=True`) first. If a sync is already running, the error includes its job ID: + monitor it with `get_cloud_sync_status` instead of retrying, or call `cancel_cloud_sync` + first if a restart is really wanted. After a `wait=True` timeout, the sync keeps running; + check `get_cloud_sync_status` rather than starting another sync. + """ workspace: CloudWorkspace = _get_cloud_workspace(ctx, workspace_id) connection = workspace.get_connection(connection_id=connection_id) sync_result = connection.run_sync(wait=wait, wait_timeout=wait_timeout) diff --git a/tests/unit_tests/test_cloud_connections.py b/tests/unit_tests/test_cloud_connections.py index c609b5ec6..b666a25d1 100644 --- a/tests/unit_tests/test_cloud_connections.py +++ b/tests/unit_tests/test_cloud_connections.py @@ -21,6 +21,7 @@ from airbyte.cloud.workspaces import CloudWorkspace from airbyte.exceptions import ( AirbyteCloudApiError, + AirbyteConnectionSyncActiveError, AirbyteConnectionSyncError, AirbyteCloudError, AirbyteMissingResourceError, @@ -826,26 +827,92 @@ def _connection_info_with_status(status: str) -> MagicMock: return info -def test_run_sync_conflict_on_disabled_connection_raises_input_error( +@pytest.mark.parametrize( + ("scenario", "connection_status"), + [ + pytest.param("running-job", "active", id="running-job"), + pytest.param("no-jobs", "active", id="no-jobs"), + pytest.param("completed-job", "active", id="completed-job"), + pytest.param("disabled", "inactive", id="disabled-connection"), + ], +) +def test_run_sync_handles_try_again_later_conflicts( monkeypatch: pytest.MonkeyPatch, + scenario: str, + connection_status: str, ) -> None: - """Verify a 409 on an inactive connection surfaces a disabled-connection hint.""" + """Explain known running-sync conflicts and preserve unrelated errors.""" connection = _connection() + body = ( + '{"type":"https://reference.airbyte.com/reference/errors#try-again-later",' + '"title":"try-again-later",' + '"detail":"A sync is already running for: connection-id"}' + ) + sdk_error = SDKError("Status 409", 409, body, MagicMock()) + sync_error = _sync_error(409) + sync_error.__cause__ = sdk_error def run_connection(**kwargs: object) -> None: """Raise the wrapped 409 sync error.""" _ = kwargs - raise _sync_error(409) + raise sync_error monkeypatch.setattr(api_util, "run_connection", run_connection) - fetch_mock = MagicMock(return_value=_connection_info_with_status("inactive")) + fetch_mock = MagicMock(return_value=_connection_info_with_status(connection_status)) monkeypatch.setattr(connection, "_fetch_connection_info", fetch_mock) - with pytest.raises(AirbyteLibInputError) as exc_info: - connection.run_sync() + if scenario == "running-job": + sync_result = MagicMock( + job_id=42, + job_url="https://cloud.airbyte.com/jobs/42", + is_job_complete=MagicMock(return_value=False), + ) + get_logs = MagicMock(return_value=[sync_result]) + monkeypatch.setattr(connection, "get_previous_sync_logs", get_logs) + + with pytest.raises(AirbyteConnectionSyncActiveError) as exc_info: + connection.run_sync() + + assert "job 42" in exc_info.value.get_message() + assert exc_info.value.context == { + "connection_id": "connection-id", + "job_id": 42, + "job_url": "https://cloud.airbyte.com/jobs/42", + } + assert "`get_cloud_sync_status`" in exc_info.value.guidance + get_logs.assert_called_once_with(limit=1, job_type=JobTypeEnum.SYNC) + assert exc_info.value.__cause__ is sync_error + elif scenario in {"no-jobs", "completed-job"}: + sync_results = ( + [] + if scenario == "no-jobs" + else [ + MagicMock( + job_id=41, + is_job_complete=MagicMock(return_value=True), + ) + ] + ) + monkeypatch.setattr( + connection, + "get_previous_sync_logs", + MagicMock(return_value=sync_results), + ) + + with pytest.raises(AirbyteConnectionSyncError) as exc_info: + connection.run_sync() + + assert exc_info.value is sync_error + else: + get_logs = MagicMock() + monkeypatch.setattr(connection, "get_previous_sync_logs", get_logs) + + with pytest.raises(AirbyteLibInputError) as exc_info: + connection.run_sync() - assert "disabled" in exc_info.value.message - assert "enabled=True" in exc_info.value.guidance + assert "disabled" in exc_info.value.message + assert "enabled=True" in exc_info.value.guidance + get_logs.assert_not_called() fetch_mock.assert_called_once_with(force_refresh=True) diff --git a/tests/unit_tests/test_mcp_tool_errors.py b/tests/unit_tests/test_mcp_tool_errors.py index 5a1aa2799..4e7aed5a6 100644 --- a/tests/unit_tests/test_mcp_tool_errors.py +++ b/tests/unit_tests/test_mcp_tool_errors.py @@ -16,6 +16,7 @@ from airbyte._util import api_util from airbyte.exceptions import ( AirbyteCloudApiError, + AirbyteConnectionSyncActiveError, AirbyteConnectionSyncError, AirbyteConnectionSyncTimeoutError, AirbyteConnectorInUseError, @@ -27,6 +28,7 @@ MCP_TOOL_USER_FACING_ERRORS, AgentErrorTextMiddleware, agent_error_text, + cloud_error_trace_message, redact_agent_text, classify_mcp_tool_error, format_user_facing_error, @@ -511,6 +513,8 @@ def test_cloud_chain_uses_exact_message_for_agent_and_span( ) error.__cause__ = requests.HTTPError(response=response) expected = f"{message} {guidance}".rstrip() if guidance else message + if carrier == "nested": + expected = f"local override local advice Airbyte Cloud response: {message}" assert format_user_facing_error(error) == expected tool = ToolError("ignored wrapper") tool.__cause__ = error @@ -518,6 +522,110 @@ def test_cloud_chain_uses_exact_message_for_agent_and_span( assert cloud_error_trace_message(tool) == message +@pytest.mark.parametrize( + ("local_message", "cloud_message", "expected"), + [ + pytest.param( + "Connection 'connection-id' is disabled (status 'inactive'), so a sync cannot " + "be started.", + "State conflict", + "Connection 'connection-id' is disabled (status 'inactive'), so a sync cannot " + "be started. Re-enable the connection first (e.g. " + "`connection.set_enabled(enabled=True)`, or the " + "`update_cloud_connection` MCP tool with `enabled=True`), then retry. " + "Airbyte Cloud response: State conflict", + id="disabled-connection", + ), + pytest.param( + "State conflict", + "State conflict", + "State conflict Local advice.", + id="cloud-message-already-present", + ), + ], +) +def test_local_guidance_is_preserved_with_cloud_error( + local_message: str, + cloud_message: str, + expected: str, +) -> None: + body = json.dumps({ + "type": "https://reference.airbyte.com/reference/errors", + "title": "state-conflict", + "detail": cloud_message, + }) + response = requests.Response() + response.status_code = 409 + response._content = body.encode() + sdk = SDKError("ignored SDK message", 409, body, response) + sync_error = AirbyteConnectionSyncError( + connection_id="connection-id", + context={"status_code": 409}, + ) + sync_error.__cause__ = sdk + error = AirbyteLibInputError( + message=local_message, + guidance=( + "Re-enable the connection first (e.g. " + "`connection.set_enabled(enabled=True)`, or the " + "`update_cloud_connection` MCP tool with `enabled=True`), then retry." + if "disabled" in local_message + else "Local advice." + ), + ) + error.__cause__ = sync_error + + assert format_user_facing_error(error) == expected + tool_error = ToolError("unformatted") + tool_error.__cause__ = error + assert agent_error_text(tool_error) == expected + + assert cloud_error_trace_message(error) == cloud_message + + +def test_active_sync_error_is_a_user_facing_mcp_error() -> None: + connection_id = "connection-id" + job_id = 42 + message = ( + f"A sync is already running for connection '{connection_id}' (job {job_id})." + ) + guidance = "Monitor it with `get_cloud_sync_status`." + body = json.dumps({ + "type": "https://reference.airbyte.com/reference/errors#try-again-later", + "title": "try-again-later", + "detail": f"A sync is already running for: {connection_id}", + }) + response = requests.Response() + response.status_code = 409 + response._content = body.encode() + sdk = SDKError("ignored SDK message", 409, body, response) + error = AirbyteConnectionSyncActiveError( + connection_id=connection_id, + message=message, + guidance=guidance, + context={"connection_id": connection_id, "job_id": job_id}, + ) + error.__cause__ = sdk + + server = mcp_server( + "test", + telemetry=False, + user_facing_errors=MCP_TOOL_USER_FACING_ERRORS, + user_facing_error_formatter=format_user_facing_error, + ) + + @server.tool + def running_sync() -> None: + raise error + + text = _call_error(server, "running_sync") + + assert text == ( + f"{message} {guidance} Airbyte Cloud response: " + f"A sync is already running for: {connection_id}" + ) + + @pytest.mark.parametrize("field", ["detail", "title"]) @pytest.mark.parametrize("scheme", ["Bearer", "bearer", "BEARER", "bEaReR"]) def test_real_middleware_redacts_selected_cloud_message_after_nested_rewrap(