[Anatolii Yatsuk (tolik0)]

What

Concurrent sources emitted stream-error TRACE messages without masking config secrets. This PR sanitizes them the same way the non-concurrent path does.

• ConcurrentReadProcessor.on_exception now yields as_sanitized_airbyte_message on both branches: an AirbyteTracedException raised by the stream (as_sanitized_airbyte_message(stream_descriptor=...)), and an unhandled exception wrapped with from_exception(..., stream_descriptor=...) (as_sanitized_airbyte_message(), since from_exception already sets the descriptor). The stream descriptor and the user-facing message are unchanged.
• AsyncJobOrchestrator._keep_api_budget_with_failed_job emits its job-creation failure trace through the message repository, which also reaches read output. It now uses as_sanitized_airbyte_message() too.

Production impact

AbstractSource (the non-concurrent path) already emits stream errors with as_sanitized_airbyte_message, which runs filter_secrets over message, internal_message and stack_trace. entrypoint.launch() prints every message from run() as-is, with no further filtering. So for concurrent sources, which include most declarative/manifest-only connectors, any stream error whose text quoted a config secret reached the platform unmasked: in the sync's failure reason, in the job logs, and in whatever the platform forwards (Sentry, attempt failure summaries).

The concrete case is source-tiktok-marketing: its oauth2/advertiser/get/?secret=...&app_id=... request quotes the app secret in the URL. When that request returns 403, the HTTPError internal message and stack trace contain the full URL, and the trace was emitted with the secret in plain text. After this change, those fields show **** instead of the secret. Manifest-only connectors such as source-tiktok-marketing pin their SDM base image, so they get the fix with their next base-image bump.

Nothing changes for messages that contain no registered secrets. filter_secrets only replaces values registered by set_up_secret_filter (the spec's airbyte_secret fields).

Paths checked and left as is

• The final "streams did not sync successfully" failure raised from ConcurrentReadProcessor.is_done goes through sys.excepthook, then AirbyteTracedException.emit_message(), which already runs filter_secrets over the serialized message. Its message is also built by generate_failed_streams_error_message, which filters each exception repr.
• connector_builder/* and test/entrypoint_wrapper.py still use as_airbyte_message(). They are not part of the launch()/read output path, so they are out of scope here.
• cli/source_declarative_manifest/_run.py prints startup-error traces before the secret filter is set up, so filter_secrets would not mask anything there.

Tests

• test_on_exception_masks_config_secrets_in_trace_message in unit_tests/sources/streams/concurrent/test_concurrent_read_processor.py, parametrized over both branches. It registers a secret with update_secrets, raises a stream exception whose text contains the secret, and asserts that the yielded TRACE has **** and not the secret in message (ATE case), internal_message and stack_trace. It also asserts that the stream descriptor and the user-facing message are preserved.
• test_given_exception_when_start_job_then_emitted_trace_masks_config_secrets in unit_tests/sources/declarative/async_job/test_job_orchestrator.py.

Mutation pins (run on a scratch copy):

Mutation	Result
ATE branch back to as_airbyte_message(stream_descriptor=...)	airbyte_traced_exception case fails
Non-ATE branch back to .as_airbyte_message()	unhandled_exception case fails
Orchestrator back to as_airbyte_message()	orchestrator test fails

Local run: unit_tests/sources/streams/concurrent, unit_tests/sources/concurrent_source, unit_tests/sources/declarative/async_job and unit_tests/utils/test_traced_exception.py all pass (359 passed).

Related and overlapping PRs

• feat(standard-tests): run the secret leak check on discover and read output #1267 (standard-tests leak check on discover and read) is the test-harness side of this work. Its tiktok dry-run found the leak this PR fixes. The two PRs share no files and merge cleanly, and no merge order is needed: tiktok's test_basic_read is red with or without either PR, because its test app gets a 403. Once this PR is released, the tiktok paragraph in feat(standard-tests): run the secret leak check on discover and read output #1267's cdk-migrations.md entry, which says the concurrent read emits stream-error traces unfiltered, needs to point to this fix.
• fix(low-code): stop leaking interpolation context in undeclared-variable errors #1241 (Devin draft) contains the same two hunks, bundled with an unrelated jinja interpolation change. The hunks are now identical, so the two PRs merge cleanly in either order.
• git merge-tree shows new textual conflicts with fix(oauth): surface token refresh retry exhaustion as transient_error #1225 (concurrent_read_processor.py), fix(async-job): preserve async and stream failure types #994 (job_orchestrator.py) and fix: treat 429 retry budget exhaustion as terminal in async job orchestrator #1048 (test_job_orchestrator.py, test additions only). Each of these is a nearby-line conflict that is quick to resolve. fix(async-job): propagate underlying failure_type through orchestrator wrappers #1002 already conflicts with main in the same files.

🤖 Generated with Claude Code

[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@fix/concurrent-sanitize-stream-error-traces#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 fix/concurrent-sanitize-stream-error-traces

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)

6 489 tests  +183   6 453 ✅ +168   7m 13s ⏱️ + 1m 41s
    1 suites ±  0      36 💤 + 15 
    1 files   ±  0       0 ❌ ±  0 

Results for commit c3b7736. ± Comparison against base commit e97f461.

This pull request removes 1 and adds 184 tests. Note that renamed tests count towards both.

unit_tests.test.test_scenario_sections ‑ test_unknown_section_is_rejected[incremental]

unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_basic_read[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_check[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_config_matches_spec[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_discover[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_docker_image_build_and_check[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_docker_image_build_and_config_matches_spec[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_docker_image_build_and_discover[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_docker_image_build_and_read[bypassed-incremental]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_docker_image_build_and_read_from_future_state['invalid_config' Test Scenario]
unit_tests.resources.source_pokeapi_w_components_py.integration_tests.test_airbyte_standards.TestSuiteSourcePokeAPI ‑ test_docker_image_build_and_read_from_future_state['valid_config' Test Scenario]
…

♻️ This comment has been updated with latest results.

[github-actions]

PyTest Results (Full)

6 492 tests   6 456 ✅  10m 55s ⏱️
    1 suites     36 💤
    1 files        0 ❌

Results for commit c3b7736.

♻️ This comment has been updated with latest results.

[Daryna Ishchenko (darynaishchenko)]

Approve. Correct, minimal fix, and the change is not breaking.

Verified

• End-to-end probe through AirbyteEntrypoint.run with a concurrent declarative stream that has an airbyte_secret in the query string and in a 403 error message. On main, the secret leaks in message, internal_message and stack_trace. At this head, all three are ****.
• I searched the rest of the read path for unmasked error messages. Only connector_builder/* and test/entrypoint_wrapper.py remain, and the PR body already scopes those out. emit_message (the excepthook path) already filters.
• Mutation pins re-run locally: each of the 3 reverts fails exactly its test. 359 tests pass in the touched suites.

Risk: low. It uses the same sanitizer AbstractSource already uses, and nothing changes when an error contains no registered secret. The Check: destination-motherduck failure is unrelated: its GSM secret has no enabled version, and that check is red on every recent PR.

One residual limitation, not caused by this PR: filter_secrets is exact-match only, so a secret that appears percent-encoded in a URL is still not masked.