Add API endpoint and UI for deadline callback log visibility

airflow-core/src/airflow/api_fastapi/core_api/datamodels/ui/deadline.py

@@ -25,6 +25,7 @@
 from pydantic import AliasPath, Field, field_validator
 
 from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.utils.state import CallbackState
 
 
 class DeadlineResponse(BaseModel):
@@ -38,6 +39,10 @@ class DeadlineResponse(BaseModel):
     dag_run_id: str = Field(validation_alias=AliasPath("dagrun", "run_id"))
     alert_id: UUID | None = Field(validation_alias="deadline_alert_id", default=None)
     alert_name: str | None = Field(validation_alias=AliasPath("deadline_alert", "name"), default=None)
+    callback_id: UUID | None = Field(validation_alias="callback_id", default=None)
+    callback_state: CallbackState | None = Field(
+        validation_alias=AliasPath("callback", "state"), default=None
+    )
 
 
 class DeadlineCollectionResponse(BaseModel):

airflow-core/src/airflow/api_fastapi/core_api/openapi/_private_ui.yaml

@@ -919,6 +919,60 @@ paths:
             application/json:
               schema:
                 $ref: '#/components/schemas/HTTPValidationError'
+  /ui/dags/{dag_id}/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs:
+    get:
+      tags:
+      - Deadlines
+      summary: Get Callback Logs
+      description: 'Get execution logs for a callback associated with a deadline.
+
+
+        Returns the logs produced during callback execution. These logs are uploaded
+
+        to remote storage (or written locally) by the callback supervisor after execution.'
+      operationId: get_callback_logs
+      security:
+      - OAuth2PasswordBearer: []
+      - HTTPBearer: []
+      parameters:
+      - name: dag_id
+        in: path
+        required: true
+        schema:
+          type: string
+          title: Dag Id
+      - name: dag_run_id
+        in: path
+        required: true
+        schema:
+          type: string
+          title: Dag Run Id
+      - name: callback_id
+        in: path
+        required: true
+        schema:
+          type: string
+          format: uuid
+          title: Callback Id
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/TaskInstancesLogResponse'
+        '404':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Not Found
+        '422':
+          description: Validation Error
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPValidationError'
   /ui/structure/structure_data:
     get:
       tags:
@@ -1936,6 +1990,17 @@ components:
       - count
       title: CalendarTimeRangeResponse
       description: Represents a summary of DAG runs for a specific calendar time range.
+    CallbackState:
+      type: string
+      enum:
+      - scheduled
+      - pending
+      - queued
+      - running
+      - success
+      - failed
+      title: CallbackState
+      description: All possible states of callbacks.
     ConfigResponse:
       properties:
         fallback_page_limit:
@@ -2582,6 +2647,16 @@ components:
           - type: string
           - type: 'null'
           title: Alert Name
+        callback_id:
+          anyOf:
+          - type: string
+            format: uuid
+          - type: 'null'
+          title: Callback Id
+        callback_state:
+          anyOf:
+          - $ref: '#/components/schemas/CallbackState'
+          - type: 'null'
       type: object
       required:
       - id
@@ -3570,6 +3645,21 @@ components:
       - nodes
       title: StructureDataResponse
       description: Structure Data serializer for responses.
+    StructuredLogMessage:
+      properties:
+        timestamp:
+          type: string
+          format: date-time
+          title: Timestamp
+        event:
+          type: string
+          title: Event
+      additionalProperties: true
+      type: object
+      required:
+      - event
+      title: StructuredLogMessage
+      description: An individual log message.
     TaskInstanceResponse:
       properties:
         id:
@@ -3838,6 +3928,28 @@ components:
       - awaiting_input
       title: TaskInstanceStateCount
       description: TaskInstance serializer for responses.
+    TaskInstancesLogResponse:
+      properties:
+        content:
+          anyOf:
+          - items:
+              $ref: '#/components/schemas/StructuredLogMessage'
+            type: array
+          - items:
+              type: string
+            type: array
+          title: Content
+        continuation_token:
+          anyOf:
+          - type: string
+          - type: 'null'
+          title: Continuation Token
+      type: object
+      required:
+      - content
+      - continuation_token
+      title: TaskInstancesLogResponse
+      description: Log serializer for responses.
     TeamCollectionResponse:
       properties:
         teams:

airflow-core/src/airflow/api_fastapi/core_api/routes/ui/deadlines.py

@@ -17,11 +17,13 @@
 
 from __future__ import annotations
 
