ronantakizawa/codereview-bench huggingface dataset:
Repository: apache/airflow
File: airflow-core/src/airflow/api_fastapi/execution_api/versions/v2025_09_23.py (Python)

Code before:
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from __future__ import annotations

from cadwyn import VersionChange, schema

from airflow.api_fastapi.execution_api.datamodels.taskinstance import TaskInstance


class AddDagVersionIdField(VersionChange):
    """Add the `dag_version_id` field to the TaskInstance model."""

    description = __doc__

    instructions_to_migrate_to_previous_version = (schema(TaskInstance).field("dag_version_id").didnt_exist,)


class AcceptNullXComSetValue(VersionChange):
    """Allow POST /xcoms/... value body to be JSON null in addition to JSON strings/objects/arrays."""

    description = __doc__
    # This is a backward-compatible relaxation of input validation; older clients are unaffected.
    # No response transformation is needed when downgrading.
    instructions_to_migrate_to_previous_version = ()


Review comment (refactor):
We can remove this since technically `pydanticJsonValue` supports `None`

Code after:
# Licensed to the Apache Software Foundation (ASF) under one
# or more contributor license agreements.  See the NOTICE file
# distributed with this work for additional information
# regarding copyright ownership.  The ASF licenses this file
# to you under the Apache License, Version 2.0 (the
# "License"); you may not use this file except in compliance
# with the License.  You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing,
# software distributed under the License is distributed on an
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from __future__ import annotations

from cadwyn import VersionChange, schema

from airflow.api_fastapi.execution_api.datamodels.taskinstance import TaskInstance


class AddDagVersionIdField(VersionChange):
    """Add the `dag_version_id` field to the TaskInstance model."""

    description = __doc__

    instructions_to_migrate_to_previous_version = (schema(TaskInstance).field("dag_version_id").didnt_exist,)
Repository: apache/airflow
File: airflow-core/src/airflow/ui/src/components/PoolBar.tsx (TypeScript)

Code before:
            const flexValue = totalSlots > 0 ? (slotValue / totalSlots) * 100 : 0;

            if (flexValue === 0) {
              return undefined;
            }

            const slotType = key.replace("_slots", "");

            const poolContent = (
              <Flex
                alignItems="center"
                bg={`${color}.solid`}
                color={`${color}.contrast`}
                gap={1}
                h="100%"
                justifyContent="center"
                px={1}
                textAlign="center"
                title={`${slotType}: ${slotValue}`}
                w="100%"
              >
                {icon}
                {flexValue > 5 && (
                  <Text fontSize="xs" fontWeight="bold">
                    {slotValue}
                  </Text>
                )}
              </Flex>
            );

            return color !== "success" && "name" in pool ? (
              <Link asChild key={key} w={`${flexValue}%`}>
                <RouterLink
                  to={`/task_instances?${SearchParamsKeys.STATE}=${color}&${SearchParamsKeys.POOL}=${pool.name}`}
                >
                  {poolContent}
                </RouterLink>
              </Link>
            ) : (
              <Box key={key} w={`${flexValue}%`}>
                {poolContent}
              </Box>
            );
          })}
      </Flex>
    </Tooltip>
  );
};


Review comment (suggestion):
Now we can have really 'thin' segment, and we have no icon etc... I think we shouldn't change that part. (or at least not in this PR).

<img width="1381" height="194" alt="Image" src="https://github.com/user-attachments/assets/b5bbb848-fbe3-42e0-a703-351834460074" />

