diff --git a/.nextchanges/air/multitask-air-tasks.md b/.nextchanges/air/multitask-air-tasks.md
new file mode 100644
index 00000000000..a8cb3da5d2d
--- /dev/null
+++ b/.nextchanges/air/multitask-air-tasks.md
@@ -0,0 +1 @@
+* `air list` and `air get` now read the AI Runtime task wherever it sits in a multi-task job (for example after a notebook prep task) instead of the first task; a job with several AI Runtime tasks lists one row per task, `air get` gains `--task-key`, and both commands report `task_key` in JSON output. ([#7632](https://github.com/databricks/cli/pull/7632))
diff --git a/acceptance/air/control-chars/output.txt b/acceptance/air/control-chars/output.txt
index b320060ce29..a08f16e8be7 100644
--- a/acceptance/air/control-chars/output.txt
+++ b/acceptance/air/control-chars/output.txt
@@ -38,6 +38,7 @@ MLflow URL: [DATABRICKS_URL]/ml/experiments/exp1/runs/run1
     "runs": [
       {
         "run_id": "123",
+        "task_key": "train",
         "run_name": "qwen\u001b[2J-train",
         "user": "user\u001b[31m@example.com",
         "status": "SUCCESS",
diff --git a/acceptance/air/get-ai-runtime/output.txt b/acceptance/air/get-ai-runtime/output.txt
index c84683e3f3e..3e63c75c7b1 100644
--- a/acceptance/air/get-ai-runtime/output.txt
+++ b/acceptance/air/get-ai-runtime/output.txt
@@ -43,6 +43,7 @@ MLflow URL: [DATABRICKS_URL]/ml/experiments/exp1/runs/run1
   "ts": "[TIMESTAMP]",
   "data": {
     "run_id": "123",
+    "task_key": "train",
     "status": "SUCCESS",
     "started_at": "[TIMESTAMP]",
     "duration_seconds": 12,
diff --git a/acceptance/air/get-multitask/out.test.toml b/acceptance/air/get-multitask/out.test.toml
new file mode 100644
index 00000000000..18b1a88417e
--- /dev/null
+++ b/acceptance/air/get-multitask/out.test.toml
@@ -0,0 +1 @@
+Cloud = false
diff --git a/acceptance/air/get-multitask/output.txt b/acceptance/air/get-multitask/output.txt
new file mode 100644
index 00000000000..ed79dd67e24
--- /dev/null
+++ b/acceptance/air/get-multitask/output.txt
@@ -0,0 +1,113 @@
+
+=== get (text): the first AI Runtime task, not the notebook prep task
+
+>>> [CLI] air get 123
+
+╭─ Metadata ─────────────────────────────────────────────────────╮
+│                                                                │
+│  Run ID        123                                             │
+│  Task          train                                           │
+│  Status        ● SUCCESS                                       │
+│  Submitted     2023-11-14 22:13 UTC                            │
+│  Retries       0                                               │
+│  Max Retries   0                                               │
+│  Duration      10s                                             │
+│  Experiment    air-byot-multitask-dag                          │
+│  MLflow Run    matmul-smoke                                    │
+│  User          user@example.com                                │
+│  Accelerators  1x A10                                          │
+│  Environment   5                                               │
+│                                                                │
+╰────────────────────────────────────────────────────────────────╯
+
+Run URL:    [DATABRICKS_URL]/jobs/runs/123?o=[WORKSPACE_ID]
+MLflow URL: [DATABRICKS_URL]/ml/experiments/exp1/runs/run1
+
+=== get --task-key eval (text)
+
+>>> [CLI] air get 123 --task-key eval
+
+╭─ Metadata ─────────────────────────────────────────────────────╮
+│                                                                │
+│  Run ID        123                                             │
+│  Task          eval                                            │
+│  Status        ● SUCCESS                                       │
+│  Submitted     2023-11-14 22:13 UTC                            │
+│  Retries       0                                               │
+│  Max Retries   2                                               │
+│  Duration      8s                                              │
+│  Experiment    eval-exp                                        │
+│  MLflow Run    matmul-smoke                                    │
+│  User          user@example.com                                │
+│  Accelerators  8x H100                                         │
+│  Environment   5                                               │
+│                                                                │
+╰────────────────────────────────────────────────────────────────╯
+
+Run URL:    [DATABRICKS_URL]/jobs/runs/123?o=[WORKSPACE_ID]
+MLflow URL: [DATABRICKS_URL]/ml/experiments/exp1/runs/run1
+
+=== get (json)
+
+>>> [CLI] air get 123 -o json
+{
+  "v": 1,
+  "ts": "[TIMESTAMP]",
+  "data": {
+    "run_id": "123",
+    "task_key": "train",
+    "status": "SUCCESS",
+    "started_at": "[TIMESTAMP]",
+    "duration_seconds": 10,
+    "attempt_number": 0,
+    "experiment_name": "air-byot-multitask-dag",
+    "dashboard_url": "[DATABRICKS_URL]/jobs/runs/123?o=[WORKSPACE_ID]",
+    "mlflow_url": "[DATABRICKS_URL]/ml/experiments/exp1/runs/run1/logs",
+    "est_remaining_seconds": null,
+    "est_percent_complete": null
+  }
+}
+
+=== get --task-key eval (json)
+
+>>> [CLI] air get 123 --task-key eval -o json
+{
+  "v": 1,
+  "ts": "[TIMESTAMP]",
+  "data": {
+    "run_id": "123",
+    "task_key": "eval",
+    "status": "SUCCESS",
+    "started_at": "[TIMESTAMP]",
+    "duration_seconds": 8,
+    "attempt_number": 0,
+    "experiment_name": "eval-exp",
+    "dashboard_url": "[DATABRICKS_URL]/jobs/runs/123?o=[WORKSPACE_ID]",
+    "mlflow_url": "[DATABRICKS_URL]/ml/experiments/exp1/runs/run1/logs",
+    "est_remaining_seconds": null,
+    "est_percent_complete": null
+  }
+}
+
+=== get --task-key prep: not an AI Runtime task
+
+>>> [CLI] air get 123 --task-key prep
+Error: run 123 has no AI Runtime task "prep"; its AI Runtime tasks are: train, eval
+
+Exit code: 1
+
+=== get --task-key prep (json)
+
+>>> [CLI] air get 123 --task-key prep -o json
+{
+  "v": 1,
+  "ts": "[TIMESTAMP]",
+  "error": {
+    "code": "INVALID_ARGS",
+    "kind": "PERMANENT",
+    "message": "run 123 has no AI Runtime task \"prep\"; its AI Runtime tasks are: train, eval",
+    "retryable": false
+  }
+}
+
+Exit code: 1
diff --git a/acceptance/air/get-multitask/script b/acceptance/air/get-multitask/script
new file mode 100644
index 00000000000..d4d4505995f
--- /dev/null
+++ b/acceptance/air/get-multitask/script
@@ -0,0 +1,17 @@
+title "get (text): the first AI Runtime task, not the notebook prep task"
+trace $CLI air get 123
+
+title "get --task-key eval (text)"
+trace $CLI air get 123 --task-key eval
+
+title "get (json)"
+trace $CLI air get 123 -o json
+
+title "get --task-key eval (json)"
+trace $CLI air get 123 --task-key eval -o json
+
+title "get --task-key prep: not an AI Runtime task"
+errcode trace $CLI air get 123 --task-key prep
+
+title "get --task-key prep (json)"
+errcode trace $CLI air get 123 --task-key prep -o json
diff --git a/acceptance/air/get-multitask/test.toml b/acceptance/air/get-multitask/test.toml
new file mode 100644
index 00000000000..751f09abb17
--- /dev/null
+++ b/acceptance/air/get-multitask/test.toml
@@ -0,0 +1,84 @@
+# The SDK occasionally probes host reachability with a HEAD request; stub it so
+# the test is deterministic.
+
+[[Server]]
+Pattern = "HEAD /"
+Response.Body = ''
+
+# A multi-task job in the shape of the official DAB example (universe
+# ai-compute/dabs/examples/03_multitask_dag): a notebook prep task first, then the
+# ai_runtime_task that depends on it, plus a second ai_runtime_task. `air get`
+# must read the AI Runtime task wherever it sits, not tasks[0].
+[[Server]]
+Pattern = "GET /api/2.2/jobs/runs/get"
+Response.Body = '''
+{
+  "run_id": 123,
+  "run_page_url": "https://my-workspace.cloud.databricks.test/jobs/runs/123",
+  "creator_user_name": "user@example.com",
+  "start_time": 1700000000000,
+  "end_time": 1700000012000,
+  "state": {"life_cycle_state": "TERMINATED", "result_state": "SUCCESS"},
+  "environments": [
+    {"environment_key": "default", "spec": {"environment_version": "5"}}
+  ],
+  "tasks": [
+    {
+      "task_key": "prep",
+      "run_id": 455,
+      "attempt_number": 0,
+      "start_time": 1700000000000,
+      "end_time": 1700000002000,
+      "state": {"life_cycle_state": "TERMINATED", "result_state": "SUCCESS"},
+      "notebook_task": {"notebook_path": "/Workspace/Users/user@example.com/prep"}
+    },
+    {
+      "task_key": "train",
+      "run_id": 456,
+      "attempt_number": 0,
+      "max_retries": 0,
+      "start_time": 1700000002000,
+      "end_time": 1700000012000,
+      "state": {"life_cycle_state": "TERMINATED", "result_state": "SUCCESS"},
+      "depends_on": [{"task_key": "prep"}],
+      "environment_key": "default",
+      "ai_runtime_task": {
+        "experiment": "air-byot-multitask-dag",
+        "deployments": [
+          {"compute": {"accelerator_type": "GPU_1xA10", "accelerator_count": 1}}
+        ]
+      }
+    },
+    {
+      "task_key": "eval",
+      "run_id": 457,
+      "attempt_number": 0,
+      "max_retries": 2,
+      "start_time": 1700000012000,
+      "end_time": 1700000020000,
+      "state": {"life_cycle_state": "TERMINATED", "result_state": "SUCCESS"},
+      "depends_on": [{"task_key": "train"}],
+      "environment_key": "default",
+      "ai_runtime_task": {
+        "experiment": "/Users/user@example.com/eval-exp",
+        "deployments": [
+          {"compute": {"accelerator_type": "GPU_8xH100", "accelerator_count": 8}}
+        ]
+      }
+    }
+  ]
+}
+'''
+
+# MLflow identifiers for the deep link, fetched for the selected task's run.
+[[Server]]
+Pattern = "GET /api/2.2/jobs/runs/get-output"
+Response.Body = '''
+{"ai_runtime_task_output": {"mlflow_experiment_id": "exp1", "mlflow_run_id": "run1"}}
+'''
+
+[[Server]]
+Pattern = "GET /api/2.0/mlflow/runs/get"
+Response.Body = '''
+{"run": {"info": {"run_name": "matmul-smoke"}}}
+'''
diff --git a/acceptance/air/get-pending/output.txt b/acceptance/air/get-pending/output.txt
index 3a5f9a72eda..5cbd71caa09 100644
--- a/acceptance/air/get-pending/output.txt
+++ b/acceptance/air/get-pending/output.txt
@@ -5,6 +5,7 @@
   "ts": "[TIMESTAMP]",
   "data": {
     "run_id": "123",
+    "task_key": "train",
     "status": "PENDING",
     "started_at": null,
     "duration_seconds": null,
diff --git a/acceptance/air/get/output.txt b/acceptance/air/get/output.txt
index 9c1fed61721..990144607ae 100644
--- a/acceptance/air/get/output.txt
+++ b/acceptance/air/get/output.txt
@@ -43,6 +43,7 @@ MLflow URL: [DATABRICKS_URL]/ml/experiments/exp1/runs/run1
   "ts": "[TIMESTAMP]",
   "data": {
     "run_id": "123",
+    "task_key": "train",
     "status": "SUCCESS",
     "started_at": "[TIMESTAMP]",
     "duration_seconds": 12,
diff --git a/acceptance/air/help/output.txt b/acceptance/air/help/output.txt
index 7b37a4c83ac..337b74a177a 100644
--- a/acceptance/air/help/output.txt
+++ b/acceptance/air/help/output.txt
@@ -92,11 +92,17 @@ Show a run's status, configuration, and timing.
 
 JOB_RUN_ID is the job run ID returned by "databricks air run".
 
+A job may run other tasks around its AI Runtime task (for example a notebook
+that prepares data first); the view shows the first AI Runtime task in task
+order. When a job has several AI Runtime tasks, pass --task-key to show another
+one.
+
 Usage:
   databricks air get JOB_RUN_ID [flags]
 
 Flags:
-  -h, --help   help for get
+  -h, --help              help for get
+      --task-key string   Show this AI Runtime task of the run (default: the first in task order)
 
 Global Flags:
       --debug            enable debug logging
diff --git a/acceptance/air/list/output.txt b/acceptance/air/list/output.txt
index 4c0c7ed3542..25a330d57a1 100644
--- a/acceptance/air/list/output.txt
+++ b/acceptance/air/list/output.txt
@@ -2,8 +2,9 @@
 === list (text)
 
 >>> [CLI] air list
-   Run ID           Experiment  Status     Started              Duration  Progress  MLflow                  User                   Accelerators
-   334747067049496  qwen-train  ● SUCCESS  [TIMESTAMP]       12s         -  qwen-train-001          [USERNAME]  8x H100
+   Run ID           Experiment              Status     Started              Duration  Progress  MLflow                  User                   Accelerators
+   334747067049496  qwen-train              ● SUCCESS  [TIMESTAMP]       12s         -  qwen-train-001          [USERNAME]  8x H100
+   334747067049500  air-byot-multitask-dag  ● SUCCESS  [TIMESTAMP]       20s         -  qwen-train-001          [USERNAME]  1x A10
 
 === list (json)
 
@@ -15,6 +16,7 @@
     "runs": [
       {
         "run_id": "334747067049496",
+        "task_key": "",
         "run_name": "qwen-train",
         "user": "[USERNAME]",
         "status": "SUCCESS",
@@ -25,6 +27,20 @@
         "accelerator_type": "GPU_8xH100",
         "num_accelerators": 8,
         "dashboard_url": "[DATABRICKS_URL]/jobs/runs/334747067049496?o=[WORKSPACE_ID]"
+      },
+      {
+        "run_id": "334747067049500",
+        "task_key": "train",
+        "run_name": "air-byot-multitask-dag",
+        "user": "[USERNAME]",
+        "status": "SUCCESS",
+        "started_at": "[TIMESTAMP]",
+        "is_sweep": false,
+        "experiment_name": "air-byot-multitask-dag",
+        "duration_seconds": 20,
+        "accelerator_type": "GPU_1xA10",
+        "num_accelerators": 1,
+        "dashboard_url": "[DATABRICKS_URL]/jobs/runs/334747067049500?o=[WORKSPACE_ID]"
       }
     ]
   }
@@ -46,6 +62,7 @@
     "runs": [
       {
         "run_id": "334747067049496",
+        "task_key": "",
         "run_name": "qwen-train",
         "user": "[USERNAME]",
         "status": "SUCCESS",
diff --git a/acceptance/air/list/test.toml b/acceptance/air/list/test.toml
index 9d5ca13df25..9ae1417ce39 100644
--- a/acceptance/air/list/test.toml
+++ b/acceptance/air/list/test.toml
@@ -10,8 +10,10 @@ Response.Body = ''
 # `air list` reads Jobs runs/list directly (expand_tasks), filtering to AIR runs.
 # The current runs carry an ai_runtime_task (not modeled by the typed SDK), which
 # is why the CLI parses the response raw. Both runs belong to the default user
-# (tester@databricks.com, from the built-in scim/v2/Me handler). The second run
-# has no AI task and must be filtered out.
+# (tester@databricks.com, from the built-in scim/v2/Me handler). The second run is
+# the official multi-task DAB shape, a notebook prep task before the
+# ai_runtime_task, and must be listed from its AIR task. The third run has no AI
+# task and must be filtered out.
 [[Server]]
 Pattern = "GET /api/2.2/jobs/runs/list"
 Response.Body = '''
@@ -32,6 +34,34 @@ Response.Body = '''
         }
       }]
     },
+    {
+      "run_id": 334747067049500,
+      "run_name": "air-byot-multitask-dag",
+      "creator_user_name": "tester@databricks.com",
+      "start_time": 1717608700000,
+      "end_time": 1717608730000,
+      "state": {"life_cycle_state": "TERMINATED", "result_state": "SUCCESS"},
+      "tasks": [
+        {
+          "task_key": "prep",
+          "run_id": 334747067049501,
+          "start_time": 1717608700000,
+          "end_time": 1717608705000,
+          "notebook_task": {"notebook_path": "/Workspace/Users/tester@databricks.com/prep"}
+        },
+        {
+          "task_key": "train",
+          "run_id": 334747067049502,
+          "start_time": 1717608710000,
+          "end_time": 1717608730000,
+          "depends_on": [{"task_key": "prep"}],
+          "ai_runtime_task": {
+            "experiment": "air-byot-multitask-dag",
+            "deployments": [{"compute": {"accelerator_type": "GPU_1xA10", "accelerator_count": 1}}]
+          }
+        }
+      ]
+    },
     {
       "run_id": 999000999000,
       "run_name": "not-an-air-run",
diff --git a/cmd/air/cancel.go b/cmd/air/cancel.go
index 0ca93c793cd..36ad20fe309 100644
--- a/cmd/air/cancel.go
+++ b/cmd/air/cancel.go
@@ -106,6 +106,9 @@ all active runs submitted by the current user.
 					fmt.Errorf("failed to list active runs: %w", err))
 			}
 
+			// A job with several AI runtime tasks lists one row per task; the
+			// run is cancelled (and previewed) once.
+			rows = firstRowPerRun(rows)
 			runIDs = make([]string, 0, len(rows))
 			for i := range rows {
 				if rows[i].RunID != "" {
@@ -215,6 +218,20 @@ func cancelRun(ctx context.Context, w *databricks.WorkspaceClient, rid string, s
 	return err
 }
 
+// firstRowPerRun keeps the first row of each run id, in order.
+func firstRowPerRun(rows []listRow) []listRow {
+	seen := make(map[string]bool, len(rows))
+	unique := make([]listRow, 0, len(rows))
+	for _, row := range rows {
+		if seen[row.RunID] {
+			continue
+		}
+		seen[row.RunID] = true
+		unique = append(unique, row)
+	}
+	return unique
+}
+
 // displayCancelPreview shows the runs that `cancel --all` is about to terminate.
 func displayCancelPreview(ctx context.Context, rows []listRow, host string) {
 	var sb strings.Builder
diff --git a/cmd/air/format.go b/cmd/air/format.go
index 68c956ad4c0..dcae15d6c62 100644
--- a/cmd/air/format.go
+++ b/cmd/air/format.go
@@ -119,10 +119,11 @@ func runWaitingForCompute(run *jobs.Run) bool {
 	if run.State.LifeCycleState == jobs.RunLifeCycleStatePending {
 		return true
 	}
-	if run.State.LifeCycleState != jobs.RunLifeCycleStateRunning || len(run.Tasks) == 0 || run.Tasks[0].State == nil {
+	task := leadTask(run)
+	if run.State.LifeCycleState != jobs.RunLifeCycleStateRunning || task == nil || task.State == nil {
 		return false
 	}
-	switch run.Tasks[0].State.LifeCycleState {
+	switch task.State.LifeCycleState {
 	case jobs.RunLifeCycleStatePending, jobs.RunLifeCycleStateQueued,
 		jobs.RunLifeCycleStateWaitingForRetry, jobs.RunLifeCycleStateBlocked:
 		return true
@@ -156,8 +157,8 @@ func terminationReason(run *jobs.Run) *string {
 			return &message
 		}
 	}
-	if len(run.Tasks) > 0 && run.Tasks[0].State != nil {
-		if message := strings.TrimSpace(run.Tasks[0].State.StateMessage); message != "" {
+	if task := leadTask(run); task != nil && task.State != nil {
+		if message := strings.TrimSpace(task.State.StateMessage); message != "" {
 			return &message
 		}
 	}
@@ -299,10 +300,11 @@ func latestAttemptNumber(run *jobs.Run) int {
 // "/Users/me@example.com/my-experiment"); we strip that prefix so users see just
 // the experiment name they chose.
 func experimentName(run *jobs.Run) *string {
-	if len(run.Tasks) == 0 {
+	t := firstTask(run)
+	if t == nil {
 		return nil
 	}
-	task := run.Tasks[0].GenAiComputeTask
+	task := t.GenAiComputeTask
 	if task == nil || task.MlflowExperimentName == "" {
 		return nil
 	}
@@ -328,10 +330,11 @@ func stripExperimentUserPrefix(name string) string {
 // accelerators returns a short description of the GPUs the run uses, such as
 // "8x H100", or an empty string if the run has no GPU compute attached.
 func accelerators(run *jobs.Run) string {
-	if len(run.Tasks) == 0 {
+	t := firstTask(run)
+	if t == nil {
 		return ""
 	}
-	task := run.Tasks[0].GenAiComputeTask
+	task := t.GenAiComputeTask
 	if task == nil || task.Compute == nil {
 		return ""
 	}
diff --git a/cmd/air/format_test.go b/cmd/air/format_test.go
index c76260d28cb..240bdc91b84 100644
--- a/cmd/air/format_test.go
+++ b/cmd/air/format_test.go
@@ -306,3 +306,18 @@ func TestTerminationReason(t *testing.T) {
 	assert.Nil(t, terminationReason(&jobs.Run{State: &jobs.RunState{ResultState: jobs.RunResultStateSuccess, StateMessage: "done"}}))
 	assert.Nil(t, terminationReason(&jobs.Run{State: &jobs.RunState{ResultState: jobs.RunResultStateCanceled, StateMessage: "cancelled"}}))
 }
+
+func TestLegacyExperimentAndAcceleratorsSkipNonAirTasks(t *testing.T) {
+	run := &jobs.Run{RunId: 5, Tasks: []jobs.RunTask{
+		notebookTask("prep", 51),
+		{TaskKey: "train", RunId: 52, GenAiComputeTask: &jobs.GenAiComputeTask{
+			TrainingScriptPath:   "/train.py",
+			MlflowExperimentName: "/Users/me@example.com/legacy-exp",
+			Compute:              &jobs.ComputeConfig{GpuType: "H100", NumGpus: 8},
+		}},
+	}}
+	name := experimentName(run)
+	require.NotNil(t, name)
+	assert.Equal(t, "legacy-exp", *name)
+	assert.Equal(t, "8x H100", accelerators(run))
+}
diff --git a/cmd/air/get.go b/cmd/air/get.go
index bac778ded51..a5980fcb836 100644
--- a/cmd/air/get.go
+++ b/cmd/air/get.go
@@ -6,6 +6,7 @@ import (
 	"fmt"
 	"path"
 	"strconv"
+	"strings"
 
 	"github.com/databricks/cli/cmd/root"
 	"github.com/databricks/cli/libs/auth"
@@ -23,7 +24,10 @@ import (
 // the machine-readable output; fields tagged `json:"-"` are shown only in the
 // human-readable text view.
 type getData struct {
-	RunID             string  `json:"run_id"`
+	RunID string `json:"run_id"`
+	// TaskKey is the AI runtime task shown (see selectAirTask), or null for a
+	// run without one.
+	TaskKey           *string `json:"task_key"`
 	Status            string  `json:"status"`
 	StartedAt         *string `json:"started_at"`
 	DurationSeconds   *int64  `json:"duration_seconds"`
@@ -46,6 +50,9 @@ type getData struct {
 	// missing. The styled single-run renderer (render.go) consumes them; the
 	// Run ID, Status, and MLflow Run cells it draws are styled and hyperlinked
 	// there rather than stored here.
+	// TaskDisplay is the "Task" cell, set only when the run has several AI
+	// runtime tasks, so the view says which one it shows.
+	TaskDisplay         string `json:"-"`
 	SubmittedDisplay    string `json:"-"`
 	DurationDisplay     string `json:"-"`
 	ExperimentDisplay   string `json:"-"`
@@ -119,18 +126,26 @@ func credentialError(ctx context.Context, w *databricks.WorkspaceClient, err err
 // newGetCommand returns the `air get JOB_RUN_ID` command, which shows status,
 // configuration, and timing details for a specific run.
 func newGetCommand() *cobra.Command {
+	var taskKey string
 	cmd := &cobra.Command{
 		Use:   "get JOB_RUN_ID",
 		Args:  root.ExactArgs(1),
 		Short: "Show a run's status, configuration, and timing",
 		Long: `Show a run's status, configuration, and timing.
 
-JOB_RUN_ID is the job run ID returned by "databricks air run".`,
+JOB_RUN_ID is the job run ID returned by "databricks air run".
+
+A job may run other tasks around its AI Runtime task (for example a notebook
+that prepares data first); the view shows the first AI Runtime task in task
+order. When a job has several AI Runtime tasks, pass --task-key to show another
+one.`,
 		Annotations: map[string]string{
 			"template": getTemplate,
 		},
 	}
 
+	cmd.Flags().StringVar(&taskKey, "task-key", "", "Show this AI Runtime task of the run (default: the first in task order)")
+
 	// Resolve and authenticate the workspace client up front so an auth failure
 	// fails fast here, before any run status or config is fetched or printed.
 	// ErrAlreadyPrinted passes through (it was handled upstream); other failures
@@ -167,6 +182,14 @@ JOB_RUN_ID is the job run ID returned by "databricks air run".`,
 				fmt.Errorf("failed to get status for run %d: %w", runID, err))
 		}
 
+		// From here on, run holds only the selected AI runtime task's task runs,
+		// so the status, timing, MLflow and config lookups all read that task.
+		fullRun := run
+		run, err = selectAirTask(run, taskKey)
+		if err != nil {
+			return renderError(ctx, cmd, "INVALID_ARGS", "PERMANENT", false, err)
+		}
+
 		// Prefer the configured workspace ID; only look it up when it is unset.
 		id, err := auth.ResolveWorkspaceID(ctx, w)
 		if err != nil {
@@ -180,6 +203,7 @@ JOB_RUN_ID is the job run ID returned by "databricks air run".`,
 		}
 
 		data := buildGetData(run)
+		annotateTask(fullRun, run, &data)
 		data.DashboardURL = dashboardURL(w.Config.Host, runID, workspaceID)
 		taskOutput, taskOutputErr := latestTaskOutput(ctx, w, run.Tasks)
 		ids := mlflowIDsFromRetries(ctx, w, run.Tasks, isTerminal(run), taskOutput, taskOutputErr)
@@ -231,6 +255,43 @@ JOB_RUN_ID is the job run ID returned by "databricks air run".`,
 	return cmd
 }
 
+// selectAirTask narrows run (see taskRun) to the AI runtime task named by key,
+// or to its first AI runtime task in task order when key is empty, so the rest
+// of the command reads that task alone. A run with no AI runtime task is
+// returned unchanged when no key is given. The error names the run's AI runtime
+// tasks when key is not one of them.
+func selectAirTask(run *jobs.Run, key string) (*jobs.Run, error) {
+	keys := airTaskKeys(run)
+	if key == "" {
+		if len(keys) == 0 {
+			return run, nil
+		}
+		key = keys[0]
+	}
+	if narrowed := taskRun(run, key); narrowed != nil {
+		return narrowed, nil
+	}
+	if len(keys) == 0 {
+		return nil, fmt.Errorf("run %d has no AI Runtime task %q; it has no AI Runtime tasks", run.RunId, key)
+	}
+	return nil, fmt.Errorf("run %d has no AI Runtime task %q; its AI Runtime tasks are: %s", run.RunId, key, strings.Join(keys, ", "))
+}
+
+// annotateTask records which AI runtime task of full the view shows: selected
+// is full narrowed to it (see selectAirTask). The text cell is set only when
+// full has several AI runtime tasks.
+func annotateTask(full, selected *jobs.Run, data *getData) {
+	t := leadTask(selected)
+	if t == nil {
+		return
+	}
+	key := t.TaskKey
+	data.TaskKey = &key
+	if len(airTaskKeys(full)) > 1 {
+		data.TaskDisplay = key
+	}
+}
+
 // enrichFromAiRuntimeTask fills the config path, experiment, and accelerators
 // from the run's ai_runtime_task. Best-effort: empty fields leave the existing
 // "N/A" fallbacks in place.
@@ -257,10 +318,11 @@ func enrichFromAiRuntimeTask(run *jobs.Run, data *getData) {
 
 // aiRuntimeTaskOf returns the run's first ai_runtime_task, or nil.
 func aiRuntimeTaskOf(run *jobs.Run) *jobs.AiRuntimeTask {
-	if len(run.Tasks) == 0 {
+	t := firstTask(run)
+	if t == nil {
 		return nil
 	}
-	return run.Tasks[0].AiRuntimeTask
+	return t.AiRuntimeTask
 }
 
 // buildGetData extracts the fields we display from a run. The text-view cells
diff --git a/cmd/air/get_test.go b/cmd/air/get_test.go
index c6e984c3b44..354fe464293 100644
--- a/cmd/air/get_test.go
+++ b/cmd/air/get_test.go
@@ -322,7 +322,7 @@ func TestBuildGetDataTerminationReasonJSON(t *testing.T) {
 	body, err := json.Marshal(failed)
 	require.NoError(t, err)
 	assert.JSONEq(t, `{
-		"run_id":"5","status":"FAILED","started_at":null,"duration_seconds":null,
+		"run_id":"5","task_key":null,"status":"FAILED","started_at":null,"duration_seconds":null,
 		"attempt_number":0,"experiment_name":null,"dashboard_url":"","mlflow_url":null,
 		"termination_reason":"out of memory","est_remaining_seconds":null,"est_percent_complete":null
 	}`, string(body))
@@ -344,7 +344,7 @@ func TestBuildGetDataWaitingForComputeStatus(t *testing.T) {
 	require.NoError(t, err)
 
 	assert.JSONEq(t, `{
-		"run_id":"8","status":"PENDING","started_at":null,"duration_seconds":null,
+		"run_id":"8","task_key":null,"status":"PENDING","started_at":null,"duration_seconds":null,
 		"attempt_number":0,"experiment_name":null,"dashboard_url":"","mlflow_url":null,
 		"est_remaining_seconds":null,"est_percent_complete":null
 	}`, string(body))
@@ -400,3 +400,94 @@ func TestBuildGetDataEmpty(t *testing.T) {
 	assert.Equal(t, na, d.UserDisplay)
 	assert.Equal(t, na, d.AcceleratorsDisplay)
 }
+
+func TestAiRuntimeTaskOfSkipsNonAirTasks(t *testing.T) {
+	run := &jobs.Run{RunId: 5, Tasks: []jobs.RunTask{
+		notebookTask("prep", 51),
+		aiRuntimeTask("train", 52, "/Users/me@example.com/exp", "GPU_1xA10", 1),
+	}}
+	task := aiRuntimeTaskOf(run)
+	require.NotNil(t, task)
+	assert.Equal(t, "/Users/me@example.com/exp", task.Experiment)
+}
+
+// multiTaskGetServer serves runs/get for a job with a notebook prep task and two
+// ai_runtime_tasks, and stubs every other request.
+func multiTaskGetServer(t *testing.T) *httptest.Server {
+	t.Helper()
+	run := jobs.Run{
+		RunId: 5,
+		State: &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateTerminated, ResultState: jobs.RunResultStateSuccess},
+		Tasks: []jobs.RunTask{
+			notebookTask("prep", 51),
+			aiRuntimeTask("train", 52, "/Users/me@example.com/exp-train", "GPU_1xA10", 1),
+			aiRuntimeTask("eval", 53, "/Users/me@example.com/exp-eval", "GPU_8xH100", 8),
+		},
+	}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/api/2.2/jobs/runs/get" {
+			_ = json.NewEncoder(w).Encode(run)
+			return
+		}
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	t.Cleanup(srv.Close)
+	return srv
+}
+
+func runGetJSONTask(t *testing.T, w *databricks.WorkspaceClient, taskKey string) (string, error) {
+	t.Helper()
+	var buf bytes.Buffer
+	ctx := cmdctx.SetWorkspaceClient(t.Context(), w)
+	ctx = cmdio.InContext(ctx, cmdio.NewIO(ctx, flags.OutputJSON, nil, &buf, &buf, "", ""))
+	cmd := withOutput(newGetCommand(), flags.OutputJSON)
+	cmd.SetContext(ctx)
+	if taskKey != "" {
+		require.NoError(t, cmd.Flags().Set("task-key", taskKey))
+	}
+	err := cmd.RunE(cmd, []string{"5"})
+	return buf.String(), err
+}
+
+func TestGetRunSelectsAirTask(t *testing.T) {
+	srv := multiTaskGetServer(t)
+	w, err := databricks.NewWorkspaceClient(&databricks.Config{Host: srv.URL, Token: "token", WorkspaceID: "67890"})
+	require.NoError(t, err)
+
+	type got struct {
+		Data struct {
+			TaskKey        *string `json:"task_key"`
+			ExperimentName *string `json:"experiment_name"`
+		} `json:"data"`
+	}
+
+	t.Run("defaults to the first AIR task, skipping the notebook", func(t *testing.T) {
+		out, err := runGetJSONTask(t, w, "")
+		require.NoError(t, err)
+		var g got
+		require.NoError(t, json.Unmarshal([]byte(out), &g))
+		require.NotNil(t, g.Data.TaskKey)
+		assert.Equal(t, "train", *g.Data.TaskKey)
+		require.NotNil(t, g.Data.ExperimentName)
+		assert.Equal(t, "exp-train", *g.Data.ExperimentName)
+	})
+
+	t.Run("--task-key selects a later AIR task", func(t *testing.T) {
+		out, err := runGetJSONTask(t, w, "eval")
+		require.NoError(t, err)
+		var g got
+		require.NoError(t, json.Unmarshal([]byte(out), &g))
+		require.NotNil(t, g.Data.TaskKey)
+		assert.Equal(t, "eval", *g.Data.TaskKey)
+		require.NotNil(t, g.Data.ExperimentName)
+		assert.Equal(t, "exp-eval", *g.Data.ExperimentName)
+	})
+
+	t.Run("--task-key must name an AIR task", func(t *testing.T) {
+		out, err := runGetJSONTask(t, w, "prep")
+		require.ErrorIs(t, err, root.ErrAlreadyPrinted)
+		var e errorEnvelope
+		require.NoError(t, json.Unmarshal([]byte(out), &e))
+		assert.Equal(t, jsonError{Code: "INVALID_ARGS", Kind: "PERMANENT", Message: `run 5 has no AI Runtime task "prep"; its AI Runtime tasks are: train, eval`}, e.Error)
+	})
+}
diff --git a/cmd/air/joblist.go b/cmd/air/joblist.go
index 680febb6338..f860a378f05 100644
--- a/cmd/air/joblist.go
+++ b/cmd/air/joblist.go
@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"net/http"
+	"slices"
 
 	"github.com/databricks/databricks-sdk-go"
 	"github.com/databricks/databricks-sdk-go/apierr"
@@ -16,20 +17,100 @@ import (
 // of run ids from the AiTrainingService index.
 const hydrateConcurrency = 16
 
-// firstTask returns the run's first task, unwrapping a foreach sweep to the
-// iterated task, or nil when the run has no tasks.
-func firstTask(r *jobs.Run) *jobs.RunTask {
-	if len(r.Tasks) == 0 {
+// isAirTask reports whether a task run is an AI runtime workload: an
+// ai_runtime_task, or a legacy gen_ai_compute_task with a training script. A
+// foreach task counts when the task it iterates is one.
+func isAirTask(t *jobs.RunTask) bool {
+	if t.ForEachTask != nil {
+		inner := t.ForEachTask.Task
+		return inner.AiRuntimeTask != nil ||
+			(inner.GenAiComputeTask != nil && inner.GenAiComputeTask.TrainingScriptPath != "")
+	}
+	return t.AiRuntimeTask != nil ||
+		(t.GenAiComputeTask != nil && t.GenAiComputeTask.TrainingScriptPath != "")
+}
+
+// firstAirTaskIndex returns the index in r.Tasks of the run's first AI runtime
+// task in task order, or -1 when it has none. A job may put other tasks before
+// it (the official multi-task DAB runs a notebook prep step first), so the
+// first task is not inspected directly.
+func firstAirTaskIndex(r *jobs.Run) int {
+	for i := range r.Tasks {
+		if isAirTask(&r.Tasks[i]) {
+			return i
+		}
+	}
+	return -1
+}
+
+// airTaskKeys returns the task keys of the run's AI runtime tasks in task
+// order, each once: a retried task appears in r.Tasks once per attempt.
+func airTaskKeys(r *jobs.Run) []string {
+	var keys []string
+	for i := range r.Tasks {
+		t := &r.Tasks[i]
+		if isAirTask(t) && !slices.Contains(keys, t.TaskKey) {
+			keys = append(keys, t.TaskKey)
+		}
+	}
+	return keys
+}
+
+// taskRun returns a copy of r narrowed to the task runs (every attempt, in
+// order) of the AI runtime task with the given key, so the display helpers,
+// which read a run's tasks, report that task alone. It returns nil when key is
+// not one of the run's AI runtime tasks.
+func taskRun(r *jobs.Run, key string) *jobs.Run {
+	if !slices.Contains(airTaskKeys(r), key) {
+		return nil
+	}
+	narrowed := *r
+	narrowed.Tasks = nil
+	for i := range r.Tasks {
+		if r.Tasks[i].TaskKey == key {
+			narrowed.Tasks = append(narrowed.Tasks, r.Tasks[i])
+		}
+	}
+	return &narrowed
+}
+
+// airWrapperTask returns the run's first AI runtime task run as the Jobs API
+// reports it (for a sweep, the foreach wrapper, which carries the state and
+// timing), or nil when the run has none.
+func airWrapperTask(r *jobs.Run) *jobs.RunTask {
+	i := firstAirTaskIndex(r)
+	if i < 0 {
 		return nil
 	}
-	t := &r.Tasks[0]
-	if t.AiRuntimeTask != nil || t.GenAiComputeTask != nil {
+	return &r.Tasks[i]
+}
+
+// leadTask returns the task run whose state and timing stand for the run: its
+// first AI runtime task (for a sweep, the foreach wrapper), else, for a run
+// without one, its first task; nil when the run has no tasks.
+func leadTask(r *jobs.Run) *jobs.RunTask {
+	if t := airWrapperTask(r); t != nil {
 		return t
 	}
+	if len(r.Tasks) == 0 {
+		return nil
+	}
+	return &r.Tasks[0]
+}
+
+// firstTask returns the run's first AI runtime task, unwrapping a foreach sweep
+// to the iterated task. A run without one falls back to its first task, so a
+// legacy run's fields are still read; nil when the run has no tasks.
+func firstTask(r *jobs.Run) *jobs.RunTask {
+	t := leadTask(r)
+	if t == nil {
+		return nil
+	}
 	if t.ForEachTask != nil {
 		// The foreach inner task is a jobs.Task; expose the AI fields via a
 		// synthetic RunTask so callers can treat both shapes uniformly.
 		return &jobs.RunTask{
+			TaskKey:          t.TaskKey,
 			AiRuntimeTask:    t.ForEachTask.Task.AiRuntimeTask,
 			GenAiComputeTask: t.ForEachTask.Task.GenAiComputeTask,
 		}
@@ -37,20 +118,16 @@ func firstTask(r *jobs.Run) *jobs.RunTask {
 	return t
 }
 
-// isAirRun reports whether a run is an AI runtime workload: an ai_runtime_task,
-// or a legacy gen_ai_compute_task with a training script.
+// isAirRun reports whether a run is an AI runtime workload: it has an
+// ai_runtime_task, or a legacy gen_ai_compute_task with a training script.
 func isAirRun(r *jobs.Run) bool {
-	t := firstTask(r)
-	if t == nil {
-		return false
-	}
-	return t.AiRuntimeTask != nil ||
-		(t.GenAiComputeTask != nil && t.GenAiComputeTask.TrainingScriptPath != "")
+	return firstAirTaskIndex(r) >= 0
 }
 
-// isSweep reports whether the run's first task fans out into iterations.
+// isSweep reports whether the run's AI runtime task fans out into iterations.
 func isSweep(r *jobs.Run) bool {
-	return len(r.Tasks) > 0 && r.Tasks[0].ForEachTask != nil
+	t := leadTask(r)
+	return t != nil && t.ForEachTask != nil
 }
 
 // retryTasks returns the run's task runs with only the fields that pick its
@@ -96,14 +173,13 @@ func jobCompute(r *jobs.Run) (string, int) {
 }
 
 // jobTiming returns the run's start and end times (epoch ms), preferring the
-// first task's window so a run reports its task attempt rather than the wrapper.
+// AI runtime task's window so a run reports its task attempt rather than the
+// wrapper.
 func jobTiming(r *jobs.Run) (startMillis, endMillis int64) {
 	startMillis, endMillis = r.StartTime, r.EndTime
-	if len(r.Tasks) > 0 {
-		if t := r.Tasks[0]; t.StartTime > 0 {
-			startMillis = t.StartTime
-			endMillis = t.EndTime
-		}
+	if t := leadTask(r); t != nil && t.StartTime > 0 {
+		startMillis = t.StartTime
+		endMillis = t.EndTime
 	}
 	return startMillis, endMillis
 }
diff --git a/cmd/air/joblist_test.go b/cmd/air/joblist_test.go
index 11e5efac8e9..494bf4716ca 100644
--- a/cmd/air/joblist_test.go
+++ b/cmd/air/joblist_test.go
@@ -5,6 +5,7 @@ import (
 	"net/http/httptest"
 	"testing"
 
+	"github.com/databricks/databricks-sdk-go/service/jobs"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 )
@@ -45,3 +46,145 @@ func TestFetchJobRunParsesAiRuntimeTask(t *testing.T) {
 	assert.Equal(t, "GPU_1xA10", gpuType)
 	assert.Equal(t, 1, count)
 }
+
+// notebookTask builds a non-AIR task run, like the prep step that precedes the
+// ai_runtime_task in the official multi-task DAB example.
+func notebookTask(key string, runID int64) jobs.RunTask {
+	return jobs.RunTask{TaskKey: key, RunId: runID, NotebookTask: &jobs.NotebookTask{NotebookPath: "/prep"}}
+}
+
+// aiRuntimeTask builds a direct ai_runtime_task task run.
+func aiRuntimeTask(key string, runID int64, experiment, accelType string, count int) jobs.RunTask {
+	return jobs.RunTask{TaskKey: key, RunId: runID, AiRuntimeTask: &jobs.AiRuntimeTask{
+		Experiment: experiment,
+		Deployments: []jobs.DeploymentSpec{{
+			Compute: jobs.ComputeSpec{AcceleratorType: jobs.ComputeSpecAcceleratorType(accelType), AcceleratorCount: count},
+		}},
+	}}
+}
+
+// forEachAiRuntimeTask builds a foreach (sweep) task whose iterated task is an
+// ai_runtime_task.
+func forEachAiRuntimeTask(key string, runID int64, experiment string) jobs.RunTask {
+	return jobs.RunTask{TaskKey: key, RunId: runID, ForEachTask: &jobs.RunForEachTask{
+		Task: jobs.Task{AiRuntimeTask: &jobs.AiRuntimeTask{Experiment: experiment}},
+	}}
+}
+
+func TestAirTaskSelection(t *testing.T) {
+	tests := []struct {
+		name       string
+		tasks      []jobs.RunTask
+		air        bool
+		keys       []string
+		experiment string
+		sweep      bool
+	}{
+		{
+			name:       "AIR task first",
+			tasks:      []jobs.RunTask{aiRuntimeTask("train", 11, "exp-train", "GPU_1xA10", 1), notebookTask("report", 12)},
+			air:        true,
+			keys:       []string{"train"},
+			experiment: "exp-train",
+		},
+		{
+			name:       "AIR task second",
+			tasks:      []jobs.RunTask{notebookTask("prep", 11), aiRuntimeTask("train", 12, "exp-train", "GPU_1xA10", 1)},
+			air:        true,
+			keys:       []string{"train"},
+			experiment: "exp-train",
+		},
+		{
+			name:       "two AIR tasks",
+			tasks:      []jobs.RunTask{aiRuntimeTask("a", 11, "exp-a", "GPU_1xA10", 1), aiRuntimeTask("b", 12, "exp-b", "GPU_8xH100", 8)},
+			air:        true,
+			keys:       []string{"a", "b"},
+			experiment: "exp-a",
+		},
+		{
+			name:       "foreach AIR task",
+			tasks:      []jobs.RunTask{forEachAiRuntimeTask("sweep", 11, "exp-sweep")},
+			air:        true,
+			keys:       []string{"sweep"},
+			experiment: "exp-sweep",
+			sweep:      true,
+		},
+		{
+			name:       "foreach AIR task after a notebook",
+			tasks:      []jobs.RunTask{notebookTask("prep", 11), forEachAiRuntimeTask("sweep", 12, "exp-sweep")},
+			air:        true,
+			keys:       []string{"sweep"},
+			experiment: "exp-sweep",
+			sweep:      true,
+		},
+		{
+			name:  "retries share one key",
+			tasks: []jobs.RunTask{notebookTask("prep", 11), aiRuntimeTask("train", 12, "exp-train", "GPU_1xA10", 1), aiRuntimeTask("train", 13, "exp-train", "GPU_1xA10", 1)},
+			air:   true, keys: []string{"train"}, experiment: "exp-train",
+		},
+		{
+			name:  "no AIR task",
+			tasks: []jobs.RunTask{notebookTask("prep", 11)},
+		},
+		{
+			name: "no tasks",
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			run := &jobs.Run{RunId: 1, Tasks: tt.tasks}
+			assert.Equal(t, tt.air, isAirRun(run), "isAirRun")
+			assert.Equal(t, tt.keys, airTaskKeys(run), "airTaskKeys")
+			assert.Equal(t, tt.experiment, jobExperiment(run), "jobExperiment")
+			assert.Equal(t, tt.sweep, isSweep(run), "isSweep")
+			if len(tt.tasks) == 0 {
+				assert.Nil(t, firstTask(run), "firstTask")
+			}
+		})
+	}
+}
+
+func TestTaskRunNarrowsToOneAirTask(t *testing.T) {
+	run := &jobs.Run{
+		RunId: 1,
+		State: &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateRunning},
+		Tasks: []jobs.RunTask{
+			notebookTask("prep", 11),
+			aiRuntimeTask("a", 12, "exp-a", "GPU_1xA10", 1),
+			{TaskKey: "a", RunId: 13, AttemptNumber: 1, AiRuntimeTask: &jobs.AiRuntimeTask{Experiment: "exp-a"}},
+			aiRuntimeTask("b", 14, "exp-b", "GPU_8xH100", 8),
+		},
+	}
+
+	a := taskRun(run, "a")
+	require.NotNil(t, a)
+	assert.Equal(t, int64(1), a.RunId)
+	assert.Equal(t, run.State, a.State)
+	require.Len(t, a.Tasks, 2, "both attempts of task a, in order")
+	assert.Equal(t, int64(12), a.Tasks[0].RunId)
+	assert.Equal(t, int64(13), a.Tasks[1].RunId)
+	assert.Equal(t, "exp-a", jobExperiment(a))
+
+	b := taskRun(run, "b")
+	require.NotNil(t, b)
+	assert.Equal(t, "exp-b", jobExperiment(b))
+	gpu, n := jobCompute(b)
+	assert.Equal(t, "GPU_8xH100", gpu)
+	assert.Equal(t, 8, n)
+
+	assert.Len(t, run.Tasks, 4, "the source run is left intact")
+	assert.Nil(t, taskRun(run, "prep"), "a non-AIR task is not selectable")
+	assert.Nil(t, taskRun(run, "missing"))
+}
+
+func TestJobTimingUsesTheAirTask(t *testing.T) {
+	prep := notebookTask("prep", 11)
+	prep.StartTime, prep.EndTime = 100, 150
+	train := aiRuntimeTask("train", 12, "exp", "GPU_1xA10", 1)
+	train.StartTime, train.EndTime = 200, 260
+	run := &jobs.Run{RunId: 1, StartTime: 90, EndTime: 300, Tasks: []jobs.RunTask{prep, train}}
+
+	start, end := jobTiming(run)
+	assert.Equal(t, int64(200), start)
+	assert.Equal(t, int64(260), end)
+}
diff --git a/cmd/air/list.go b/cmd/air/list.go
index 1627470dc8a..335de42fb39 100644
--- a/cmd/air/list.go
+++ b/cmd/air/list.go
@@ -37,7 +37,10 @@ type listData struct {
 // machine-readable output; fields tagged `json:"-"` are shown only in the
 // human-readable table.
 type listRow struct {
-	RunID           string  `json:"run_id"`
+	RunID string `json:"run_id"`
+	// TaskKey names the AI runtime task the row shows. A job with several AI
+	// runtime tasks has one row per task, all with the same RunID.
+	TaskKey         string  `json:"task_key"`
 	RunName         string  `json:"run_name"`
 	User            string  `json:"user"`
 	Status          string  `json:"status"`
@@ -68,6 +71,7 @@ type listRow struct {
 type listedRun struct {
 	row      listRow
 	runID    int64
+	taskKey  string
 	tasks    []jobs.RunTask // see retryTasks
 	terminal bool
 }
@@ -347,10 +351,12 @@ func (s *jobsScanStrategy) next(want int) ([]listedRun, error) {
 		if s.userFilter != "" && run.CreatorUserName != s.userFilter {
 			continue
 		}
-		if !s.filters.matches(run) {
-			continue
+		// One row per AI runtime task, each filtered on its own task.
+		for _, e := range buildListRows(run, s.host, s.workspaceID) {
+			if s.filters.matchesFields(e.fields) {
+				entries = append(entries, e.entry)
+			}
 		}
-		entries = append(entries, listedRun{row: buildListRow(run, s.host, s.workspaceID), runID: run.RunId, tasks: retryTasks(run), terminal: isTerminal(run)})
 	}
 	return entries, nil
 }
diff --git a/cmd/air/list_cache.go b/cmd/air/list_cache.go
index 941327e93b1..898c3747fa7 100644
--- a/cmd/air/list_cache.go
+++ b/cmd/air/list_cache.go
@@ -14,7 +14,7 @@ import (
 // TTL matches AICM's ~60-day retention, after which the run drops out of the
 // index anyway.
 const (
-	listCacheComponent = "air-list-runs-v2"
+	listCacheComponent = "air-list-runs-v3"
 	listCacheTTL       = 60 * 24 * time.Hour
 )
 
@@ -25,11 +25,18 @@ type listCacheKey struct {
 	RunID int64  `json:"run_id"`
 }
 
-// cachedRun is the persisted value: every listRow field (including the
+// cachedRuns is the persisted value for a run: one cachedRun per AI runtime
+// task (see buildListRows).
+type cachedRuns struct {
+	Rows []cachedRun `json:"rows"`
+}
+
+// cachedRun is one persisted list row: every listRow field (including the
 // table-only columns, which listRow tags json:"-" and so wouldn't survive a
 // direct marshal), the filter inputs, and the submit time.
 type cachedRun struct {
 	RunID           string       `json:"run_id"`
+	TaskKey         string       `json:"task_key"`
 	RunName         string       `json:"run_name"`
 	User            string       `json:"user"`
 	Status          string       `json:"status"`
@@ -56,7 +63,7 @@ type cachedRun struct {
 
 func (c cachedRun) toRow() listRow {
 	return listRow{
-		RunID: c.RunID, RunName: c.RunName, User: c.User, Status: c.Status,
+		RunID: c.RunID, TaskKey: c.TaskKey, RunName: c.RunName, User: c.User, Status: c.Status,
 		StartedAt: c.StartedAt, IsSweep: c.IsSweep, ExperimentName: c.ExperimentName,
 		DurationSeconds: c.DurationSeconds, AcceleratorType: c.AcceleratorType, NumAccelerators: c.NumAccelerators, DashboardURL: c.DashboardURL, Experiment: c.Experiment,
 		Duration: c.Duration, MLflowURL: c.MLflowURL, MLflowLabel: c.MLflowLabel,
@@ -66,7 +73,7 @@ func (c cachedRun) toRow() listRow {
 
 func cachedRunFromRow(r listRow, tasks []jobs.RunTask, fields filterFields, submitTimeMs int64) cachedRun {
 	return cachedRun{
-		RunID: r.RunID, RunName: r.RunName, User: r.User, Status: r.Status,
+		RunID: r.RunID, TaskKey: r.TaskKey, RunName: r.RunName, User: r.User, Status: r.Status,
 		StartedAt: r.StartedAt, IsSweep: r.IsSweep, ExperimentName: r.ExperimentName,
 		DurationSeconds: r.DurationSeconds, AcceleratorType: r.AcceleratorType, NumAccelerators: r.NumAccelerators, DashboardURL: r.DashboardURL, Experiment: r.Experiment,
 		Duration: r.Duration, MLflowURL: r.MLflowURL, MLflowLabel: r.MLflowLabel,
@@ -81,27 +88,39 @@ func newListCache(ctx context.Context) *cache.Cache {
 	return cache.NewCache(ctx, listCacheComponent, listCacheTTL, nil)
 }
 
-// cachedRow returns the cached row, filter fields, and task runs for a run.
-// Entries written before task runs were cached are treated as misses.
-func cachedRow(ctx context.Context, c *cache.Cache, host string, runID int64) (listRow, filterFields, []jobs.RunTask, bool) {
-	entry, ok := cache.Get[cachedRun](ctx, c, listCacheKey{Host: host, RunID: runID})
-	if !ok || len(entry.Tasks) == 0 {
-		return listRow{}, filterFields{}, nil, false
+// cachedRows returns the cached rows of a run, one per AI runtime task. An
+// entry with a row that lacks task runs is treated as a miss.
+func cachedRows(ctx context.Context, c *cache.Cache, host string, runID int64) ([]cachedRun, bool) {
+	entry, ok := cache.Get[cachedRuns](ctx, c, listCacheKey{Host: host, RunID: runID})
+	if !ok || len(entry.Rows) == 0 {
+		return nil, false
 	}
-	return entry.toRow(), entry.Fields, entry.Tasks, true
+	for _, row := range entry.Rows {
+		if len(row.Tasks) == 0 {
+			return nil, false
+		}
+	}
+	return entry.Rows, true
 }
 
-// putRow caches a terminal run's base row and filter fields under its submit time.
-func putRow(ctx context.Context, c *cache.Cache, host string, runID int64, tasks []jobs.RunTask, submitTimeMs int64, row listRow, fields filterFields) {
-	cache.Put(ctx, c, listCacheKey{Host: host, RunID: runID}, cachedRunFromRow(row, tasks, fields, submitTimeMs))
+// putRows caches a terminal run's rows.
+func putRows(ctx context.Context, c *cache.Cache, host string, runID int64, rows []cachedRun) {
+	cache.Put(ctx, c, listCacheKey{Host: host, RunID: runID}, cachedRuns{Rows: rows})
 }
 
-// updateCachedRow replaces the display row while preserving cache metadata.
-func updateCachedRow(ctx context.Context, c *cache.Cache, host string, runID int64, row listRow) {
+// updateCachedRow replaces the display row of the run's task while preserving
+// cache metadata. A run or task that is not cached is left alone.
+func updateCachedRow(ctx context.Context, c *cache.Cache, host string, runID int64, taskKey string, row listRow) {
 	key := listCacheKey{Host: host, RunID: runID}
-	entry, ok := cache.Get[cachedRun](ctx, c, key)
+	entry, ok := cache.Get[cachedRuns](ctx, c, key)
 	if !ok {
 		return
 	}
-	cache.Put(ctx, c, key, cachedRunFromRow(row, entry.Tasks, entry.Fields, entry.SubmitTimeMs))
+	for i, cached := range entry.Rows {
+		if cached.TaskKey == taskKey {
+			entry.Rows[i] = cachedRunFromRow(row, cached.Tasks, cached.Fields, cached.SubmitTimeMs)
+			cache.Put(ctx, c, key, entry)
+			return
+		}
+	}
 }
diff --git a/cmd/air/list_cache_test.go b/cmd/air/list_cache_test.go
index 00c821ac010..71001e00d87 100644
--- a/cmd/air/list_cache_test.go
+++ b/cmd/air/list_cache_test.go
@@ -1,6 +1,7 @@
 package aircmd
 
 import (
+	"context"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
@@ -12,39 +13,74 @@ import (
 	"github.com/stretchr/testify/require"
 )
 
+// putRow caches a single-row run, the common case in these tests.
+func putRow(ctx context.Context, c *cache.Cache, host string, runID int64, tasks []jobs.RunTask, submitTimeMs int64, row listRow, fields filterFields) {
+	putRows(ctx, c, host, runID, []cachedRun{cachedRunFromRow(row, tasks, fields, submitTimeMs)})
+}
+
 func TestListCacheRoundTrip(t *testing.T) {
 	t.Setenv("DATABRICKS_CACHE_DIR", t.TempDir())
 	ctx := t.Context()
 	c := newListCache(ctx)
 
-	_, _, _, ok := cachedRow(ctx, c, "https://host.test", 42)
+	_, ok := cachedRows(ctx, c, "https://host.test", 42)
 	require.False(t, ok, "miss before write")
 
 	row := listRow{
-		RunID: "42", Status: "SUCCESS", ExperimentName: new("exp"), DurationSeconds: new(int64(12)),
+		RunID: "42", TaskKey: "train", Status: "SUCCESS", ExperimentName: new("exp"), DurationSeconds: new(int64(12)),
 		AcceleratorType: new("GPU_1xA10"), NumAccelerators: new(1), DashboardURL: "https://host.test/jobs/runs/42?o=1", Experiment: "exp",
 	}
 	fields := filterFields{Experiment: "exp", GPUType: "GPU_1xA10", GPUCount: 1}
 	tasks := []jobs.RunTask{{RunId: 420, TaskKey: "train"}, {RunId: 421, TaskKey: "train", AttemptNumber: 1}}
 	putRow(ctx, c, "https://host.test", 42, tasks, 1700000000000, row, fields)
 
-	got, gotFields, gotTasks, ok := cachedRow(ctx, c, "https://host.test", 42)
+	got, ok := cachedRows(ctx, c, "https://host.test", 42)
 	require.True(t, ok, "hit after write")
-	assert.Equal(t, row, got)
-	assert.Equal(t, fields, gotFields)
+	require.Len(t, got, 1)
+	assert.Equal(t, row, got[0].toRow())
+	assert.Equal(t, fields, got[0].Fields)
+	gotTasks := got[0].Tasks
 	for i := range gotTasks {
 		gotTasks[i].ForceSendFields = nil // set by the JSON round trip
 	}
 	assert.Equal(t, tasks, gotTasks)
 
 	// Different host is a different key.
-	_, _, _, ok = cachedRow(ctx, c, "https://other.test", 42)
+	_, ok = cachedRows(ctx, c, "https://other.test", 42)
 	assert.False(t, ok)
 
-	// Entries from previous cache formats have no task runs and must be refreshed.
+	// Entries from previous cache formats have no rows and must be refreshed.
 	cache.Put(ctx, c, listCacheKey{Host: "https://host.test", RunID: 43}, map[string]any{"run_id": "43", "task_run_id": 430})
-	_, _, _, ok = cachedRow(ctx, c, "https://host.test", 43)
+	_, ok = cachedRows(ctx, c, "https://host.test", 43)
 	assert.False(t, ok)
+	cache.Put(ctx, c, listCacheKey{Host: "https://host.test", RunID: 44}, cachedRuns{Rows: []cachedRun{{RunID: "44"}}})
+	_, ok = cachedRows(ctx, c, "https://host.test", 44)
+	assert.False(t, ok, "a row without task runs is a miss")
+}
+
+func TestListCacheKeepsOneRowPerAirTask(t *testing.T) {
+	t.Setenv("DATABRICKS_CACHE_DIR", t.TempDir())
+	ctx := t.Context()
+	c := newListCache(ctx)
+	host := "https://host.test"
+
+	putRows(ctx, c, host, 42, []cachedRun{
+		cachedRunFromRow(listRow{RunID: "42", TaskKey: "train", Experiment: "exp-a"}, []jobs.RunTask{{RunId: 420, TaskKey: "train"}}, filterFields{Experiment: "exp-a"}, 1000),
+		cachedRunFromRow(listRow{RunID: "42", TaskKey: "eval", Experiment: "exp-b"}, []jobs.RunTask{{RunId: 421, TaskKey: "eval"}}, filterFields{Experiment: "exp-b"}, 1000),
+	})
+
+	// The MLflow link of one task updates that row alone.
+	updateCachedRow(ctx, c, host, 42, "eval", listRow{RunID: "42", TaskKey: "eval", Experiment: "exp-b", MLflowURL: "https://host.test/ml/x"})
+	updateCachedRow(ctx, c, host, 42, "missing", listRow{RunID: "42", TaskKey: "missing"})
+
+	got, ok := cachedRows(ctx, c, host, 42)
+	require.True(t, ok)
+	require.Len(t, got, 2)
+	assert.Equal(t, "train", got[0].TaskKey)
+	assert.Empty(t, got[0].MLflowURL)
+	assert.Equal(t, "eval", got[1].TaskKey)
+	assert.Equal(t, "https://host.test/ml/x", got[1].MLflowURL)
+	assert.Equal(t, int64(1000), got[1].SubmitTimeMs, "cache metadata is preserved")
 }
 
 func TestIndexStrategyServesCachedRowWithoutFetch(t *testing.T) {
diff --git a/cmd/air/list_detail.go b/cmd/air/list_detail.go
index 2045bb819d5..5197833150e 100644
--- a/cmd/air/list_detail.go
+++ b/cmd/air/list_detail.go
@@ -12,8 +12,9 @@ import (
 )
 
 // runDetailText fetches a run and renders the same styled view as `air get` into
-// a string, for the list picker's detail pane.
-func runDetailText(ctx context.Context, w *databricks.WorkspaceClient, runID int64) (string, error) {
+// a string, for the list picker's detail pane. taskKey selects the AI runtime
+// task shown, as `air get --task-key` does; "" means the first.
+func runDetailText(ctx context.Context, w *databricks.WorkspaceClient, runID int64, taskKey string) (string, error) {
 	run, err := w.Jobs.GetRun(ctx, jobs.GetRunRequest{RunId: runID})
 	if err != nil {
 		if errors.Is(err, apierr.ErrResourceDoesNotExist) {
@@ -21,8 +22,14 @@ func runDetailText(ctx context.Context, w *databricks.WorkspaceClient, runID int
 		}
 		return "", fmt.Errorf("failed to fetch run: %w", err)
 	}
+	fullRun := run
+	run, err = selectAirTask(run, taskKey)
+	if err != nil {
+		return "", err
+	}
 
 	data := buildGetData(run)
+	annotateTask(fullRun, run, &data)
 	// As in `air get`: an ai_runtime_task run keeps its experiment, accelerators
 	// and config path on the task.
 	if genAIComputeTask(run) == nil {
diff --git a/cmd/air/list_detail_test.go b/cmd/air/list_detail_test.go
index 29813d96300..de108720ece 100644
--- a/cmd/air/list_detail_test.go
+++ b/cmd/air/list_detail_test.go
@@ -83,7 +83,7 @@ func TestListRowAndPanesLinkTheSameOlderRetry(t *testing.T) {
 		currentUser: selfUser, limit: 10, fetchMLflow: true,
 	}).next(10)
 	require.NoError(t, err)
-	info, err := runDetailText(ctx, w, 7)
+	info, err := runDetailText(ctx, w, 7, "")
 	require.NoError(t, err)
 	logs, err := runLogsSnapshot(ctx, w, 7)
 	require.NoError(t, err)
@@ -135,7 +135,7 @@ func TestUnnamedOlderRetryLabelFitsTheListColumnAndIsWholeInTheInfoPane(t *testi
 		currentUser: selfUser, limit: 10, fetchMLflow: true,
 	}).next(10)
 	require.NoError(t, err)
-	info, err := runDetailText(ctx, w, 7)
+	info, err := runDetailText(ctx, w, 7, "")
 	require.NoError(t, err)
 
 	// Then the list row keeps the end of the ID and the whole retry note within
@@ -177,7 +177,7 @@ func TestRunDetailTextShowsAiRuntimeTaskFields(t *testing.T) {
 	t.Cleanup(srv.Close)
 	ctx := cmdio.MockDiscard(t.Context())
 
-	info, err := runDetailText(ctx, newTestWorkspaceClient(t, srv.URL), 7)
+	info, err := runDetailText(ctx, newTestWorkspaceClient(t, srv.URL), 7, "")
 	require.NoError(t, err)
 
 	// The pane shows what the list row and `air get` show, with the experiment
@@ -187,3 +187,32 @@ func TestRunDetailTextShowsAiRuntimeTaskFields(t *testing.T) {
 	assert.Contains(t, info, "Accelerators  1x A10")
 	assert.Contains(t, info, "command: python train.py")
 }
+
+func TestRunDetailTextShowsTheSelectedAirTask(t *testing.T) {
+	run := jobs.Run{RunId: 7, State: &jobs.RunState{LifeCycleState: jobs.RunLifeCycleStateTerminated, ResultState: jobs.RunResultStateSuccess}, Tasks: []jobs.RunTask{
+		notebookTask("prep", 71),
+		aiRuntimeTask("train", 72, "/Users/me@example.com/exp-train", "GPU_1xA10", 1),
+		aiRuntimeTask("eval", 73, "/Users/me@example.com/exp-eval", "GPU_8xH100", 8),
+	}}
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path == "/api/2.2/jobs/runs/get" {
+			_ = json.NewEncoder(w).Encode(run)
+			return
+		}
+		_, _ = w.Write([]byte(`{}`))
+	}))
+	t.Cleanup(srv.Close)
+	ctx := cmdio.MockDiscard(t.Context())
+	w := newTestWorkspaceClient(t, srv.URL)
+
+	info, err := runDetailText(ctx, w, 7, "")
+	require.NoError(t, err)
+	assert.Contains(t, info, "Experiment    exp-train", "the default is the first AIR task, not the notebook")
+	assert.Contains(t, info, "Task          train")
+
+	info, err = runDetailText(ctx, w, 7, "eval")
+	require.NoError(t, err)
+	assert.Contains(t, info, "Experiment    exp-eval")
+	assert.Contains(t, info, "Accelerators  8x H100")
+	assert.Contains(t, info, "Task          eval")
+}
diff --git a/cmd/air/list_format.go b/cmd/air/list_format.go
index cda11b49a58..a828fab6312 100644
--- a/cmd/air/list_format.go
+++ b/cmd/air/list_format.go
@@ -6,10 +6,45 @@ import (
 	"github.com/databricks/databricks-sdk-go/service/jobs"
 )
 
-// buildListRow extracts the columns shown for one run. Optional cells fall back
-// to "-"; MLflowURL starts as "-" and setMLflowLinks fills it in for text output.
-// host and workspaceID are used for building dashboard URLs.
+// listedRunFields is a list entry with the task-filter inputs it was built from.
+type listedRunFields struct {
+	entry  listedRun
+	fields filterFields
+}
+
+// buildListRows builds one list entry per AI runtime task of the run, in task
+// order, each read from that task's own task runs (see taskRun). A run without
+// an AI runtime task yields a single entry read from the run as is.
+func buildListRows(run *jobs.Run, host string, workspaceID int64) []listedRunFields {
+	keys := airTaskKeys(run)
+	if len(keys) == 0 {
+		return []listedRunFields{listedRunFor(run, host, workspaceID)}
+	}
+	entries := make([]listedRunFields, 0, len(keys))
+	for _, key := range keys {
+		entries = append(entries, listedRunFor(taskRun(run, key), host, workspaceID))
+	}
+	return entries
+}
+
+// listedRunFor builds the list entry for a run (or a run narrowed to one task).
+func listedRunFor(run *jobs.Run, host string, workspaceID int64) listedRunFields {
+	row := buildListRow(run, host, workspaceID)
+	return listedRunFields{
+		entry:  listedRun{row: row, runID: run.RunId, taskKey: row.TaskKey, tasks: retryTasks(run), terminal: isTerminal(run)},
+		fields: filterFieldsFromRun(run),
+	}
+}
+
+// buildListRow extracts the columns shown for one run, read from its first AI
+// runtime task. Optional cells fall back to "-"; MLflowURL starts as "-" and
+// setMLflowLinks fills it in for text output. host and workspaceID are used for
+// building dashboard URLs.
 func buildListRow(run *jobs.Run, host string, workspaceID int64) listRow {
+	taskKey := ""
+	if t := airWrapperTask(run); t != nil {
+		taskKey = t.TaskKey
+	}
 	experiment := jobExperiment(run)
 	experimentName := nilIfEmpty(experiment)
 	if experiment == "" {
@@ -38,6 +73,7 @@ func buildListRow(run *jobs.Run, host string, workspaceID int64) listRow {
 
 	return listRow{
 		RunID:           strconv.FormatInt(run.RunId, 10),
+		TaskKey:         taskKey,
 		RunName:         run.RunName,
 		User:            run.CreatorUserName,
 		Status:          displayRunStatus(run),
diff --git a/cmd/air/list_index.go b/cmd/air/list_index.go
index 3bd1839f7b6..9469bae2559 100644
--- a/cmd/air/list_index.go
+++ b/cmd/air/list_index.go
@@ -106,9 +106,11 @@ func (s *indexStrategy) hydrate(ids []int64) ([]listedRun, error) {
 	// The cache key excludes the filter, so a cached row is still run through the
 	// active filter; a non-matching hit is dropped, not re-fetched.
 	for _, id := range ids {
-		if row, fields, tasks, ok := cachedRow(s.ctx, s.cache, host, id); ok {
-			if s.filters.matchesFields(fields) {
-				rows = append(rows, listedRun{row: row, runID: id, tasks: tasks, terminal: true})
+		if cached, ok := cachedRows(s.ctx, s.cache, host, id); ok {
+			for _, c := range cached {
+				if s.filters.matchesFields(c.Fields) {
+					rows = append(rows, listedRun{row: c.toRow(), runID: id, taskKey: c.TaskKey, tasks: c.Tasks, terminal: true})
+				}
 			}
 			continue
 		}
@@ -120,16 +122,19 @@ func (s *indexStrategy) hydrate(ids []int64) ([]listedRun, error) {
 		return nil, err
 	}
 	for _, run := range runs {
-		fields := filterFieldsFromRun(run)
-		if !s.filters.matchesFields(fields) {
-			continue
+		// One row per AI runtime task; the whole set is cached so a later lookup
+		// by run id can re-filter every task.
+		entries := buildListRows(run, s.host, s.workspaceID)
+		start, _ := jobTiming(run)
+		cached := make([]cachedRun, 0, len(entries))
+		for _, e := range entries {
+			cached = append(cached, cachedRunFromRow(e.entry.row, e.entry.tasks, e.fields, start))
+			if s.filters.matchesFields(e.fields) {
+				rows = append(rows, e.entry)
+			}
 		}
-		row := buildListRow(run, s.host, s.workspaceID)
-		tasks := retryTasks(run)
-		rows = append(rows, listedRun{row: row, runID: run.RunId, tasks: tasks, terminal: isTerminal(run)})
 		if isTerminal(run) {
-			start, _ := jobTiming(run)
-			putRow(s.ctx, s.cache, host, run.RunId, tasks, start, row, fields)
+			putRows(s.ctx, s.cache, host, run.RunId, cached)
 		}
 	}
 
@@ -146,7 +151,7 @@ func (s *indexStrategy) hydrate(ids []int64) ([]listedRun, error) {
 func (s *indexStrategy) cacheMLflowLinks(entries []listedRun) {
 	for _, entry := range entries {
 		if entry.row.MLflowURL != "" && entry.row.MLflowURL != "-" {
-			updateCachedRow(s.ctx, s.cache, s.host, entry.runID, entry.row)
+			updateCachedRow(s.ctx, s.cache, s.host, entry.runID, entry.taskKey, entry.row)
 		}
 	}
 }
diff --git a/cmd/air/list_test.go b/cmd/air/list_test.go
index 5a3490d538a..5d79a5b59a6 100644
--- a/cmd/air/list_test.go
+++ b/cmd/air/list_test.go
@@ -240,7 +240,7 @@ func TestBuildListRow(t *testing.T) {
 	body, err := json.Marshal(row)
 	require.NoError(t, err)
 	assert.JSONEq(t, `{
-		"run_id":"123","run_name":"run-123","user":"me@example.com","status":"SUCCESS",
+		"run_id":"123","task_key":"","run_name":"run-123","user":"me@example.com","status":"SUCCESS",
 		"started_at":"2023-11-14T22:13:20+00:00","is_sweep":false,
 		"experiment_name":"exp","duration_seconds":12,
 		"accelerator_type":"GPU_8xH100","num_accelerators":8,
@@ -274,7 +274,7 @@ func TestBuildListRowWaitingForComputeStatus(t *testing.T) {
 	assert.Equal(t, "PENDING", row.Status)
 	assert.Equal(t, listRowStatus(row), row.Status)
 	assert.JSONEq(t, `{
-		"run_id":"123","run_name":"run-123","user":"me@example.com","status":"PENDING",
+		"run_id":"123","task_key":"","run_name":"run-123","user":"me@example.com","status":"PENDING",
 		"started_at":null,"is_sweep":false,"experiment_name":"exp","duration_seconds":null,
 		"accelerator_type":"GPU_8xH100","num_accelerators":8,
 		"dashboard_url":"https://example.test/jobs/runs/123?o=456"
@@ -296,7 +296,7 @@ func TestBuildListRowMissingValues(t *testing.T) {
 	body, err := json.Marshal(row)
 	require.NoError(t, err)
 	assert.JSONEq(t, `{
-		"run_id":"7","run_name":"","user":"","status":"UNKNOWN","started_at":null,"is_sweep":false,
+		"run_id":"7","task_key":"","run_name":"","user":"","status":"UNKNOWN","started_at":null,"is_sweep":false,
 		"experiment_name":null,"duration_seconds":null,"accelerator_type":null,"num_accelerators":null,
 		"dashboard_url":"https://example.test/jobs/runs/7"
 	}`, string(body))
@@ -406,3 +406,73 @@ func TestListValidatesAcceleratorOnlyAfterNoMatches(t *testing.T) {
 		})
 	}
 }
+
+func TestBuildListRowsOnePerAirTask(t *testing.T) {
+	run := jobs.Run{RunId: 7, RunName: "dag", CreatorUserName: "me@example.com", Tasks: []jobs.RunTask{
+		notebookTask("prep", 71),
+		aiRuntimeTask("train", 72, "/Users/me@example.com/exp-a", "GPU_1xA10", 1),
+		aiRuntimeTask("eval", 73, "/Users/me@example.com/exp-b", "GPU_8xH100", 8),
+	}}
+
+	rows := buildListRows(&run, "https://example.test", 0)
+	require.Len(t, rows, 2)
+	assert.Equal(t, "7", rows[0].entry.row.RunID)
+	assert.Equal(t, "train", rows[0].entry.row.TaskKey)
+	assert.Equal(t, "exp-a", rows[0].entry.row.Experiment)
+	assert.Equal(t, "1x A10", rows[0].entry.row.Accelerators)
+	assert.Equal(t, []int64{72}, taskRunIDs(rows[0].entry.tasks), "the row's MLflow lookup sees only its own task runs")
+	assert.Equal(t, "7", rows[1].entry.row.RunID)
+	assert.Equal(t, "eval", rows[1].entry.row.TaskKey)
+	assert.Equal(t, "exp-b", rows[1].entry.row.Experiment)
+	assert.Equal(t, "8x H100", rows[1].entry.row.Accelerators)
+	assert.Equal(t, []int64{73}, taskRunIDs(rows[1].entry.tasks))
+}
+
+func taskRunIDs(tasks []jobs.RunTask) []int64 {
+	ids := make([]int64, len(tasks))
+	for i, t := range tasks {
+		ids[i] = t.RunId
+	}
+	return ids
+}
+
+func TestListAirRunsAirTaskAfterNotebook(t *testing.T) {
+	// The official multi-task DAB example: a notebook prep task, then the
+	// ai_runtime_task that depends on it.
+	dag := airBaseRun(1, "me@example.com", "GPU_1xA10", 1, "/Users/me@example.com/air-byot-multitask-dag")
+	dag.Tasks[0].TaskKey, dag.Tasks[0].RunId = "train", 12
+	dag.Tasks = []jobs.RunTask{notebookTask("prep", 11), dag.Tasks[0]}
+	srv := runsServer(t, runsListBody(t, "", dag))
+
+	rows, err := newRunFetcher(t.Context(), newTestWorkspaceClient(t, srv.URL), listQuery{activeOnly: true}).next(10)
+	require.NoError(t, err)
+	require.Len(t, rows, 1)
+	assert.Equal(t, "1", rows[0].RunID)
+	assert.Equal(t, "train", rows[0].TaskKey)
+	assert.Equal(t, "air-byot-multitask-dag", rows[0].Experiment)
+	assert.Equal(t, "1x A10", rows[0].Accelerators)
+}
+
+func TestListAirRunsOneRowPerAirTask(t *testing.T) {
+	two := airBaseRun(1, "me@example.com", "GPU_1xA10", 1, "/Users/me@example.com/exp-a")
+	two.Tasks = []jobs.RunTask{
+		aiRuntimeTask("a", 11, "/Users/me@example.com/exp-a", "GPU_1xA10", 1),
+		aiRuntimeTask("b", 12, "/Users/me@example.com/exp-b", "GPU_8xH100", 8),
+	}
+	srv := runsServer(t, runsListBody(t, "", two, airBaseRun(2, "me@example.com", "GPU_1xH100", 1, "/Users/me@example.com/exp-c")))
+
+	rows, err := newRunFetcher(t.Context(), newTestWorkspaceClient(t, srv.URL), listQuery{
+		activeOnly: true,
+		filters:    listFilters{Experiment: "exp-b"},
+	}).next(10)
+	require.NoError(t, err)
+	require.Len(t, rows, 1, "the task filter applies per AIR task")
+	assert.Equal(t, "1", rows[0].RunID)
+	assert.Equal(t, "b", rows[0].TaskKey)
+
+	rows, err = newRunFetcher(t.Context(), newTestWorkspaceClient(t, srv.URL), listQuery{activeOnly: true}).next(10)
+	require.NoError(t, err)
+	require.Len(t, rows, 3)
+	assert.Equal(t, []string{"1", "1", "2"}, []string{rows[0].RunID, rows[1].RunID, rows[2].RunID})
+	assert.Equal(t, []string{"a", "b", ""}, []string{rows[0].TaskKey, rows[1].TaskKey, rows[2].TaskKey})
+}
diff --git a/cmd/air/list_tui.go b/cmd/air/list_tui.go
index 53ae3b3980f..5f102bcf6b1 100644
--- a/cmd/air/list_tui.go
+++ b/cmd/air/list_tui.go
@@ -285,7 +285,7 @@ func (m listModel) Update(msg tea.Msg) (tea.Model, tea.Cmd) {
 				m.detailLoading = true
 				m.detailTitle = "Run Details"
 				runID, _ := parseRunID(m.rows[m.cursor].RunID)
-				return m, m.fetchRunDetail(runID)
+				return m, m.fetchRunDetail(runID, m.rows[m.cursor].TaskKey)
 			}
 		case "L", "l":
 			// Open the logs snapshot pane; the fetch fills it in.
@@ -375,12 +375,13 @@ func (m listModel) renderHint() string {
 	return faint.Render(hint)
 }
 
-// fetchRunDetail returns a tea.Cmd that fetches run details in the background.
-func (m listModel) fetchRunDetail(runID int64) tea.Cmd {
+// fetchRunDetail returns a tea.Cmd that fetches the details of the run's task
+// in the background.
+func (m listModel) fetchRunDetail(runID int64, taskKey string) tea.Cmd {
 	ctx := m.fetcher.ctx
 	w := m.fetcher.w
 	return func() tea.Msg {
-		body, err := runDetailText(ctx, w, runID)
+		body, err := runDetailText(ctx, w, runID, taskKey)
 		if err != nil {
 			return detailMsg{title: "Run Details", err: err}
 		}
diff --git a/cmd/air/render.go b/cmd/air/render.go
index 6e2f3f5b1a3..87ca14bcce5 100644
--- a/cmd/air/render.go
+++ b/cmd/air/render.go
@@ -74,6 +74,9 @@ func newPalette(r *lipgloss.Renderer) palette {
 type runView struct {
 	runID        string
 	dashboardURL string
+	// task is the selected AI runtime task's key, shown only when the run has
+	// several AI runtime tasks to tell apart.
+	task         string
 	status       string
 	submitted    string
 	retries      int
@@ -107,6 +110,7 @@ func renderRunText(ctx context.Context, out io.Writer, w *databricks.WorkspaceCl
 	view := runView{
 		runID:        data.RunID,
 		dashboardURL: data.DashboardURL,
+		task:         data.TaskDisplay,
 		status:       data.Status,
 		submitted:    data.SubmittedDisplay,
 		retries:      data.AttemptNumber,
@@ -156,12 +160,14 @@ func renderRunText(ctx context.Context, out io.Writer, w *databricks.WorkspaceCl
 	}
 }
 
-// genAIComputeTask returns the run's first GenAI-compute task, or nil.
+// genAIComputeTask returns the run's first AI runtime task's GenAI-compute
+// task, or nil.
 func genAIComputeTask(run *jobs.Run) *jobs.GenAiComputeTask {
-	if len(run.Tasks) == 0 {
+	t := firstTask(run)
+	if t == nil {
 		return nil
 	}
-	return run.Tasks[0].GenAiComputeTask
+	return t.GenAiComputeTask
 }
 
 // aiRuntimeEnvironmentVersion returns the serverless environment version an
@@ -386,12 +392,17 @@ func renderFields(p palette, colorOn bool, v runView) string {
 	status := statusStyle(p, v.status).Render("● " + v.status)
 	rows := []string{
 		field(p, "Run ID", link(colorOn, p.blue, v.runID, v.dashboardURL)),
+	}
+	if v.task != "" {
+		rows = append(rows, field(p, "Task", p.n12.Render(stripControl(v.task))))
+	}
+	rows = append(rows,
 		field(p, "Status", status),
 		field(p, "Submitted", p.n12.Render(v.submitted)),
 		field(p, "Retries", p.n12.Render(strconv.Itoa(v.retries))),
 		field(p, "Max Retries", p.n12.Render(v.maxRetries)),
 		field(p, "Duration", p.n12.Render(v.duration)),
-	}
+	)
 	// Progress is shown only for a running run we could estimate; amber marks it
 	// as live, in-progress information.
 	if v.progress != "" {