Logs produced by deferrable tasks disapper from Airflow UI after completion of the task

### Under which category would you file this issue?

Airflow Core

### Apache Airflow version

3.3.0

### What happened and how to reproduce it?

**Issue Description**

This issue is related to https://github.com/apache/airflow/issues/70314. Even though I was able to implement a workaround that sends logs produced by deferrable triggers to AWS CloudWatch, Airflow UI doesn't seem to be able to display them after completion of the task.

The way it works at the moment in 3.3.0 with remote logging configured to send logs to AWS CloudWatch is that Airflow UI pulls trigger logs for deferrable tasks from the triggerer machine via HTTP while the task is running in deferred state. However, once the task completes these logs disappear from Airflow UI.

**Issue Analysis**

I don't know much about internals of Airflow, but this is what my AI assistant and I were able to identify while debugging this problem (note that it assumes that trigger logs were successfully sent to AWS CloudWatch via the workaround listed in https://github.com/apache/airflow/issues/70314).

> When the UI fetches logs for a completed deferrable task, [`FileTaskHandler._read_remote_logs()`](https://github.com/apache/airflow/blob/3.3.0/airflow-core/src/airflow/utils/log/file_task_handler.py#L1002-L1005) calls `CloudWatchRemoteLogIO.stream(path, ti)` with the base task log path (e.g. `dag_id=.../attempt=1.log`). However, trigger logs are written to a separate CloudWatch stream with a [`.trigger.{job_id}.log` suffix](https://github.com/apache/airflow/blob/3.3.0/airflow-core/src/airflow/jobs/triggerer_job_runner.py#L861), and [`CloudWatchRemoteLogIO.stream()`](https://github.com/apache/airflow/blob/3.3.0/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py#L199-L213) reads exactly one stream — the one it's given — with no discovery of related trigger streams. The local-file equivalent doesn't have this problem because [`_read_from_local()`](https://github.com/apache/airflow/blob/3.3.0/airflow-core/src/airflow/utils/log/file_task_handler.py#L884) uses `glob(worker_log_path.name + "*")` which naturally picks up both the base log and any `.trigger.*.log` files. CloudWatch has no glob, so `stream()` needs an explicit `DescribeLogStreams` prefix scan to find trigger streams.

**Steps to reproduce**

I've created a docker compose simulation that can be used to reproduce the problem.

**Prerequisites:** the simulation expects one to have an AWS account with privileges needed to write logs to a CloudWatch group.

**File: ./Dockerfile**
```
FROM apache/airflow:3.3.0-python3.13

USER airflow

RUN pip install --no-cache-dir \
apache-airflow-providers-amazon==9.31.0 \
apache-airflow-providers-standard==1.16.0
```

**File: ./docker-compose.yml**
```
x-airflow-common:
&airflow-common
build: .
environment:
&airflow-common-env
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN: postgresql+psycopg2://airflow:airflow@postgres:5432/airflow
AIRFLOW__CORE__EXECUTOR: LocalExecutor
AIRFLOW__CORE__FERNET_KEY: 'olFLXR4nU_6WBvIsSDPIDCKWjh3OCWSMfWjt6XU0VfM='
AIRFLOW__CORE__EXECUTION_API_SERVER_URL: http://airflow-webserver:8080/execution/
AIRFLOW__CORE__SIMPLE_AUTH_MANAGER_ALL_ADMINS: 'True'
AIRFLOW__API__SECRET_KEY: reproduction-secret-key-not-for-production
AIRFLOW__API_AUTH__JWT_SECRET: reproduction-jwt-secret-not-for-production
AIRFLOW__API_AUTH__JWT_ALGORITHM: HS512
AIRFLOW__LOGGING__REMOTE_LOGGING: 'True'
AIRFLOW__LOGGING__REMOTE_BASE_LOG_FOLDER: 'cloudwatch://arn:aws:logs:${AWS_REGION}:${AWS_ACCOUNT_ID}:log-group:${AWS_LOG_GROUP_NAME}'
AIRFLOW__LOGGING__REMOTE_LOG_CONN_ID: aws_default
AIRFLOW__LOGGING__DELETE_LOCAL_LOGS: 'True'
AIRFLOW__LOGGING__LOGGING_LEVEL: INFO
AWS_DEFAULT_REGION: ${AWS_REGION}
AWS_REGION: ${AWS_REGION}
AWS_ACCESS_KEY_ID: ${AWS_ACCESS_KEY_ID}
AWS_SECRET_ACCESS_KEY: ${AWS_SECRET_ACCESS_KEY}
AWS_SESSION_TOKEN: ${AWS_SESSION_TOKEN:-}
PYTHONPATH: '/opt/airflow/config:/opt/airflow/dags'
PYTHONUNBUFFERED: '1'
volumes:
- ./dags:/opt/airflow/dags
- ./config:/opt/airflow/config
depends_on:
postgres:
condition: service_healthy

services:
postgres:
image: postgres:16
environment:
POSTGRES_USER: airflow
POSTGRES_PASSWORD: airflow
POSTGRES_DB: airflow
healthcheck:
test: ["CMD-SHELL", "pg_isready -U airflow"]
interval: 10s
timeout: 5s
retries: 5

airflow-init:
<<: *airflow-common
entrypoint: /bin/bash
command:
- -c
- |
set -ex
airflow db migrate
airflow connections create-default-connections
echo "Airflow init complete."
environment:
<<: *airflow-common-env
TASK_NAME: init

airflow-scheduler:
<<: *airflow-common
command: airflow scheduler
environment:
<<: *airflow-common-env
TASK_NAME: scheduler
depends_on:
airflow-init:
condition: service_completed_successfully

airflow-webserver:
<<: *airflow-common
command: airflow api-server
ports:
- "8080:8080"
environment:
<<: *airflow-common-env
TASK_NAME: webserver
depends_on:
airflow-init:
condition: service_completed_successfully

airflow-dag-processor:
<<: *airflow-common
command: airflow dag-processor
environment:
<<: *airflow-common-env
TASK_NAME: dag-processor
depends_on:
airflow-init:
condition: service_completed_successfully

airflow-triggerer:
<<: *airflow-common
command: airflow triggerer
environment:
<<: *airflow-common-env
TASK_NAME: triggerer
depends_on:
airflow-init:
condition: service_completed_successfully
```

**File: ./dags/test_triggerer_logging.py**
```
from datetime import datetime, timedelta

from airflow.sdk import DAG
from airflow.providers.standard.operators.python import PythonOperator
from airflow.providers.standard.sensors.time_delta import TimeDeltaSensor

with DAG(
dag_id="test_triggerer_logging",
schedule=None,
start_date=datetime(2020, 1, 1),
catchup=False
) as dag:

before = PythonOperator(
task_id="before_deferral",
python_callable=lambda: print("Pre-deferral task log."))

# Logs produced by this task will be missing in CloudWatch
test_triggerer = TimeDeltaSensor(
task_id="test_triggerer",
delta=timedelta(seconds=120),
deferrable=True)

after = PythonOperator(
task_id="after_deferral",
python_callable=lambda: print("Post-deferral task log."))

before >> test_triggerer >> after
```

**File: ./config/airflow_local_settings.py**
```
# Note: workaround from #70314 to send trigger logs to AWS CloudWatch

import os

def _patch_dictconfig_for_watchtower():
import logging
import logging.config
import watchtower

_original_dictConfig = logging.config.dictConfig

def _dictConfig_hide_watchtower(config):
saved = []
remaining = []
for ref in logging._handlerList:
h = ref()
if h is not None and isinstance(h, watchtower.CloudWatchLogHandler):
saved.append(ref)
else:
remaining.append(ref)

logging._handlerList[:] = remaining
try:
_original_dictConfig(config)
finally:
logging._handlerList.extend(saved)

logging.config.dictConfig = _dictConfig_hide_watchtower

if os.environ.get("TASK_NAME") == "triggerer":
_patch_dictconfig_for_watchtower()
```

**File: ./.env**
```
# AWS region and account ID
AWS_REGION=change_me
AWS_ACCOUNT_ID=change_me

# AWS credentials — ECS injects these via task role; Docker needs them explicitly.
# Populate via: aws configure export-credentials --profile change_me --format env
AWS_ACCESS_KEY_ID=change_me
AWS_SECRET_ACCESS_KEY=change_me
AWS_SESSION_TOKEN=change_me
AWS_LOG_GROUP_NAME=change_me
```

Trigger execution of the `test_triggerer_logging` DAG, then look at the logs tab of the `test_triggerer` task. You'll see that logs produced by the trigger gradually appear over time in Airflow UI while the task is running. However, once execution of the task is completed these logs disappear from Airflow UI. Logs produced by non-deferrable tasks remain in place after completion of those tasks.

Checking AWS CloudWatch reveals that trigger logs have been preserved with correct stream name right next to logs produced by non-deferrable tasks. So the issue is that Airflow UI isn't pulling trigger logs from AWS CloudWatch when it is configured to use remote logging.

### What you think should happen instead?

Logs produced by triggers in deferrable tasks should remain visible in Airflow UI after the deferrable task is complete.

### Operating System

Docker with apache/airflow:3.3.0-python3.13 image

### Deployment

Other

### Apache Airflow Provider(s)

_No response_

### Versions of Apache Airflow Providers

apache-airflow-providers-amazon==9.31.0
apache-airflow-providers-standard==1.16.0

### Official Helm Chart version

Not Applicable

### Kubernetes Version

_No response_

### Helm Chart configuration

_No response_

### Docker Image customizations

See the supplied `Dockerfile` and `docker-compose.yml` in the "how to reproduce it" section.

### Anything else?

I was able to find a workaround that works via monkey-patching. Even though it seems to work, it is pretty fragile as it can break at any point if Airflow decides to change something in future versions.

**Add** the following code to the `./config/airflow_local_settings.py` file and re-start Docker Compose:
```
def _patch_cloudwatch_stream_for_trigger_logs():
from airflow.providers.amazon.aws.log.cloudwatch_task_handler import CloudWatchRemoteLogIO
from airflow.utils.state import TaskInstanceState

_original_stream = CloudWatchRemoteLogIO.stream

def _stream_with_trigger_logs(self, relative_path, ti):
sources, logs = _original_stream(self, relative_path, ti)

if getattr(ti, "state", None) == TaskInstanceState.DEFERRED:
return sources, logs

try:
trigger_prefix = relative_path.replace(":", "_") + ".trigger."
response = self.hook.conn.describe_log_streams(
logGroupName=self.log_group,
logStreamNamePrefix=trigger_prefix,
)
for stream_info in response.get("logStreams", []):
stream_name = stream_info["logStreamName"]
trigger_sources, trigger_logs = _original_stream(self, stream_name, ti)
sources.extend(trigger_sources)
logs.extend(trigger_logs)
except Exception:
pass

return sources, logs

CloudWatchRemoteLogIO.stream = _stream_with_trigger_logs

if os.environ.get("TASK_NAME") == "webserver":
_patch_cloudwatch_stream_for_trigger_logs()
```

The idea of this patch is to use `describe_log_streams` as a replacement for `glob()` to find all AWS CloudWatch streams that are related to the task. So that Airflow UI can display all of them, and not only those produced by non-deferrable tasks.

### Are you willing to submit PR?

- [ ] Yes I am willing to submit a PR!

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)

Start with airflow/utils/log/file_task_handler.py, especially _read_remote_logs() and _read_from_local(), then inspect CloudWatchRemoteLogIO.stream() in airflow/providers/amazon/src/airflow/providers/amazon/aws/log/cloudwatch_task_handler.py and the trigger stream naming in triggerer_job_runner.py. Run the supplied Docker Compose reproduction with CloudWatch configured. Done means trigger logs remain visible in the Airflow UI after a deferrable task completes.

Written by the indexing model from the issue text.