Fix task-level audit logs missing success/running events in Airflow 3.1.x

[sjyangkevin]

closes: #58381

Issue

In Airflow 2, task instance state transition such as RUNNING, SUCCESS, SKIPPED, FAILED were logged into the log table through the TaskInstance model methods. In Airflow 3, the task state update/management is moved to the execution API endpoints ti_run and ti_update_state. However, the endpoints were not wired up to create log records, resulting in missing audit logs for task instance state transitions.

Fix

Added session.add(Log(...)) calls to both ti_run and ti_update_state when the task instance state is updated. The call will be executed in the same transaction as the state update in both endpoints.

In ti_run, the audit log is placed inside the else branch such that duplicated requests, or conflict request, due to network glitch, will not be logged.

In ti_update_state, the TI select query is updated to fetch task_id, run_id, and map_index.

Caveat

The following fields in the log table are missing. Require extra query/join to fetch the following information.

• logical_date (previously execution_date)
• owner (the value is airflow in Airflow 2)
• owner_display_name (this field is also empty for task state transition)
• extra (full_command will not be available as it is not run/update through CLI; hostname is not available in ti_update_state)

EmptyOperator or operators skipped by branch doesn't have audit log

---

DAG Test Samples

Edited on Feb 27, 2026

Gather data for owner, and logical_date from DagModel and DagRun, and fill extra with host_name.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: [Antigravity] following the guidelines

The tool is used to analyze the existing test cases to understand the expected behavior of both ti_run and ti_update_state. It is also used to generate new test cases.

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst or {issue_number}.significant.rst, in airflow-core/newsfragments.

[amoghrajesh]

Thanks for taking this on @sjyangkevin, some comments

[sjyangkevin]

Hi @amoghrajesh , thanks for the feedback!

I wonder if I should create dedicated tests for those situations (e.g., on duplicate requests, conflict) instead of putting them into the existing tests. I feel those might be good to catch.

I will parameterize the payload and probably task instance param to reduce duplication in the state change test case.

Also, will check how to handle the e2e test.

Thanks!

[sjyangkevin]

I've cleaned up the test cases, and merge them into a single parameterized one. I feel we don't actually need to set up the time_machine such that the test structure can be unified by parameterizing the payload and state.

Still trying to understand the end-to-end testing failure as I am not very familiar with the context. At high-level understanding, it seems like the test expect the User field to be not None, but will verify it. Would appreciate if anyone familiar with this can provide some insights.

Thanks!

[sjyangkevin]

I updated the query to join with DagModel and DagRun to fetch logical_date and owner, and update with_for_update to lock only TI in the join query. Construct a JSON string only with host_name in extra, which is also present in earlier version of Airflow.

[sjyangkevin]

Consolidate all the test cases into a parameterized one.

[sjyangkevin]

because the TI query return more values, update the mock.

[potiuk]

LGTM, I would need more maintainers to confirm though