From f4b6b33f78646cd6a9a22610aa8207310a1ad183 Mon Sep 17 00:00:00 2001
From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
Date: Thu, 30 Jul 2026 06:16:44 +0000
Subject: [PATCH] fix: emit CONTROL config messages as a single atomic stdout
 write

Co-Authored-By: bot_apk <apk@cognition.ai>
---
 airbyte_cdk/config_observation.py     | 10 +++-
 unit_tests/test_config_observation.py | 72 +++++++++++++++++++++++++++
 2 files changed, 81 insertions(+), 1 deletion(-)

diff --git a/airbyte_cdk/config_observation.py b/airbyte_cdk/config_observation.py
index ae85e82773..325124edc7 100644
--- a/airbyte_cdk/config_observation.py
+++ b/airbyte_cdk/config_observation.py
@@ -6,6 +6,7 @@
     annotations,
 )
 
+import sys
 import time
 from copy import copy
 from typing import Any, List, MutableMapping
@@ -92,7 +93,14 @@ def emit_configuration_as_airbyte_control_message(config: MutableMapping[str, An
     See the airbyte_cdk.sources.message package
     """
     airbyte_message = create_connector_config_control_message(config)
-    print(orjson.dumps(AirbyteMessageSerializer.dump(airbyte_message)).decode())
+    serialized_message = orjson.dumps(AirbyteMessageSerializer.dump(airbyte_message)).decode()
+    # Emit the payload and its trailing newline in a single write. A bare `print(x)` issues two
+    # separate writes (the payload, then "\n"), so a CONTROL message emitted from a worker thread
+    # (e.g. a single-use refresh-token rotation during a concurrent sync) can interleave with a
+    # RECORD line printed from the main thread and corrupt stdout line framing, causing the
+    # platform to drop a record. Writing the newline as part of the payload keeps the emission
+    # line-atomic, mirroring the main read loop in `entrypoint.launch`.
+    sys.stdout.write(f"{serialized_message}\n")
 
 
 def create_connector_config_control_message(config: MutableMapping[str, Any]) -> AirbyteMessage:
diff --git a/unit_tests/test_config_observation.py b/unit_tests/test_config_observation.py
index 71bccd628b..7e8d29cd5f 100644
--- a/unit_tests/test_config_observation.py
+++ b/unit_tests/test_config_observation.py
@@ -3,7 +3,10 @@
 #
 
 import json
+import sys
 import time
+from io import StringIO
+from threading import Barrier, Thread
 
 import pytest
 
@@ -11,9 +14,11 @@
     ConfigObserver,
     ObservedDict,
     create_connector_config_control_message,
+    emit_configuration_as_airbyte_control_message,
     observe_connector_config,
 )
 from airbyte_cdk.models import AirbyteControlConnectorConfigMessage, OrchestratorType, Type
+from airbyte_cdk.utils.print_buffer import PrintBuffer
 
 
 class TestObservedDict:
@@ -97,3 +102,70 @@ def test_create_connector_config_control_message():
     assert message.control.type == OrchestratorType.CONNECTOR_CONFIG
     assert message.control.connectorConfig == AirbyteControlConnectorConfigMessage(config=A_CONFIG)
     assert message.control.emitted_at is not None
+
+
+def test_emit_configuration_as_airbyte_control_message_is_line_atomic(monkeypatch):
+    writes = []
+
+    class RecordingStream:
+        def write(self, message):
+            writes.append(message)
+
+    monkeypatch.setattr(sys, "stdout", RecordingStream())
+
+    emit_configuration_as_airbyte_control_message({"foo": "bar"})
+
+    assert len(writes) == 1
+    assert writes[0].endswith("\n")
+    assert writes[0].count("\n") == 1
+    assert json.loads(writes[0])["type"] == "CONTROL"
+
+
+def test_emit_configuration_as_airbyte_control_message_concurrent_output_is_well_framed(
+    monkeypatch,
+):
+    captured_output = []
+    print_buffer = PrintBuffer(flush_interval=float("inf"))
+
+    def capture_flush():
+        captured_output.append(print_buffer.buffer.getvalue())
+        print_buffer.buffer = StringIO()
+
+    monkeypatch.setattr(print_buffer, "flush", capture_flush)
+    monkeypatch.setattr(sys, "stdout", print_buffer)
+
+    worker_count = 4
+    iterations = 100
+    start_barrier = Barrier(worker_count + 1)
+
+    def print_records():
+        start_barrier.wait()
+        for index in range(worker_count * iterations):
+            record = {"type": "RECORD", "record": {"data": {"index": index}}}
+            print(f"{json.dumps(record)}\n", end="")
+
+    def emit_control_messages(worker_id):
+        start_barrier.wait()
+        for index in range(iterations):
+            emit_configuration_as_airbyte_control_message({"worker_id": worker_id, "index": index})
+
+    threads = [
+        Thread(target=print_records),
+        *[
+            Thread(target=emit_control_messages, args=(worker_id,))
+            for worker_id in range(worker_count)
+        ],
+    ]
+    for thread in threads:
+        thread.start()
+    for thread in threads:
+        thread.join()
+    print_buffer.flush()
+
+    output = "".join(captured_output)
+    lines = output.split("\n")
+    assert lines[-1] == ""
+    lines = lines[:-1]
+    assert lines
+    assert all(line for line in lines)
+    assert all(json.loads(line) for line in lines)