From 7bf66199f90f86b2c334d7349cddddface035b29 Mon Sep 17 00:00:00 2001
From: Sameer Mesiah <smesiah971@gmail.com>
Date: Sat, 11 Jul 2026 22:29:14 +0100
Subject: [PATCH 1/6] Create a default pool for each team when it is created
 and automatically assign tasks without an explicit pool to the team's default
 pool during DAG parsing. Add unit tests covering pool assignment and team
 creation, and update the multi-team documentation to describe the new
 behaviour.

---
 .../docs/core-concepts/multi-team.rst         |  5 ++
 airflow-core/newsfragments/69768.feature.rst  |  1 +
 .../src/airflow/cli/commands/team_command.py  | 17 +++++
 .../src/airflow/dag_processing/dagbag.py      | 26 +++++++
 airflow-core/src/airflow/models/pool.py       |  4 ++
 .../unit/cli/commands/test_team_command.py    | 19 ++++++
 .../tests/unit/dag_processing/test_dagbag.py  | 68 +++++++++++++++++++
 airflow-core/tests/unit/models/test_pool.py   |  3 +
 8 files changed, 143 insertions(+)
 create mode 100644 airflow-core/newsfragments/69768.feature.rst

diff --git a/airflow-core/docs/core-concepts/multi-team.rst b/airflow-core/docs/core-concepts/multi-team.rst
index c587023408021..eebcb162a83b1 100644
--- a/airflow-core/docs/core-concepts/multi-team.rst
+++ b/airflow-core/docs/core-concepts/multi-team.rst
@@ -269,6 +269,11 @@ Use the ``--team-name`` option with ``airflow pools set`` to assign a pool to a
     The ``--team-name`` option is rejected when ``core.multi_team`` is disabled.
     The specified team must exist in the database (create it first with ``airflow teams create``).
 
+    When ``core.multi_team`` is enabled, ``airflow teams create`` automatically
+    creates a default pool named ``default_pool_<team_name>``. Tasks in DAG
+    bundles associated with that team that do not explicitly specify a pool will
+    use the team's default pool automatically.
+
 Creating Team-scoped Pools via the REST API
 """""""""""""""""""""""""""""""""""""""""""
 
diff --git a/airflow-core/newsfragments/69768.feature.rst b/airflow-core/newsfragments/69768.feature.rst
new file mode 100644
index 0000000000000..b8ced1f7296b2
--- /dev/null
+++ b/airflow-core/newsfragments/69768.feature.rst
@@ -0,0 +1 @@
+Add automatic creation of team default pools in multi-team deployments and assign tasks without an explicitly configured pool to their team's default pool during DAG parsing.
diff --git a/airflow-core/src/airflow/cli/commands/team_command.py b/airflow-core/src/airflow/cli/commands/team_command.py
index 931b5507be325..04a7a445fb83e 100644
--- a/airflow-core/src/airflow/cli/commands/team_command.py
+++ b/airflow-core/src/airflow/cli/commands/team_command.py
@@ -25,6 +25,7 @@
 from sqlalchemy.exc import IntegrityError
 
 from airflow.cli.simple_table import AirflowConsole
+from airflow.configuration import conf
 from airflow.dag_processing.bundles.manager import DagBundlesManager
 from airflow.models.connection import Connection
 from airflow.models.pool import Pool
@@ -74,8 +75,24 @@ def team_create(args, *, session=NEW_SESSION):
 
     try:
         session.add(new_team)
+        session.flush()
+
+        if conf.getboolean("core", "multi_team"):
+            Pool.create_or_update_pool(
+                name=Pool.get_default_team_pool_name(team_name),
+                slots=conf.getint(
+                    "core",
+                    "default_pool_task_slot_count",
+                ),
+                description=f"Default pool for team '{team_name}'",
+                include_deferred=False,
+                team_name=team_name,
+                session=session,
+            )
+
         session.commit()
         print(f"Team '{team_name}' created successfully.")
+
     except IntegrityError as e:
         session.rollback()
         raise SystemExit(f"Failed to create team '{team_name}': {e}")
diff --git a/airflow-core/src/airflow/dag_processing/dagbag.py b/airflow-core/src/airflow/dag_processing/dagbag.py
index 3bf82804bffe1..bf088c04c1b2a 100644
--- a/airflow-core/src/airflow/dag_processing/dagbag.py
+++ b/airflow-core/src/airflow/dag_processing/dagbag.py
@@ -43,6 +43,7 @@
 )
 from airflow.executors.executor_loader import ExecutorLoader
 from airflow.listeners.listener import get_listener_manager
+from airflow.models.pool import Pool
 from airflow.serialization.definitions.notset import NOTSET, ArgNotSet, is_arg_set
 from airflow.serialization.serialized_objects import LazyDeserializedDAG
 from airflow.utils.file import correct_maybe_zipped
