Bug description
When Cosmos invokes dbt in-process via InvocationMode.DBT_RUNNER, dbt's flags.INVOCATION_COMMAND resolves to the host process's command line — e.g. the Celery worker startup command — rather than the dbt command Cosmos actually ran.
Users who log {{ flags.INVOCATION_COMMAND }} from an on-run-end hook (or via packages such as dbt_artifacts / Elementary that persist invocation metadata) get rows like:
dbt celery worker --queues dbt-producer-queue -c 6 --without-gossip --without-mingle
instead of the expected:
dbt --log-format json --debug build -s tag:MY_TAG --vars {"dag_id": "..."} --target PROD
This makes the invocation-command column useless for auditing which dbt selector / target / vars produced a given run.
Root cause
dbt-core builds the value from sys.argv, unconditionally, with no dbtRunner-aware path (dbt/cli/flags.py, dbt-core 1.11.x — same in earlier 1.x):
# Add entire invocation command to flags
object.__setattr__(self, "INVOCATION_COMMAND", "dbt " + " ".join(sys.argv[1:]))
Cosmos calls dbtRunner.invoke(cli_args) in the Airflow task process and never adjusts sys.argv, so dbt sees the worker's argv (cosmos/dbt/runner.py::run_command).
This is not specific to ExecutionMode.WATCHER. Any code path that lands on InvocationMode.DBT_RUNNER is affected. It is easy to hit unknowingly because DbtLocalBaseOperator._discover_invocation_mode() (cosmos/operators/local.py) auto-selects DBT_RUNNER whenever dbt-core is importable in the Airflow environment, and DbtProducerWatcherOperator intentionally leaves that discovery in place. Reported by a user on Cosmos 1.14.2 who noticed it after adopting WATCHER, but it reproduces on main and predates WATCHER.
How to reproduce
- Airflow environment with
dbt-core importable (so InvocationMode.DBT_RUNNER is auto-selected), Celery executor.
- dbt project with an
on-run-end hook that inserts {{ flags.INVOCATION_COMMAND }} into an audit table.
- Run any Cosmos
DbtDag / DbtTaskGroup without setting invocation_mode explicitly.
- The audit table records the Celery worker command line, not the dbt command.
Confirm the mode from the task log — it prints either dbtRunner is available. Using dbtRunner for invoking dbt. or Could not import dbtRunner. Falling back to subprocess for invoking dbt.
Expected behaviour
INVOCATION_COMMAND should match the dbt command Cosmos ran, so that DBT_RUNNER and SUBPROCESS produce equivalent invocation metadata.
Proposed fix
In cosmos/dbt/runner.py::run_command, temporarily set sys.argv to the dbt command for the duration of runner.invoke() and restore it afterwards. The command list is already available there (command[0] is the dbt executable, command[1:] are the CLI args passed to invoke), so the string dbt derives will match what SUBPROCESS mode would have produced.
This fits the existing structure: run_command already composes exclude_dags_folder_from_sys_path(), change_working_directory(cwd) and environ(env) around the invocation, so this would be one more restore-on-exit contextmanager next to them in cosmos/dbt/project.py.
Points to settle during implementation:
sys.argv is process-global. Restore must happen on every exit path (try/finally, as done today for _cleanup_dbt_adapters), and we should confirm behaviour for any in-process concurrent dbt invocations (e.g. dbt ls graph parsing in the DAG processor) rather than assume one-invocation-per-process.
- Decide whether this applies to all
run_command callers (including parsing-time dbt ls) or only operator execution.
Workaround
Set the invocation mode explicitly, which sends dbt through a real subprocess whose argv is the dbt command:
execution_config = ExecutionConfig(
execution_mode=ExecutionMode.WATCHER,
invocation_mode=InvocationMode.SUBPROCESS,
)
Trade-off: SUBPROCESS is measurably slower than DBT_RUNNER — the benchmark in #850, which introduced dbtRunner, showed ~8.4s vs ~23.8s for 10 models (roughly 1-2s of extra overhead per task). Fixing this in Cosmos would let users keep dbtRunner's speed and get the correct logged command.
Versions
- Cosmos: reproduced on 1.14.2 and on
main
- dbt-core: 1.x (checked against 1.11.11)
Bug description
When Cosmos invokes dbt in-process via
InvocationMode.DBT_RUNNER, dbt'sflags.INVOCATION_COMMANDresolves to the host process's command line — e.g. the Celery worker startup command — rather than the dbt command Cosmos actually ran.Users who log
{{ flags.INVOCATION_COMMAND }}from anon-run-endhook (or via packages such as dbt_artifacts / Elementary that persist invocation metadata) get rows like:instead of the expected:
This makes the invocation-command column useless for auditing which dbt selector / target / vars produced a given run.
Root cause
dbt-core builds the value from
sys.argv, unconditionally, with no dbtRunner-aware path (dbt/cli/flags.py, dbt-core 1.11.x — same in earlier 1.x):Cosmos calls
dbtRunner.invoke(cli_args)in the Airflow task process and never adjustssys.argv, so dbt sees the worker's argv (cosmos/dbt/runner.py::run_command).This is not specific to
ExecutionMode.WATCHER. Any code path that lands onInvocationMode.DBT_RUNNERis affected. It is easy to hit unknowingly becauseDbtLocalBaseOperator._discover_invocation_mode()(cosmos/operators/local.py) auto-selectsDBT_RUNNERwheneverdbt-coreis importable in the Airflow environment, andDbtProducerWatcherOperatorintentionally leaves that discovery in place. Reported by a user on Cosmos 1.14.2 who noticed it after adopting WATCHER, but it reproduces onmainand predates WATCHER.How to reproduce
dbt-coreimportable (soInvocationMode.DBT_RUNNERis auto-selected), Celery executor.on-run-endhook that inserts{{ flags.INVOCATION_COMMAND }}into an audit table.DbtDag/DbtTaskGroupwithout settinginvocation_modeexplicitly.Confirm the mode from the task log — it prints either
dbtRunner is available. Using dbtRunner for invoking dbt.orCould not import dbtRunner. Falling back to subprocess for invoking dbt.Expected behaviour
INVOCATION_COMMANDshould match the dbt command Cosmos ran, so thatDBT_RUNNERandSUBPROCESSproduce equivalent invocation metadata.Proposed fix
In
cosmos/dbt/runner.py::run_command, temporarily setsys.argvto the dbt command for the duration ofrunner.invoke()and restore it afterwards. The command list is already available there (command[0]is the dbt executable,command[1:]are the CLI args passed toinvoke), so the string dbt derives will match whatSUBPROCESSmode would have produced.This fits the existing structure:
run_commandalready composesexclude_dags_folder_from_sys_path(),change_working_directory(cwd)andenviron(env)around the invocation, so this would be one more restore-on-exit contextmanager next to them incosmos/dbt/project.py.Points to settle during implementation:
sys.argvis process-global. Restore must happen on every exit path (try/finally, as done today for_cleanup_dbt_adapters), and we should confirm behaviour for any in-process concurrent dbt invocations (e.g.dbt lsgraph parsing in the DAG processor) rather than assume one-invocation-per-process.run_commandcallers (including parsing-timedbt ls) or only operator execution.Workaround
Set the invocation mode explicitly, which sends dbt through a real subprocess whose argv is the dbt command:
Trade-off:
SUBPROCESSis measurably slower thanDBT_RUNNER— the benchmark in #850, which introduced dbtRunner, showed ~8.4s vs ~23.8s for 10 models (roughly 1-2s of extra overhead per task). Fixing this in Cosmos would let users keep dbtRunner's speed and get the correct logged command.Versions
main