Add UI endpoint for deadline callback log visibility

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

@@ -24,6 +24,7 @@
 from pydantic import AliasPath, Field
 
 from airflow.api_fastapi.core_api.base import BaseModel
+from airflow.utils.state import CallbackState
 
 
 class DeadlineResponse(BaseModel):
@@ -37,6 +38,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

@@ -989,6 +989,85 @@ 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: callback_id
+        in: path
+        required: true
+        schema:
+          type: string
+          format: uuid
+          title: Callback Id
+      - 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: accept
+        in: header
+        required: false
+        schema:
+          type: string
+          enum:
+          - application/json
+          - application/x-ndjson
+          - '*/*'
+          default: '*/*'
+          title: Accept
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/TaskInstancesLogResponse'
+            application/x-ndjson:
+              schema:
+                type: string
+                example: '{"content": "content"}
+
+                  {"content": "content"}
+
+                  '
+        '400':
+          content:
+            application/json:
+              schema:
+                $ref: '#/components/schemas/HTTPExceptionResponse'
+          description: Bad Request
+        '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:
@@ -2236,6 +2315,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:
@@ -2924,6 +3014,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
@@ -3942,6 +4042,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:
@@ -4215,6 +4330,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

@@ -18,13 +18,16 @@
 from __future__ import annotations
 
 from typing import Annotated
+from uuid import UUID
 
 from fastapi import Depends, HTTPException, status
+from fastapi.responses import StreamingResponse
 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
+from airflow.api_fastapi.common.headers import HeaderAcceptJsonOrNdjson
 from airflow.api_fastapi.common.parameters import (
     FilterParam,
     QueryLimit,
@@ -35,16 +38,23 @@
     filter_param_factory,
 )
 from airflow.api_fastapi.common.router import AirflowRouter
+from airflow.api_fastapi.common.types import Mimetype
+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.routes.public.log import (
+    _buffered_ndjson_stream,
+    ndjson_example_response_for_get_log,
+)
 from airflow.api_fastapi.core_api.security import ReadableDagRunsFilterDep, requires_access_dag
 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, validate_log_path_component
 
 deadlines_router = AirflowRouter(prefix="/dags/{dag_id}", tags=["Deadlines"])
 
@@ -106,7 +116,7 @@ def get_deadlines(
         .options(
             contains_eager(Deadline.dagrun).options(noload(DagRun.deadlines)),
             contains_eager(Deadline.deadline_alert),
-            noload(Deadline.callback),
+            joinedload(Deadline.callback),
         )
     )
 
@@ -201,3 +211,75 @@ def get_dag_deadline_alerts(
     alerts = session.scalars(alerts_select)
 
     return DeadlineAlertCollectionResponse(deadline_alerts=alerts, total_entries=total_entries)
+
+
+def _validated_log_path_params(dag_id: str, dag_run_id: str) -> tuple[str, str]:
+    """Reject dag_id/dag_run_id values that are unsafe as log path components (path traversal)."""
+    for param_name, param_value in (("dag_id", dag_id), ("dag_run_id", dag_run_id)):
+        try:
+            validate_log_path_component(param_value)
+        except ValueError:
+            raise HTTPException(status.HTTP_400_BAD_REQUEST, f"Invalid characters in {param_name}")
+    return dag_id, dag_run_id
+
+
+@deadlines_router.get(
+    "/dagRuns/{dag_run_id}/callbacks/{callback_id}/logs",
+    responses={
+        **create_openapi_http_exception_doc([status.HTTP_400_BAD_REQUEST, status.HTTP_404_NOT_FOUND]),
+        status.HTTP_200_OK: {
+            "description": "Successful Response",
+            "content": ndjson_example_response_for_get_log,
+        },
+    },
+    dependencies=[
+        Depends(
+            requires_access_dag(
+                method="GET",
+                access_entity=DagAccessEntity.TASK_LOGS,
+            )
+        ),
+    ],
+    response_model=TaskInstancesLogResponse,
+    response_model_exclude_unset=True,
+)
+def get_callback_logs(
+    path_params: Annotated[tuple[str, str], Depends(_validated_log_path_params)],
+    callback_id: UUID,
+    accept: HeaderAcceptJsonOrNdjson,
+    session: SessionDep,
+):
+    """
+    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.
+    """
+    dag_id, dag_run_id = path_params
+
+    # A single exists-only check that the callback belongs to this dag run via its Deadline.
+    deadline_exists = session.scalar(
+        select(Deadline.id)
+        .join(Deadline.dagrun)
+        .where(
+            Deadline.callback_id == callback_id,
+            DagRun.dag_id == dag_id,
+            DagRun.run_id == dag_run_id,
+        )
+        .limit(1)
+    )
+    if deadline_exists is None:
+        raise HTTPException(
+            status.HTTP_404_NOT_FOUND,
+            f"Callback `{callback_id}` with a deadline for DagRun `{dag_run_id}` of Dag `{dag_id}` was not found",
+        )
+
+    log_stream = read_callback_log(dag_id=dag_id, run_id=dag_run_id, callback_id=str(callback_id))
+
+    if accept == Mimetype.NDJSON:
+        return StreamingResponse(
+            media_type="application/x-ndjson",
+            content=_buffered_ndjson_stream(f"{log.model_dump_json()}\n" for log in log_stream),
+        )
+
+    return TaskInstancesLogResponse.model_construct(content=list(log_stream), continuation_token=None)

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

@@ -421,7 +421,8 @@ def from_api_response(cls, response: HITLDetailResponse) -> HITLDetailResponseRe
 class TriggerLoggingFactory:
     log_path: str
 
-    ti: RuntimeTI = attrs.field(repr=False)
+    # Callback triggers have no task instance; ``upload_to_remote`` accepts ``ti=None``.
+    ti: RuntimeTI | None = attrs.field(default=None, repr=False)
 
     bound_logger: WrappedLogger = attrs.field(init=False, repr=False)
 
@@ -846,6 +847,19 @@ def _create_workload(
             if trigger.assets:
                 watched_assets = {a.name: a.uri for a in trigger.assets}
 
+            if callback := getattr(trigger, "callback", None):
+                # Callback triggers get dedicated logging so their output is captured to a
+                # file the UI callback log endpoint can read. dag_id is stored on the callback
+                # data; run_id comes from the deadline context injected at miss time.
+                callback_data = callback.data or {}
+                context = (callback_data.get("kwargs") or {}).get("context") or {}
+                dag_run_data = context.get("dag_run") or {}
+                dag_id = callback_data.get("dag_id") or dag_run_data.get("dag_id") or "unknown"
+                run_id = dag_run_data.get("dag_run_id") or "unknown"
+                self.logger_cache[trigger.id] = TriggerLoggingFactory(
+                    log_path=f"triggerer_callbacks/{dag_id}/{run_id}/{callback.id}",
+                )
+
             return workloads.RunTrigger(
                 id=trigger.id,
                 classpath=trigger.classpath,