Code after:
                </Flex>
              </Tooltip>
            );

            return slot.color !== "success" && "name" in pool ? (
              <Link asChild flex={flexValue} key={slot.key}>
                <RouterLink
                  to={`/task_instances?${SearchParamsKeys.STATE}=${slot.color}&${SearchParamsKeys.POOL}=${pool.name}`}
                >
                  {poolContent}
                </RouterLink>
              </Link>
            ) : (
              <Box flex={flexValue} key={slot.key}>
                {poolContent}
              </Box>
            );
          })}
      </Flex>

      <HStack gap={4} wrap="wrap">
        {preparedSlots
          .filter((slot) => infoSlots.includes(slot.slotType) && slot.slotValue > 0)
          .map((slot) => (
            <HStack gap={1} key={slot.key}>
              <StateIcon size={12} state={slot.slotType} />
              <Text color="fg.muted" fontSize="xs" fontWeight="medium">
                {slot.label}: {slot.slotValue}
              </Text>
            </HStack>
          ))}
      </HStack>
    </VStack>
  );
};
Repository: apache/airflow
File: airflow-core/src/airflow/listeners/spec/taskinstance.py (Python)

Code before:

@hookspec
def on_task_instance_failed(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
    error: None | str | BaseException,
):
    """Execute when task state changes to FAIL. previous_state can be None."""


@hookspec
def on_task_instance_skipped(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
):
    """
    Execute when a task instance skips itself by raising AirflowSkipException.

    This hook is called only when a task has started execution and then
    intentionally skips itself by raising AirflowSkipException.

    Note: This does NOT cover tasks skipped by:
        - Trigger rules (e.g., upstream failures)
        - BranchPythonOperator (tasks not in selected branch)
        - ShortCircuitOperator
        - Scheduler-level decisions

    For comprehensive skip tracking, use DAG-level listeners
    (on_dag_run_success/on_dag_run_failed) which provide complete task state.

    :param previous_state: Previous state of the task instance (can be None)
    :param task_instance: The task instance object (RuntimeTaskInstance when called
        from task execution context, TaskInstance when called from API server)
    """


Review comment (bug):
```suggestion
    Note: This function will NOT cover tasks that were skipped by scheduler, before execution began, such as:
        - Skips due to trigger rules (e.g., upstream failures)
        - Skips from operators like BranchPythonOperator, ShortCircuitOperator, or similar mechanisms
        - Any other situation in which the scheduler decides not to schedule a task for execution
```

Code after:

@hookspec
def on_task_instance_failed(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
    error: None | str | BaseException,
):
    """Execute when task state changes to FAIL. previous_state can be None."""


@hookspec
def on_task_instance_skipped(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
):
    """
    Execute when a task instance skips itself during execution.

    This hook is called only when a task has started execution and then
    intentionally skips itself (e.g., by raising AirflowSkipException).

    Note: This function will NOT cover tasks that were skipped by scheduler, before execution began, such as:
        - Skips due to trigger rules (e.g., upstream failures)
        - Skips from operators like BranchPythonOperator, ShortCircuitOperator, or similar mechanisms
        - Any other situation in which the scheduler decides not to schedule a task for execution

    For comprehensive tracking of skipped tasks, use DAG-level listeners
    (on_dag_run_success/on_dag_run_failed) which may have access to all task states.

    :param previous_state: Previous state of the task instance (can be None)
    :param task_instance: The task instance object (RuntimeTaskInstance when called
        from task execution context, TaskInstance when called from API server)
    """
Repository: apache/airflow
File: airflow-core/src/airflow/listeners/spec/taskinstance.py (Python)

Code before:
def on_task_instance_failed(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
    error: None | str | BaseException,
):
    """Execute when task state changes to FAIL. previous_state can be None."""


@hookspec
def on_task_instance_skipped(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
):
    """
    Execute when a task instance skips itself by raising AirflowSkipException.

    This hook is called only when a task has started execution and then
    intentionally skips itself by raising AirflowSkipException.

    Note: This does NOT cover tasks skipped by:
        - Trigger rules (e.g., upstream failures)
        - BranchPythonOperator (tasks not in selected branch)
        - ShortCircuitOperator
        - Scheduler-level decisions

    For comprehensive skip tracking, use DAG-level listeners
    (on_dag_run_success/on_dag_run_failed) which provide complete task state.

    :param previous_state: Previous state of the task instance (can be None)
    :param task_instance: The task instance object (RuntimeTaskInstance when called
        from task execution context, TaskInstance when called from API server)
    """


