fix: emit CONTROL config messages as a single atomic stdout write

[devin-ai-integration]

Summary

emit_configuration_as_airbyte_control_message() printed the CONTROL message with a bare print(x). CPython's print writes the payload and the trailing newline as two separate write() calls, so the emission is not line-atomic:

print(payload)          # -> write(payload); write("\n")

Since #877, SingleUseRefreshTokenOauth2Authenticator._emit_control_message always routes through this function so that rotated single-use refresh tokens are persisted immediately. In a concurrent declarative sync, token refreshes happen on the worker thread that made the HTTP request, while RECORD lines are printed from the main thread — which is line-atomic (entrypoint.launch deliberately does print(f"{message}\n", end="")). A worker-thread CONTROL emission can therefore interleave between a main-thread RECORD write:

write("<control payload>")   # worker thread
write("<record payload>\n")  # main thread  <-- lands mid-line
write("\n")                  # worker thread

The result is <control><record>\n\n: the platform rejects the concatenated line as Malformed non-Airbyte record and drops that RECORD, but the entrypoint's sourceStats.recordCount already counted it. Every subsequent state message then fails platform checksum validation with an off-by-one record count, so no state is committed at all and the sync ultimately fails.

The fix makes the emission line-atomic by writing the payload and its newline in one call, mirroring the main read loop:

-print(orjson.dumps(AirbyteMessageSerializer.dump(airbyte_message)).decode())
+serialized_message = orjson.dumps(AirbyteMessageSerializer.dump(airbyte_message)).decode()
+sys.stdout.write(f"{serialized_message}\n")

sys.stdout.write is used rather than print(..., end="") because CPython's print still issues a second (empty) write("") for end, which defeats a strict single-write guarantee. sys.stdout is resolved at call time, so this is transparent to the PrintBuffer redirection in entrypoint.launch.

This preserves the immediate-persistence guarantee that PR 877 was added for — the CONTROL message is still written straight to stdout from the refreshing thread, just without the framing hazard. ConfigObserver.update() benefits from the same fix.

Resolves https://github.com/airbytehq/oncall/issues/13201:

• https://github.com/airbytehq/oncall/issues/13201

Test Coverage

Both tests are in unit_tests/test_config_observation.py:

1. test_emit_configuration_as_airbyte_control_message_is_line_atomic — deterministic regression test. Monkeypatches sys.stdout with a stream that records every write() call and asserts exactly one write, ending in exactly one \n, parsing as a CONTROL message. This fails on main (2 writes) and passes with the fix.
2. test_emit_configuration_as_airbyte_control_message_concurrent_output_is_well_framed — end-to-end framing guard. Routes stdout through the real PrintBuffer and runs one record-printing thread against four concurrent control-emitting threads (800 lines total), then asserts every emitted line is non-empty and valid JSON — i.e. no concatenated or split lines.

7 passed, 3 warnings in 0.04s

ruff check, ruff format --check, and mypy --config-file mypy.ini airbyte_cdk (454 files) all pass.

Declarative-First Evaluation

Not applicable in the connector sense — this is a fix in CDK core stdout emission, not a connector manifest change. The affected connector (manifest-only) needs no changes; no declarative component can influence how the CDK frames protocol messages on stdout.

Breaking Change Evaluation

Not breaking. No schema, spec, state, primary-key, cursor, or stream changes. The only observable difference is that the CONTROL message's newline is written together with its payload instead of in a follow-up write; the bytes on stdout are identical. No public interface changed.

Link to Devin session: https://app.devin.ai/sessions/ad48da8684fc45e29826c13cd2c85c8e

[devin-ai-integration]

🤖 Devin AI Engineer

I'll be helping with this pull request! Here's what you should know:

✅ I will automatically:

• Address comments on this PR. Add '(aside)' to your comment to have me ignore it.
• Look at CI failures and help fix them

Note: I can only respond to comments from users who have write access to this repository.

⚙️ Control Options:

• Disable automatic comment, CI, and merge conflict monitoring

[github-actions]

👋 Greetings, Airbyte Team Member!

Here are some helpful tips and reminders for your convenience.

💡 Show Tips and Tricks

Testing This CDK Version

You can test this version of the CDK using the following:

# Run the CLI from this branch:
uvx 'git+https://github.com/airbytehq/airbyte-python-cdk.git@devin/1785392204-fix-control-message-line-atomic#egg=airbyte-python-cdk[dev]' --help

# Update a connector to use the CDK from this branch ref:
cd airbyte-integrations/connectors/source-example
poe use-cdk-branch devin/1785392204-fix-control-message-line-atomic

PR Slash Commands

Airbyte Maintainers can execute the following slash commands on your PR:

• /autofix - Fixes most formatting and linting issues
• /poetry-lock - Updates poetry.lock file
• /test - Runs connector tests with the updated CDK
• /prerelease - Triggers a prerelease publish with default arguments
• /poe build - Regenerate git-committed build artifacts, such as the pydantic models which are generated from the manifest JSON schema in YAML.
• /poe <command> - Runs any poe command in the CDK environment

📚 Show Repo Guidance

Helpful Resources

• CDK API Reference

📝 Edit this welcome message.

[github-actions]

PyTest Results (Fast)

4 141 tests  +2   4 129 ✅ +2   7m 47s ⏱️ +7s
    1 suites ±0      12 💤 ±0 
    1 files   ±0       0 ❌ ±0 

Results for commit f4b6b33. ± Comparison against base commit 73a9a55.

[github-actions]

PyTest Results (Full)

4 144 tests   4 132 ✅  12m 24s ⏱️
    1 suites     12 💤
    1 files        0 ❌

Results for commit f4b6b33.