@@ -161,6 +162,30 @@ def _validate_executor_fields(dag: DAG, bundle_name: str | None = None) -> None:
             )
 
 
+def _assign_default_team_pools(
+    dag: DAG,
+    bundle_name: str | None = None,
+) -> None:
+    """Assign the default team pool to tasks that do not explicitly specify a pool."""
+    dag_team_name = None
+
+    if conf.getboolean("core", "multi_team"):
+        if bundle_name:
+            from airflow.dag_processing.bundles.manager import DagBundlesManager
+
+            bundle_manager = DagBundlesManager()
+            bundle_config = bundle_manager._bundle_config[bundle_name]
+
+            dag_team_name = bundle_config.team_name
+
+    if not dag_team_name:
+        return
+
+    for task in dag.tasks:
+        if task.pool == Pool.DEFAULT_POOL_NAME:
+            task.pool = Pool.get_default_team_pool_name(dag_team_name)
+
+
 class DagBag(LoggingMixin):
     """
     A dagbag is a collection of dags, parsed out of a folder tree and has high level configuration settings.
@@ -344,6 +369,7 @@ def process_file(self, filepath, only_if_updated=True, safe_mode=True):
                 # Validate before adding to bag (matches original _process_modules behavior)
                 dag.validate()
                 _validate_executor_fields(dag, self.bundle_name)
+                _assign_default_team_pools(dag, self.bundle_name)
                 self.bag_dag(dag=dag)
                 bagged_dags.append(dag)
             except AirflowClusterPolicySkipDag:
diff --git a/airflow-core/src/airflow/models/pool.py b/airflow-core/src/airflow/models/pool.py
index d6a4915ea2bb2..339abaddf5dc8 100644
--- a/airflow-core/src/airflow/models/pool.py
+++ b/airflow-core/src/airflow/models/pool.py
@@ -121,6 +121,10 @@ def get_default_pool(*, session: Session = NEW_SESSION) -> Pool | None:
         """
         return Pool.get_pool(Pool.DEFAULT_POOL_NAME, session=session)
 