Review comment (suggestion):
```suggestion
    For comprehensive tracking of skipped tasks, use DAG-level listeners
```

Code after:
def on_task_instance_failed(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
    error: None | str | BaseException,
):
    """Execute when task state changes to FAIL. previous_state can be None."""


@hookspec
def on_task_instance_skipped(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
):
    """
    Execute when a task instance skips itself during execution.

    This hook is called only when a task has started execution and then
    intentionally skips itself (e.g., by raising AirflowSkipException).

    Note: This function will NOT cover tasks that were skipped by scheduler, before execution began, such as:
        - Skips due to trigger rules (e.g., upstream failures)
        - Skips from operators like BranchPythonOperator, ShortCircuitOperator, or similar mechanisms
        - Any other situation in which the scheduler decides not to schedule a task for execution

    For comprehensive tracking of skipped tasks, use DAG-level listeners
    (on_dag_run_success/on_dag_run_failed) which may have access to all task states.

    :param previous_state: Previous state of the task instance (can be None)
    :param task_instance: The task instance object (RuntimeTaskInstance when called
        from task execution context, TaskInstance when called from API server)
    """
Repository: apache/airflow
File: airflow-core/src/airflow/listeners/spec/taskinstance.py (Python)

Code before:
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
    error: None | str | BaseException,
):
    """Execute when task state changes to FAIL. previous_state can be None."""


@hookspec
def on_task_instance_skipped(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
):
    """
    Execute when a task instance skips itself by raising AirflowSkipException.

    This hook is called only when a task has started execution and then
    intentionally skips itself by raising AirflowSkipException.

    Note: This does NOT cover tasks skipped by:
        - Trigger rules (e.g., upstream failures)
        - BranchPythonOperator (tasks not in selected branch)
        - ShortCircuitOperator
        - Scheduler-level decisions

    For comprehensive skip tracking, use DAG-level listeners
    (on_dag_run_success/on_dag_run_failed) which provide complete task state.

    :param previous_state: Previous state of the task instance (can be None)
    :param task_instance: The task instance object (RuntimeTaskInstance when called
        from task execution context, TaskInstance when called from API server)
    """


Review comment (bug):
```suggestion
    (on_dag_run_success/on_dag_run_failed) which may have access to all task states.
```

Code after:
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
    error: None | str | BaseException,
):
    """Execute when task state changes to FAIL. previous_state can be None."""


@hookspec
def on_task_instance_skipped(
    previous_state: TaskInstanceState | None,
    task_instance: RuntimeTaskInstance | TaskInstance,
):
    """
    Execute when a task instance skips itself during execution.

    This hook is called only when a task has started execution and then
    intentionally skips itself (e.g., by raising AirflowSkipException).

    Note: This function will NOT cover tasks that were skipped by scheduler, before execution began, such as:
        - Skips due to trigger rules (e.g., upstream failures)
        - Skips from operators like BranchPythonOperator, ShortCircuitOperator, or similar mechanisms
        - Any other situation in which the scheduler decides not to schedule a task for execution

    For comprehensive tracking of skipped tasks, use DAG-level listeners
    (on_dag_run_success/on_dag_run_failed) which may have access to all task states.

    :param previous_state: Previous state of the task instance (can be None)
    :param task_instance: The task instance object (RuntimeTaskInstance when called
        from task execution context, TaskInstance when called from API server)
    """
Repository: apache/airflow
File: airflow-core/src/airflow/ui/src/pages/TaskInstance/ExtraLinks.tsx (TypeScript)

Code before:

import { useTaskInstanceServiceGetExtraLinks } from "openapi/queries";
import { useAutoRefresh } from "src/utils";