+import os
 from typing import Annotated
+from uuid import UUID
 
 from fastapi import Depends, HTTPException, status
 from sqlalchemy import select
-from sqlalchemy.orm import contains_eager, noload
+from sqlalchemy.orm import contains_eager, joinedload, noload
 
 from airflow.api_fastapi.auth.managers.models.resource_details import DagAccessEntity
 from airflow.api_fastapi.common.db.common import SessionDep, paginated_select
@@ -35,16 +37,19 @@
     filter_param_factory,
 )
 from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.core_api.datamodels.log import TaskInstancesLogResponse
 from airflow.api_fastapi.core_api.datamodels.ui.deadline import (
     DeadlineAlertCollectionResponse,
     DeadlineCollectionResponse,
 )
 from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc
 from airflow.api_fastapi.core_api.security import ReadableDagRunsFilterDep, requires_access_dag
+from airflow.models.callback import Callback
 from airflow.models.dagrun import DagRun
 from airflow.models.deadline import Deadline
 from airflow.models.deadline_alert import DeadlineAlert
 from airflow.models.serialized_dag import SerializedDagModel
+from airflow.utils.log.callback_log_reader import read_callback_log
 
 deadlines_router = AirflowRouter(prefix="/dags/{dag_id}", tags=["Deadlines"])
 
@@ -106,7 +111,7 @@ def get_deadlines(
         .options(
             contains_eager(Deadline.dagrun).options(noload(DagRun.deadlines)),
             contains_eager(Deadline.deadline_alert),
-            noload(Deadline.callback),
+            joinedload(Deadline.callback),
         )
     )
 
@@ -209,3 +214,83 @@ def get_dag_deadline_alerts(
     alerts = session.scalars(alerts_select)
 
     return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+@deadlines_router.get(
+    "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+    responses=create_openapi_http_exception_doc(
+        [
+            status.HTTP_404_NOT_FOUND,
+        ]
+    ),
+    dependencies=[
+        Depends(
+            requires_access_dag(
+                method="GET",
+                access_entity=DagAccessEntity.TASK_LOGS,
+            )
+        ),
+    ],
+    response_model=TaskInstancesLogResponse,
+    response_model_exclude_unset=True,
+)
+def get_callback_logs(
+    dag_id: str,
+    dag_run_id: str,
+    callback_id: UUID,
+    session: SessionDep,
+) -> TaskInstancesLogResponse:
+    """
+    Get execution logs for a callback associated with a deadline.
+
+    Returns the logs produced during callback execution. These logs are uploaded
+    to remote storage (or written locally) by the callback supervisor after execution.
+    """
+    # Sanitize path components to prevent path traversal via URL parameters.
+    for param_name, param_value in [("dag_id", dag_id), ("dag_run_id", dag_run_id)]:
+        if os.sep in param_value or "\\" in param_value or ".." in param_value:
+            raise HTTPException(
+                status.HTTP_400_BAD_REQUEST,
+                f"Invalid characters in {param_name}",
+            )
+
+    # Verify the callback exists
+    callback = session.scalar(select(Callback).where(Callback.id == callback_id))
+    if callback is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            f"Callback with id `{callback_id}` was not found",
+        )
+
+    # Verify the dag_run exists with matching dag_id
+    dag_run = session.scalar(select(DagRun).where(DagRun.dag_id == dag_id, DagRun.run_id == dag_run_id))
+    if dag_run is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            f"DagRun with dag_id: `{dag_id}` and run_id: `{dag_run_id}` was not found",
+        )
+
+    # Verify the callback actually belongs to this dag_run (via the Deadline relationship)
+    deadline = session.scalar(
+        select(Deadline).where(Deadline.callback_id == callback_id, Deadline.dagrun_id == dag_run.id)
+    )
+    if deadline is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            f"Callback `{callback_id}` is not associated with DagRun `{dag_run_id}` of Dag `{dag_id}`",
+        )
+
+    # read_callback_log is a generator: path validation (which raises ValueError) runs lazily during
+    # iteration, not when the generator is created -- so list() must be inside the try for the 400 to fire.
+    try:
+        content = list(
+            read_callback_log(
+                dag_id=dag_id,
+                run_id=dag_run_id,
+                callback_id=str(callback_id),
+            )
+        )
+    except ValueError:
+        raise HTTPException(status.HTTP_400_BAD_REQUEST, "Invalid callback log path")
+
+    return TaskInstancesLogResponse.model_construct(content=content, continuation_token=None)