+    @staticmethod
+    def get_default_team_pool_name(team_name: str) -> str:
+        return f"default_pool_{team_name}"
+
     @staticmethod
     @provide_session
     def create_or_update_pool(
diff --git a/airflow-core/tests/unit/cli/commands/test_team_command.py b/airflow-core/tests/unit/cli/commands/test_team_command.py
index 1df56d9c3cc3f..12720854f1557 100644
--- a/airflow-core/tests/unit/cli/commands/test_team_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_team_command.py
@@ -79,6 +79,25 @@ def test_team_create_success(self, stdout_capture):
         assert "Team 'test-team' created successfully" in output
         assert str(team.name) in output
 
+    def test_team_create_creates_default_pool(self, stdout_capture):
+        """Test that creating a team also creates its default pool."""
+        with conf_vars(
+            {
+                ("core", "multi_team"): "True",
+                ("multi_team", "default_team_pool_slots"): "128",
+            }
+        ):
+            with stdout_capture:
+                team_command.team_create(self.parser.parse_args(["teams", "create", "team-a"]))
+
+        pool = self.session.scalar(select(Pool).where(Pool.pool == Pool.get_default_team_pool_name("team-a")))
+
+        assert pool is not None
+        assert pool.team_name == "team-a"
+        assert pool.slots == 128
+        assert pool.include_deferred is False
+        assert pool.description == "Default pool for team 'team-a'"
+
     def test_team_create_empty_name(self):
         """Test team creation with empty name."""
         with pytest.raises(SystemExit, match="Team name cannot be empty"):
diff --git a/airflow-core/tests/unit/dag_processing/test_dagbag.py b/airflow-core/tests/unit/dag_processing/test_dagbag.py
index 0ea0a29fc145b..ac3e06fb177c9 100644
--- a/airflow-core/tests/unit/dag_processing/test_dagbag.py
+++ b/airflow-core/tests/unit/dag_processing/test_dagbag.py
@@ -46,6 +46,7 @@
 from airflow.executors.executor_loader import ExecutorLoader
 from airflow.models.dag import DagModel
 from airflow.models.dagwarning import DagWarning, DagWarningType
+from airflow.models.pool import Pool
 from airflow.models.serialized_dag import SerializedDagModel
 from airflow.sdk import DAG, BaseOperator
 
@@ -383,6 +384,73 @@ def test_dag_with_bundle_name(self, tmp_path):
         for dag in dagbag2.dags.values():
             assert dag.bundle_name is None
 
+    @pytest.mark.parametrize(
+        ("team_name", "operator_args", "expected_pool"),
+        [
+            pytest.param(
+                "team_a",
+                "",
+                Pool.get_default_team_pool_name("team_a"),
+                id="default-pool-replaced",
+            ),
+            pytest.param(
+                "team_a",
+                ', pool="custom_pool"',
+                "custom_pool",
+                id="custom-pool-preserved",
+            ),
+            pytest.param(
+                None,
+                "",
+                Pool.DEFAULT_POOL_NAME,
+                id="no-team",
+            ),
+        ],
+    )
+    @patch("airflow.dag_processing.bundles.manager.DagBundlesManager")
+    def test_default_pool_replaced_with_team_pool(
+        self,
+        mock_manager,
+        tmp_path,
+        team_name,
+        operator_args,
+        expected_pool,
+    ):
+        mock_bundle = mock.MagicMock()
+        mock_bundle.team_name = team_name
+        mock_manager.return_value._bundle_config = {
+            "test_bundle": mock_bundle,
+        }
+
+        dag_file = tmp_path / "test_dag.py"
+        dag_file.write_text(
+            textwrap.dedent(
+                f"""
+                from airflow.sdk import dag
+
+                from airflow.providers.standard.operators.empty import EmptyOperator
+
+                @dag(schedule=None)
+                def my_dag():
+                    EmptyOperator(task_id="task1"{operator_args})
+
+                my_dag()
+                """
+            )
+        )
+
+        with conf_vars({("core", "multi_team"): "True"}):
+            dagbag = DagBag(
+                dag_folder=os.fspath(tmp_path),
+                bundle_name="test_bundle",
+            )
+
+        dag = dagbag.get_dag("my_dag")
+
+        assert dag is not None
+        assert not dagbag.import_errors
+        assert dag.task_dict["task1"].pool == expected_pool
+
     def test_get_existing_dag(self, tmp_path, standard_example_dags_folder):
         """
         Test that we're able to parse some example DAGs and retrieve them
diff --git a/airflow-core/tests/unit/models/test_pool.py b/airflow-core/tests/unit/models/test_pool.py
index 2c7db9d9800e4..d6bccc7ccf99c 100644
--- a/airflow-core/tests/unit/models/test_pool.py
+++ b/airflow-core/tests/unit/models/test_pool.py
@@ -286,6 +286,9 @@ def test_get_pools(self):
         assert pools[0].pool == self.pools[0].pool
         assert pools[1].pool == self.pools[1].pool
 
+    def test_default_team_pool_name(self):
+        assert Pool.get_default_team_pool_name("team_a") == "default_pool_team_a"
+
     def test_create_pool(self, session):
         self.add_pools()
         pool = Pool.create_or_update_pool(name="foo", slots=5, description="", include_deferred=True)

From 52d26a4e63895ce52944050d8214087fa2481556 Mon Sep 17 00:00:00 2001
From: Sameer Mesiah <smesiah971@gmail.com>
Date: Sat, 18 Jul 2026 20:16:51 +0100
Subject: [PATCH 2/6] Reconcile default team pools during teams sync, including
 recreating missing pools for existing teams. Automatically remove the
 system-managed default team pool during  while preserving protection for
 user-created team pools. Add tests covering reconciliation and cleanup.

---
 .../src/airflow/cli/commands/team_command.py  | 37 ++++++--
 .../unit/cli/commands/test_team_command.py    | 87 +++++++++++++++----
 2 files changed, 102 insertions(+), 22 deletions(-)

diff --git a/airflow-core/src/airflow/cli/commands/team_command.py b/airflow-core/src/airflow/cli/commands/team_command.py
index 04a7a445fb83e..2ad0e607fda5c 100644
--- a/airflow-core/src/airflow/cli/commands/team_command.py
+++ b/airflow-core/src/airflow/cli/commands/team_command.py
@@ -135,9 +135,19 @@ def team_delete(args, *, session=NEW_SESSION):
         associations.append(f"{variable_count} variable(s)")
 
     # Check pool associations
-    if pool_count := session.scalar(select(func.count(Pool.id)).where(Pool.team_name == team.name)):
+    if pool_count := session.scalar(
+        select(func.count(Pool.id)).where(
+            Pool.team_name == team.name,
+            Pool.pool != Pool.get_default_team_pool_name(team.name),
+        )
+    ):
         associations.append(f"{pool_count} pool(s)")
 
+    default_pool = session.scalar(select(Pool).where(Pool.pool == Pool.get_default_team_pool_name(team.name)))
+
+    if default_pool:
+        session.delete(default_pool)
+
     # If there are associations, prevent deletion
     if associations:
         association_list = ", ".join(associations)
@@ -189,10 +199,27 @@ def team_sync(args, *, session=NEW_SESSION):
     teams_added = 0
 
     try:
-        for team_name in dag_bundle_teams - Team.get_all_team_names(session=session):
-            team = Team(name=team_name)
-            session.add(team)
-            teams_added += 1
+        existing_teams = Team.get_all_team_names(session=session)
+        for team_name in dag_bundle_teams:
+            if team_name not in existing_teams:
+                session.add(Team(name=team_name))
+                teams_added += 1
+
+            session.flush()
+
+            if conf.getboolean("core", "multi_team"):
+                Pool.create_or_update_pool(
+                    name=Pool.get_default_team_pool_name(team_name),
+                    slots=conf.getint(
+                        "core",
+                        "default_pool_task_slot_count",
+                    ),
+                    description=f"Default pool for team '{team_name}'",
+                    include_deferred=False,
+                    team_name=team_name,
+                    session=session,
+                )
+
         session.commit()
     except Exception as e:
         session.rollback()
diff --git a/airflow-core/tests/unit/cli/commands/test_team_command.py b/airflow-core/tests/unit/cli/commands/test_team_command.py
index 12720854f1557..8b2003cef55eb 100644
--- a/airflow-core/tests/unit/cli/commands/test_team_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_team_command.py
@@ -157,20 +157,29 @@ def test_team_list_with_output_format(self):
 
     def test_team_delete_success(self, stdout_capture):
         """Test successful team deletion."""
-        # Create team first
-        team_command.team_create(self.parser.parse_args(["teams", "create", "delete-me"]))
-
-        # Verify team exists
-        team = self.session.scalar(select(Team).where(Team.name == "delete-me"))
-        assert team is not None
-
-        # Delete team with --yes flag
-        with stdout_capture as stdout:
-            team_command.team_delete(self.parser.parse_args(["teams", "delete", "delete-me", "--yes"]))
-
-        # Verify team was deleted
-        team = self.session.scalar(select(Team).where(Team.name == "delete-me"))
-        assert team is None
+        with conf_vars({("core", "multi_team"): "True"}):
+            # Create team first
+            team_command.team_create(self.parser.parse_args(["teams", "create", "delete-me"]))
+
+            # Verify team exists
+            team = self.session.scalar(select(Team).where(Team.name == "delete-me"))
+            assert team is not None
+
+            # Delete team with --yes flag
+            with stdout_capture as stdout:
+                team_command.team_delete(self.parser.parse_args(["teams", "delete", "delete-me", "--yes"]))
+
+            # Verify team was deleted
+            team = self.session.scalar(select(Team).where(Team.name == "delete-me"))
+            assert team is None
+
+            # Verify default pool was deleted
+            assert (
+                self.session.scalar(
+                    select(Pool).where(Pool.pool == Pool.get_default_team_pool_name("delete-me"))
+                )
+                is None
+            )
 
         # Verify output message
         output = stdout.getvalue()
@@ -419,6 +428,50 @@ def test_team_sync(self):
         teams = self.session.scalars(select(Team)).all()
         assert len(teams) == 2
 
-        team_names = [team.name for team in teams]
-        assert "team1" in team_names
-        assert "team2" in team_names
+        team_names = {team.name for team in teams}
+        assert team_names == {"team1", "team2"}
+
+        team1_pool = self.session.scalar(
+            select(Pool).where(Pool.pool == Pool.get_default_team_pool_name("team1"))
+        )
+        team2_pool = self.session.scalar(
+            select(Pool).where(Pool.pool == Pool.get_default_team_pool_name("team2"))
+        )
+
+        assert team1_pool is not None
+        assert team1_pool.team_name == "team1"
+
+        assert team2_pool is not None
+        assert team2_pool.team_name == "team2"
+
+    def test_team_sync_creates_missing_default_pool(self):
+        bundle_config = [
+            {
+                "name": "bundleone",
+                "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+                "kwargs": {"path": "/dev/null", "refresh_interval": 0},
+                "team_name": "team1",
+            },
+        ]
+
+        # Simulate an existing team created before automatic default pools existed.
+        self.session.add(Team(name="team1"))
+        self.session.commit()
+
+        assert (
+            self.session.scalar(select(Pool).where(Pool.pool == Pool.get_default_team_pool_name("team1")))
+            is None
+        )
+
+        with conf_vars(
+            {
+                ("core", "multi_team"): "True",
+                ("dag_processor", "dag_bundle_config_list"): json.dumps(bundle_config),
+            }
+        ):
+            team_command.team_sync(self.parser.parse_args(["teams", "sync"]))
+
+        pool = self.session.scalar(select(Pool).where(Pool.pool == Pool.get_default_team_pool_name("team1")))
+
+        assert pool is not None
+        assert pool.team_name == "team1"

From 3266993514ccd47c1fac1bc2bf71a89abb873425 Mon Sep 17 00:00:00 2001
From: Sameer Mesiah <smesiah971@gmail.com>
Date: Sat, 18 Jul 2026 22:53:36 +0100
Subject: [PATCH 3/6] Update documentation to instruct pro-active team sync and
 fix incorrect config key in test.

---
 airflow-core/docs/core-concepts/multi-team.rst            | 3 +++
 airflow-core/tests/unit/cli/commands/test_team_command.py | 2 +-
 2 files changed, 4 insertions(+), 1 deletion(-)

diff --git a/airflow-core/docs/core-concepts/multi-team.rst b/airflow-core/docs/core-concepts/multi-team.rst
index eebcb162a83b1..53712ac23389d 100644
--- a/airflow-core/docs/core-concepts/multi-team.rst
+++ b/airflow-core/docs/core-concepts/multi-team.rst
@@ -274,6 +274,9 @@ Use the ``--team-name`` option with ``airflow pools set`` to assign a pool to a
     bundles associated with that team that do not explicitly specify a pool will
     use the team's default pool automatically.
 
+    Existing multi-team deployments should run ``airflow teams sync`` after upgrading
+    to provision default pools for existing teams before DAG parsing assigns tasks to them.
+
 Creating Team-scoped Pools via the REST API
 """""""""""""""""""""""""""""""""""""""""""
 
diff --git a/airflow-core/tests/unit/cli/commands/test_team_command.py b/airflow-core/tests/unit/cli/commands/test_team_command.py
index 8b2003cef55eb..6cd5fd7a96c9a 100644
--- a/airflow-core/tests/unit/cli/commands/test_team_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_team_command.py
@@ -84,7 +84,7 @@ def test_team_create_creates_default_pool(self, stdout_capture):
         with conf_vars(
             {
                 ("core", "multi_team"): "True",
-                ("multi_team", "default_team_pool_slots"): "128",
+                ("multi_team", "default_pool_task_slot_count"): "128",
             }
         ):
             with stdout_capture:

From 4a84c82284b60edb5a8bce7917758368c771c327 Mon Sep 17 00:00:00 2001
From: Sameer Mesiah <smesiah971@gmail.com>
Date: Fri, 24 Jul 2026 00:48:18 +0100
Subject: [PATCH 4/6] Refine team default pool management

Only create missing default team pools during teams sync,
improve cleanup during team deletion, and update the
documentation and tests.
---
 .../docs/core-concepts/multi-team.rst         |  9 +--
 airflow-core/newsfragments/69768.feature.rst  |  2 +-
 .../src/airflow/cli/commands/team_command.py  | 68 +++++++++++--------
 .../unit/cli/commands/test_team_command.py    |  4 +-
 4 files changed, 45 insertions(+), 38 deletions(-)

diff --git a/airflow-core/docs/core-concepts/multi-team.rst b/airflow-core/docs/core-concepts/multi-team.rst
index 53712ac23389d..4a2a4c6e15cd0 100644
--- a/airflow-core/docs/core-concepts/multi-team.rst
+++ b/airflow-core/docs/core-concepts/multi-team.rst
@@ -270,12 +270,9 @@ Use the ``--team-name`` option with ``airflow pools set`` to assign a pool to a
     The specified team must exist in the database (create it first with ``airflow teams create``).
 
     When ``core.multi_team`` is enabled, ``airflow teams create`` automatically
-    creates a default pool named ``default_pool_<team_name>``. Tasks in DAG
-    bundles associated with that team that do not explicitly specify a pool will
-    use the team's default pool automatically.
-
-    Existing multi-team deployments should run ``airflow teams sync`` after upgrading
-    to provision default pools for existing teams before DAG parsing assigns tasks to them.
+    creates a default pool named ``default_pool_<team_name>``. By default, tasks
+    in Dag bundles associated with that team are automatically assigned to
+    the team's default pool unless another pool is explicitly configured.
 
 Creating Team-scoped Pools via the REST API
 """""""""""""""""""""""""""""""""""""""""""
diff --git a/airflow-core/newsfragments/69768.feature.rst b/airflow-core/newsfragments/69768.feature.rst
index b8ced1f7296b2..f57d07e20be17 100644
--- a/airflow-core/newsfragments/69768.feature.rst
+++ b/airflow-core/newsfragments/69768.feature.rst
@@ -1 +1 @@
-Add automatic creation of team default pools in multi-team deployments and assign tasks without an explicitly configured pool to their team's default pool during DAG parsing.
+Add automatic creation of team default pools in multi-team deployments and assign tasks without an explicitly configured pool to their team's default pool during DAG parsing. Existing multi-team deployments should run ``airflow teams sync`` after upgrading to provision default pools for existing teams.
diff --git a/airflow-core/src/airflow/cli/commands/team_command.py b/airflow-core/src/airflow/cli/commands/team_command.py
index 2ad0e607fda5c..ccc0f07d1fc3a 100644
--- a/airflow-core/src/airflow/cli/commands/team_command.py
+++ b/airflow-core/src/airflow/cli/commands/team_command.py
@@ -20,6 +20,7 @@
 from __future__ import annotations
 
 import re
+from typing import TYPE_CHECKING
 
 from sqlalchemy import func, select
 from sqlalchemy.exc import IntegrityError
@@ -35,6 +36,9 @@
 from airflow.utils.providers_configuration_loader import providers_configuration_loaded
 from airflow.utils.session import NEW_SESSION, provide_session
 
+if TYPE_CHECKING:
+    from sqlalchemy.orm import Session
+
 NO_TEAMS_LIST_MSG = "No teams found."
 
 
@@ -59,6 +63,17 @@ def _extract_team_name(args):
     return team_name
 
 
+def _create_default_team_pool(team_name: str, *, session: Session) -> None:
+    Pool.create_or_update_pool(
+        name=Pool.get_default_team_pool_name(team_name),
+        slots=conf.getint("core", "default_pool_task_slot_count"),
+        description=f"Default pool for team '{team_name}'",
+        include_deferred=False,
+        team_name=team_name,
+        session=session,
+    )
+
+
 @cli_utils.action_cli
 @providers_configuration_loaded
 @provide_session
@@ -78,17 +93,7 @@ def team_create(args, *, session=NEW_SESSION):
         session.flush()
 
         if conf.getboolean("core", "multi_team"):
-            Pool.create_or_update_pool(
-                name=Pool.get_default_team_pool_name(team_name),
-                slots=conf.getint(
-                    "core",
-                    "default_pool_task_slot_count",
-                ),
-                description=f"Default pool for team '{team_name}'",
-                include_deferred=False,
-                team_name=team_name,
-                session=session,
-            )
+            _create_default_team_pool(team_name=team_name, session=session)
 
         session.commit()
         print(f"Team '{team_name}' created successfully.")
@@ -143,11 +148,6 @@ def team_delete(args, *, session=NEW_SESSION):
     ):
         associations.append(f"{pool_count} pool(s)")
 
-    default_pool = session.scalar(select(Pool).where(Pool.pool == Pool.get_default_team_pool_name(team.name)))
-
-    if default_pool:
-        session.delete(default_pool)
-
     # If there are associations, prevent deletion
     if associations:
         association_list = ", ".join(associations)
@@ -166,6 +166,17 @@ def team_delete(args, *, session=NEW_SESSION):
     # Delete the team
     try:
         session.delete(team)
+
+        default_pool = session.scalar(
+            select(Pool).where(
+                Pool.pool == Pool.get_default_team_pool_name(team.name),
+                Pool.team_name == team.name,
+            )
+        )
+
+        if default_pool:
+            session.delete(default_pool)
+
         session.commit()
         print(f"Team '{team_name}' deleted successfully")
     except Exception as e:
@@ -190,6 +201,9 @@ def team_list(args, *, session=NEW_SESSION):
 @provide_session
 def team_sync(args, *, session=NEW_SESSION):
     """Sync missing teams from the dag bundle config."""
+    if not conf.getboolean("core", "multi_team"):
+        return
+
     dag_bundle_teams = {
         bundle.team_name
         for bundle in DagBundlesManager()._bundle_config.values()
@@ -203,22 +217,18 @@ def team_sync(args, *, session=NEW_SESSION):
         for team_name in dag_bundle_teams:
             if team_name not in existing_teams:
                 session.add(Team(name=team_name))
+                session.flush()
                 teams_added += 1
 
-            session.flush()
-
-            if conf.getboolean("core", "multi_team"):
-                Pool.create_or_update_pool(
-                    name=Pool.get_default_team_pool_name(team_name),
-                    slots=conf.getint(
-                        "core",
-                        "default_pool_task_slot_count",
-                    ),
-                    description=f"Default pool for team '{team_name}'",
-                    include_deferred=False,
-                    team_name=team_name,
-                    session=session,
+            pool = session.scalar(
+                select(Pool).where(
+                    Pool.pool == Pool.get_default_team_pool_name(team_name),
+                    Pool.team_name == team_name,
                 )
+            )
+
+            if pool is None:
+                _create_default_team_pool(team_name=team_name, session=session)
 
         session.commit()
     except Exception as e:
diff --git a/airflow-core/tests/unit/cli/commands/test_team_command.py b/airflow-core/tests/unit/cli/commands/test_team_command.py
index 6cd5fd7a96c9a..02d8a44516f0e 100644
--- a/airflow-core/tests/unit/cli/commands/test_team_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_team_command.py
@@ -84,7 +84,7 @@ def test_team_create_creates_default_pool(self, stdout_capture):
         with conf_vars(
             {
                 ("core", "multi_team"): "True",
-                ("multi_team", "default_pool_task_slot_count"): "128",
+                ("core", "default_pool_task_slot_count"): "111",
             }
         ):
             with stdout_capture:
@@ -94,7 +94,7 @@ def test_team_create_creates_default_pool(self, stdout_capture):
 
         assert pool is not None
         assert pool.team_name == "team-a"
-        assert pool.slots == 128
+        assert pool.slots == 111
         assert pool.include_deferred is False
         assert pool.description == "Default pool for team 'team-a'"
 

From 713d90eaaa58bd2f12e98746558333bd80e5d0b5 Mon Sep 17 00:00:00 2001
From: Sameer Mesiah <smesiah971@gmail.com>
Date: Thu, 30 Jul 2026 14:40:15 +0100
Subject: [PATCH 5/6] Add warning when multi-team is disabled to teams_sync and
 fix acronym in newsfragment.

---
 airflow-core/newsfragments/69768.feature.rst          | 2 +-
 airflow-core/src/airflow/cli/commands/team_command.py | 1 +
 2 files changed, 2 insertions(+), 1 deletion(-)

diff --git a/airflow-core/newsfragments/69768.feature.rst b/airflow-core/newsfragments/69768.feature.rst
index f57d07e20be17..fadc8880faa25 100644
--- a/airflow-core/newsfragments/69768.feature.rst
+++ b/airflow-core/newsfragments/69768.feature.rst
@@ -1 +1 @@
-Add automatic creation of team default pools in multi-team deployments and assign tasks without an explicitly configured pool to their team's default pool during DAG parsing. Existing multi-team deployments should run ``airflow teams sync`` after upgrading to provision default pools for existing teams.
+Add automatic creation of team default pools in multi-team deployments and assign tasks without an explicitly configured pool to their team's default pool during Dag parsing. Existing multi-team deployments should run ``airflow teams sync`` after upgrading to provision default pools for existing teams.
diff --git a/airflow-core/src/airflow/cli/commands/team_command.py b/airflow-core/src/airflow/cli/commands/team_command.py
index ccc0f07d1fc3a..c108ac203f8bf 100644
--- a/airflow-core/src/airflow/cli/commands/team_command.py
+++ b/airflow-core/src/airflow/cli/commands/team_command.py
@@ -202,6 +202,7 @@ def team_list(args, *, session=NEW_SESSION):
 def team_sync(args, *, session=NEW_SESSION):
     """Sync missing teams from the dag bundle config."""
     if not conf.getboolean("core", "multi_team"):
+        print("Warning: multi-team is not enabled; nothing to synchronize.")
         return
 
     dag_bundle_teams = {

From e1e279a1f7f6fae915cd0ff89ed59064ae3c0946 Mon Sep 17 00:00:00 2001
From: Sameer Mesiah <smesiah971@gmail.com>
Date: Sun, 2 Aug 2026 19:50:15 +0100
Subject: [PATCH 6/6] Add teams verify CLI command

Adds a new  command to validate multi-team
configuration consistency. The command checks that every team has its
system-managed default pool, that DAG bundles reference existing teams,
and that team-owned pools follow the expected naming convention. Any
issues are reported together before the command exits with a non-zero
status.

Added unit tests covering successful verification, disabled multi-team
mode, missing default pools, and invalid DAG bundle team references.
---
 airflow-core/newsfragments/70947.feature.rst  |  1 +
 airflow-core/src/airflow/cli/cli_config.py    |  7 +++
 .../src/airflow/cli/commands/team_command.py  | 43 +++++++++++++
 .../unit/cli/commands/test_team_command.py    | 60 +++++++++++++++++++
 4 files changed, 111 insertions(+)
 create mode 100644 airflow-core/newsfragments/70947.feature.rst

diff --git a/airflow-core/newsfragments/70947.feature.rst b/airflow-core/newsfragments/70947.feature.rst
new file mode 100644
index 0000000000000..1d0f8ab6affde
--- /dev/null
+++ b/airflow-core/newsfragments/70947.feature.rst
@@ -0,0 +1 @@
+Add an ``airflow teams verify`` CLI command for validating multi-team configuration. The command checks that each team has its system-managed default pool, that DAG bundles reference existing teams, and reports any configuration inconsistencies.
diff --git a/airflow-core/src/airflow/cli/cli_config.py b/airflow-core/src/airflow/cli/cli_config.py
index 46a0ebfa391f0..1a83af7a2726c 100644
--- a/airflow-core/src/airflow/cli/cli_config.py
+++ b/airflow-core/src/airflow/cli/cli_config.py
@@ -1681,6 +1681,13 @@ class GroupCommand(NamedTuple):
         func=lazy_load_command("airflow.cli.commands.team_command.team_sync"),
         args=(ARG_VERBOSE,),
     ),
+    ActionCommand(
+        name="verify",
+        help="Verify multi-team configuration",
+        description=("Verify that the multi-team configuration is internally consistent.\n"),
+        func=lazy_load_command("airflow.cli.commands.team_command.team_verify"),
+        args=(ARG_VERBOSE,),
+    ),
 )
 STATE_STORE_COMMANDS = (
     ActionCommand(
diff --git a/airflow-core/src/airflow/cli/commands/team_command.py b/airflow-core/src/airflow/cli/commands/team_command.py
index c108ac203f8bf..855a42071de74 100644
--- a/airflow-core/src/airflow/cli/commands/team_command.py
+++ b/airflow-core/src/airflow/cli/commands/team_command.py
@@ -238,3 +238,46 @@ def team_sync(args, *, session=NEW_SESSION):
 
     if teams_added > 0:
         print(f"{teams_added} teams added.")
+
+
+@cli_utils.action_cli
+@providers_configuration_loaded
+@provide_session
+def team_verify(args, *, session=NEW_SESSION):
+    """Verify that the multi-team configuration is consistent."""
+    if not conf.getboolean("core", "multi_team"):
+        print("Multi-team is not enabled.")
+        return
+
+    issues: list[str] = []
+
+    teams = session.scalars(select(Team)).all()
+
+    for team in teams:
+        default_pool_name = Pool.get_default_team_pool_name(team.name)
+
+        default_pool = session.scalar(
+            select(Pool).where(
+                Pool.pool == default_pool_name,
+                Pool.team_name == team.name,
+            )
+        )
+
+        if default_pool is None:
+            issues.append(f"Team '{team.name}' is missing default pool '{default_pool_name}'.")
+
+    existing_teams = {team.name for team in teams}
+
+    for bundle_name, bundle in DagBundlesManager()._bundle_config.items():
+        if bundle.team_name and bundle.team_name not in existing_teams:
+            issues.append(f"DAG bundle '{bundle_name}' references unknown team '{bundle.team_name}'.")
+
+    if issues:
+        print("Verification failed.\n")
+
+        for issue in issues:
+            print(f"✗ {issue}")
+
+        raise SystemExit(1)
+
+    print("Verification succeeded.")
diff --git a/airflow-core/tests/unit/cli/commands/test_team_command.py b/airflow-core/tests/unit/cli/commands/test_team_command.py
index 02d8a44516f0e..3af2e4ee5b8c9 100644
--- a/airflow-core/tests/unit/cli/commands/test_team_command.py
+++ b/airflow-core/tests/unit/cli/commands/test_team_command.py
@@ -475,3 +475,63 @@ def test_team_sync_creates_missing_default_pool(self):
 
         assert pool is not None
         assert pool.team_name == "team1"
+
+    def test_team_verify_multi_team_disabled(self, stdout_capture):
+        with conf_vars({("core", "multi_team"): "False"}):
+            with stdout_capture as stdout:
+                team_command.team_verify(self.parser.parse_args(["teams", "verify"]))
+
+        assert "Multi-team is not enabled." in stdout.getvalue()
+
+    def test_team_verify_success(self, stdout_capture):
+        bundle_config = [
+            {
+                "name": "bundleone",
+                "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+                "kwargs": {"path": "/dev/null", "refresh_interval": 0},
+                "team_name": "team1",
+            },
+        ]
+
+        with conf_vars(
+            {
+                ("core", "multi_team"): "True",
+                ("dag_processor", "dag_bundle_config_list"): json.dumps(bundle_config),
+            }
+        ):
+            team_command.team_sync(self.parser.parse_args(["teams", "sync"]))
+
+            with stdout_capture as stdout:
+                team_command.team_verify(self.parser.parse_args(["teams", "verify"]))
+
+        assert "Verification succeeded." in stdout.getvalue()
+
+    def test_team_verify_missing_default_pool(self):
+        self.session.add(Team(name="team1"))
+        self.session.commit()
+
+        with conf_vars({("core", "multi_team"): "True"}):
+            with pytest.raises(SystemExit):
+                team_command.team_verify(self.parser.parse_args(["teams", "verify"]))
+
+    def test_team_verify_unknown_bundle_team(self, stdout_capture):
+        bundle_config = [
+            {
+                "name": "bundleone",
+                "classpath": "airflow.dag_processing.bundles.local.LocalDagBundle",
+                "kwargs": {"path": "/dev/null", "refresh_interval": 0},
+                "team_name": "missing-team",
+            },
+        ]
+
+        with conf_vars(
+            {
+                ("core", "multi_team"): "True",
+                ("dag_processor", "dag_bundle_config_list"): json.dumps(bundle_config),
+            }
+        ):
+            with stdout_capture as stdout:
+                with pytest.raises(SystemExit):
+                    team_command.team_verify(self.parser.parse_args(["teams", "verify"]))
+
+        assert "references unknown team 'missing-team'" in stdout.getvalue()