export const ExtraLinks = () => {
  const { t: translate } = useTranslation("dag");
  const { dagId = "", mapIndex = "-1", runId = "", taskId = "" } = useParams();

  const refetchInterval = useAutoRefresh({ dagId });

  const { data } = useTaskInstanceServiceGetExtraLinks(
    {
      dagId,
      dagRunId: runId,
      mapIndex: parseInt(mapIndex, 10),
      taskId,
    },
    undefined,
    {
      refetchInterval: (query) => {
        // If we have extra links, stop polling to reduce server load
        const hasLinks =
          query.state.data?.extra_links && Object.keys(query.state.data.extra_links).length > 0;

        // Continue polling if auto-refresh is enabled and we don't have links yet
        return !hasLinks && refetchInterval ? refetchInterval : false;
      },
    },
  );

  return data && Object.keys(data.extra_links).length > 0 ? (
    <Box py={1}>
      <Heading size="sm">{translate("extraLinks")}</Heading>
      <HStack gap={2} py={2}>
        {Object.entries(data.extra_links).map(([key, value], _) =>
          value === null ? undefined : (
            <Button asChild colorPalette="brand" key={key} variant="surface">
              <a href={value} rel="noopener noreferrer" target="_blank">
                {key}
              </a>
            </Button>
          ),
        )}
      </HStack>
    </Box>
  ) : undefined;
};


Review comment (suggestion):
This will poll forever if the task doesn't have extra links. Lets also call `useTaskServiceGetTask` and see if `task.extraLinks.length > 0` first.

We probably want to make that change in `TaskInstance/Details.tsx` to check if we even want to render the extra links component at all

Code after:

import { useTaskInstanceServiceGetExtraLinks } from "openapi/queries";

type ExtraLinksProps = {
  readonly refetchInterval: number | false;
};

export const ExtraLinks = ({ refetchInterval }: ExtraLinksProps) => {
  const { t: translate } = useTranslation("dag");
  const { dagId = "", mapIndex = "-1", runId = "", taskId = "" } = useParams();

  const { data } = useTaskInstanceServiceGetExtraLinks(
    {
      dagId,
      dagRunId: runId,
      mapIndex: parseInt(mapIndex, 10),
      taskId,
    },
    undefined,
    {
      refetchInterval,
    },
  );

  return data && Object.keys(data.extra_links).length > 0 ? (
    <Box py={1}>
      <Heading size="sm">{translate("extraLinks")}</Heading>
      <HStack gap={2} py={2}>
        {Object.entries(data.extra_links).map(([key, value], _) =>
          value === null ? undefined : (
            <Button asChild colorPalette="brand" key={key} variant="surface">
              <a href={value} rel="noopener noreferrer" target="_blank">
                {key}
              </a>
            </Button>
          ),
        )}
      </HStack>
    </Box>
  ) : undefined;
};
Repository: apache/airflow
File: task-sdk/tests/task_sdk/execution_time/test_task_runner.py (Python)

Code before:
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from __future__ import annotations

import contextlib
import functools
import json
import os
import textwrap
import time
from collections.abc import Iterable
from datetime import datetime, timedelta, timezone as dt_timezone
from pathlib import Path
from typing import TYPE_CHECKING, Any
from unittest import mock
from unittest.mock import call, patch

import pandas as pd
import pytest
from task_sdk import FAKE_BUNDLE
from uuid6 import uuid7

from airflow.exceptions import AirflowRescheduleException
from airflow.listeners import hookimpl
from airflow.listeners.listener import get_listener_manager
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import (
    DAG,
    BaseOperator,
    BaseOperatorLink,
    Connection,
    dag as dag_decorator,
    get_current_context,
    task as task_decorator,
    timezone,
)
from airflow.sdk.api.datamodels._generated import (
    AssetProfile,
    AssetResponse,
    DagRun,
    DagRunState,
    PreviousTIResponse,
    TaskInstance,
    TaskInstanceState,
    TIRunContext,
)
from airflow.sdk.bases.xcom import BaseXCom
from airflow.sdk.definitions._internal.types import NOTSET, SET_DURING_EXECUTION, is_arg_set

Review comment (suggestion):
This should import from `airflow.sdk.exceptions` instead.

Code after:
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
# KIND, either express or implied.  See the License for the
# specific language governing permissions and limitations
# under the License.

