[Bug] dbt INVOCATION_COMMAND reports the Airflow worker's argv instead of the dbt command when using InvocationMode.DBT_RUNNER #2969

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

1. Airflow environment with dbt-core importable (so InvocationMode.DBT_RUNNER is auto-selected), Celery executor.
2. dbt project with an on-run-end hook that inserts {{ flags.INVOCATION_COMMAND }} into an audit table.
3. Run any Cosmos DbtDag / DbtTaskGroup without setting invocation_mode explicitly.
4. 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)