airflow-core/src/airflow/jobs/triggerer_job_runner.py

@@ -393,7 +393,7 @@ def from_api_response(cls, response: HITLDetailResponse) -> HITLDetailResponseRe
 class TriggerLoggingFactory:
     log_path: str
 
-    ti: RuntimeTI = attrs.field(repr=False)
+    ti: RuntimeTI | None = attrs.field(default=None, repr=False)
 
     bound_logger: WrappedLogger = attrs.field(init=False, repr=False)
 
@@ -430,8 +430,35 @@ def upload_to_remote(self):
             # Never actually called, nothing to do
             return
 
+        if self.ti is None:
+            # Callback triggers have no task instance — upload using the path directly.
+            self._upload_callback_log_to_remote()
+            return
+
         upload_to_remote(self.bound_logger, self.ti)
 
+    def _upload_callback_log_to_remote(self):
+        """Upload callback trigger logs to remote storage without a task instance."""
+        from airflow.sdk.log import load_remote_log_handler, relative_path_from_logger
+
+        handler = load_remote_log_handler()
+        if not handler:
+            return
+
+        raw_logger = getattr(self.bound_logger, "_logger")
+        try:
+            relative_path = relative_path_from_logger(raw_logger)
+        except Exception:
+            return
+        if not relative_path:
+            return
+
+        log_relative_path = relative_path.as_posix()
+        try:
+            handler.upload(log_relative_path, None)  # type: ignore[arg-type]
+        except Exception:
+            log.warning("Failed to upload callback trigger logs to remote", log_path=log_relative_path)
+
 
 def in_process_api_server() -> InProcessExecutionAPI:
     from airflow.api_fastapi.execution_api.app import InProcessExecutionAPI
@@ -789,6 +816,21 @@ def _create_workload(
             # protocol — asset-only triggers (and spec'd test mocks) may not expose it, so read
             # it defensively. Only callback triggers fetch DagRun context.
             callback = getattr(trigger, "callback", None)
+
+            # Set up dedicated logging for callback triggers first so their output is captured
+            # to a file the UI log endpoint can read — regardless of whether context fetch below
+            # succeeds (we still want logs for a callback that ends up skipped/failed).
+            if callback:
+                callback_data = callback.data or {}
+                dag_id = callback_data.get("dag_id", "unknown")
+                run_id = callback_data.get("run_id", "unknown")
+                callback_id = str(callback.id)
+                log_path = f"triggerer_callbacks/{dag_id}/{run_id}/{callback_id}"
+                self.logger_cache[trigger.id] = TriggerLoggingFactory(
+                    log_path=log_path,
+                    ti=None,
+                )
+
             # Only fetch DagRun data for callback triggers (not all non-task triggers).
             dag_run_data = self._fetch_callback_dag_run_data(trigger, session=session) if callback else None
             if callback and dag_run_data is None:

airflow-core/src/airflow/ui/openapi-gen/queries/common.ts

@@ -948,6 +948,14 @@ export const UseDeadlinesServiceGetDagDeadlineAlertsKeyFn = ({ dagId, limit, off
   offset?: number;
   orderBy?: string[];
 }, queryKey?: Array<unknown>) => [useDeadlinesServiceGetDagDeadlineAlertsKey, ...(queryKey ?? [{ dagId, limit, offset, orderBy }])];
+export type DeadlinesServiceGetCallbackLogsDefaultResponse = Awaited<ReturnType<typeof DeadlinesService.getCallbackLogs>>;
+export type DeadlinesServiceGetCallbackLogsQueryResult<TData = DeadlinesServiceGetCallbackLogsDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
+export const useDeadlinesServiceGetCallbackLogsKey = "DeadlinesServiceGetCallbackLogs";
+export const UseDeadlinesServiceGetCallbackLogsKeyFn = ({ callbackId, dagId, dagRunId }: {
+  callbackId: string;
+  dagId: string;
+  dagRunId: string;
+}, queryKey?: Array<unknown>) => [useDeadlinesServiceGetCallbackLogsKey, ...(queryKey ?? [{ callbackId, dagId, dagRunId }])];
 export type StructureServiceStructureDataDefaultResponse = Awaited<ReturnType<typeof StructureService.structureData>>;
 export type StructureServiceStructureDataQueryResult<TData = StructureServiceStructureDataDefaultResponse, TError = unknown> = UseQueryResult<TData, TError>;
 export const useStructureServiceStructureDataKey = "StructureServiceStructureData";