from __future__ import annotations

import contextlib
import functools
import json
import os
import textwrap
import time
from collections.abc import Iterable
from datetime import datetime, timedelta, timezone as dt_timezone
from pathlib import Path
from typing import TYPE_CHECKING, Any
from unittest import mock
from unittest.mock import call, patch

import pandas as pd
import pytest
from task_sdk import FAKE_BUNDLE
from uuid6 import uuid7

from airflow.listeners import hookimpl
from airflow.providers.standard.operators.python import PythonOperator
from airflow.sdk import (
    DAG,
    BaseOperator,
    BaseOperatorLink,
    Connection,
    dag as dag_decorator,
    get_current_context,
    task as task_decorator,
    timezone,
)
from airflow.sdk.api.datamodels._generated import (
    AssetProfile,
    AssetResponse,
    DagRun,
    DagRunState,
    PreviousTIResponse,
    TaskInstance,
    TaskInstanceState,
    TIRunContext,
)
from airflow.sdk.bases.xcom import BaseXCom
from airflow.sdk.definitions._internal.types import NOTSET, SET_DURING_EXECUTION, is_arg_set
from airflow.sdk.definitions.asset import Asset, AssetAlias, AssetUniqueKey, Dataset, Model
from airflow.sdk.definitions.param import DagParam
Repository: apache/airflow
File: airflow-core/src/airflow/ui/tests/e2e/pages/EventsPage.ts (TypeScript)

Code before:
      return [];
    }
  }

  /**
   * Get all event types from the current page
   */
  public async getEventTypes(): Promise<Array<string>> {
    try {
      const rows = await this.getEventLogRows();

      if (rows.length === 0) {
        return [];
      }

      const eventTypes: Array<string> = [];

      for (const row of rows) {
        const cells = row.locator("td");
        const eventCell = cells.nth(1);
        const text = await eventCell.textContent();

        if (text !== null) {
          eventTypes.push(text.trim());
        }
      }

      return eventTypes;
    } catch {
      return [];
    }
  }

  /**
   * Check if next page is available
   */
  public async hasNextPage(): Promise<boolean> {
    try {
      return await this.paginationNextButton.isEnabled();
    } catch {
      return false;
    }
  }

  /**
   * Check if previous page is available
   */
  public async hasPreviousPage(): Promise<boolean> {
    try {
      return await this.paginationPrevButton.isEnabled();
    } catch {

Review comment (bug):
Methods like isAuditLogTableVisible(), hasNextPage(), getEventLogCount() catch errors and return false This makes debugging difficult and can cause tests to pass when they shouldn't.

Code after:

    if (!allPages) {
      return eventTypes;
    }

    const allEventTypes = [...eventTypes];

    while (await this.hasNextPage()) {
      await this.clickNextPage();
      const pageEvents = await this.getEventTypes(false);

      allEventTypes.push(...pageEvents);
    }

    while ((await this.paginationPrevButton.count()) > 0 && (await this.paginationPrevButton.isEnabled())) {
      await this.clickPrevPage();
    }

    return allEventTypes;
  }

  public async hasNextPage(): Promise<boolean> {
    const count = await this.paginationNextButton.count();

    if (count === 0) {
      return false;
    }

    return await this.paginationNextButton.isEnabled();
  }

  public async navigateToAuditLog(dagId: string, limit?: number): Promise<void> {
    this.currentDagId = dagId;
    this.currentLimit = limit;

    const baseUrl = EventsPage.getEventsUrl(dagId);
    const url = limit === undefined ? baseUrl : `${baseUrl}?offset=0&limit=${limit}`;

    await this.page.goto(url, {
      timeout: 30_000,
      waitUntil: "domcontentloaded",
    });
    await this.waitForTableLoad();
  }

  /**
   * Wait for table to finish loading
   */
  public async waitForTableLoad(): Promise<void> {
    await this.eventsTable.waitFor({ state: "visible", timeout: 60_000 });