diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 0df4da6386..a874a023b9 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -92,6 +92,7 @@ jobs:
     outputs:
       core-ci: ${{ steps.determine.outputs.core-ci }}
       integration-tests: ${{ steps.determine.outputs.integration-tests }}
+      integration-run-all: ${{ steps.determine.outputs.integration-run-all }}
       integration-test-buckets: ${{ steps.determine.outputs.integration-test-buckets }}
       clang-tidy: ${{ steps.determine.outputs.clang-tidy }}
       clang-tidy-mode: ${{ steps.determine.outputs.clang-tidy-mode }}
@@ -152,6 +153,9 @@ jobs:
           # Extract individual fields
           echo "core-ci=$(echo "$output" | jq -r '.core_ci')" >> $GITHUB_OUTPUT
           echo "integration-tests=$(echo "$output" | jq -r '.integration_tests')" >> $GITHUB_OUTPUT
+          # A missing key must fail here, not silently disable the junit upload
+          run_all=$(echo "$output" | jq -r 'if has("integration_run_all") then .integration_run_all else error("integration_run_all missing") end')
+          echo "integration-run-all=${run_all}" >> $GITHUB_OUTPUT
           echo "integration-test-buckets=$(echo "$output" | jq -c '.integration_test_buckets')" >> $GITHUB_OUTPUT
           echo "clang-tidy=$(echo "$output" | jq -r '.clang_tidy')" >> $GITHUB_OUTPUT
           echo "clang-tidy-mode=$(echo "$output" | jq -r '.clang_tidy_mode')" >> $GITHUB_OUTPUT
@@ -427,8 +431,25 @@ jobs:
         run: |
           . venv/bin/activate
           mapfile -t test_files < <(echo "$BUCKET_TESTS" | jq -r '.[]')
+          if [ "${#test_files[@]}" -eq 0 ]; then
+            echo "::error::Empty integration test bucket; pytest would collect the whole tree"
+            exit 1
+          fi
           echo "Bucket ${{ matrix.bucket.name }}: running ${#test_files[@]} integration tests"
-          pytest -vv --no-cov --tb=native --durations=30 -n auto "${test_files[@]}"
+          pytest -vv --no-cov --tb=native --durations=30 -n auto --dist worksteal \
+            --junitxml=junit-integration.xml "${test_files[@]}"
+      - name: Upload junit timings
+        # Consumed by sync-integration-durations.yml through
+        # script/update_integration_test_durations.py; only full matrix dev
+        # runs produce usable data.
+        if: github.ref == 'refs/heads/dev' && needs.determine-jobs.outputs.integration-run-all == 'true'
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: junit-integration-${{ strategy.job-index }}
+          path: junit-integration.xml
+          if-no-files-found: error
+          # A full cron period of margin for the weekly refresh
+          retention-days: 14
       - name: Print ccache statistics
         # esphome stores the PlatformIO ccache under the machine-global cache
         # dir (see _ccache_env() in esphome/platformio/toolchain.py).
diff --git a/.github/workflows/codeql.yml b/.github/workflows/codeql.yml
index b46f9adab6..aab3dea592 100644
--- a/.github/workflows/codeql.yml
+++ b/.github/workflows/codeql.yml
@@ -56,7 +56,7 @@ jobs:
 
       # Initializes the CodeQL tools for scanning.
       - name: Initialize CodeQL
-        uses: github/codeql-action/init@db488ddef3bf6cb639b32c2e9a7c0a7ea8271d28 # v4.37.8
+        uses: github/codeql-action/init@cdf488f595d80d6e07e03d4674febd5ab45fa938 # v4.37.9
         with:
           languages: ${{ matrix.language }}
           build-mode: ${{ matrix.build-mode }}
@@ -84,6 +84,6 @@ jobs:
           exit 1
 
       - name: Perform CodeQL Analysis
-        uses: github/codeql-action/analyze@db488ddef3bf6cb639b32c2e9a7c0a7ea8271d28 # v4.37.8
+        uses: github/codeql-action/analyze@cdf488f595d80d6e07e03d4674febd5ab45fa938 # v4.37.9
         with:
           category: "/language:${{matrix.language}}"
diff --git a/.github/workflows/sync-integration-durations.yml b/.github/workflows/sync-integration-durations.yml
new file mode 100644
index 0000000000..d09a1cf242
--- /dev/null
+++ b/.github/workflows/sync-integration-durations.yml
@@ -0,0 +1,98 @@
+---
+name: Refresh integration test durations
+
+on:
+  workflow_dispatch:
+  schedule:
+    - cron: "45 5 * * 1"
+
+# Repo writes (branch push, PR open) happen via the App token minted below,
+# so the workflow's GITHUB_TOKEN does not need any write scopes.
+permissions:
+  contents: read
+  actions: read  # gh api / gh run download for the CI junit artifacts
+
+jobs:
+  sync:
+    name: Refresh integration test durations
+    runs-on: ubuntu-latest
+    if: github.repository == 'esphome/esphome'
+    steps:
+      - name: Generate a token
+        id: generate-token
+        uses: actions/create-github-app-token@bcd2ba49218906704ab6c1aa796996da409d3eb1 # v3.2.0
+        with:
+          client-id: ${{ vars.ESPHOME_GITHUB_APP_CLIENT_ID }}
+          private-key: ${{ secrets.ESPHOME_GITHUB_APP_PRIVATE_KEY }}
+          permission-contents: write       # push the sync branch
+          permission-pull-requests: write  # open or refresh the sync PR
+
+      - name: Checkout
+        uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1
+
+      - name: Setup Python
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.13"
+
+      - name: Refresh from the newest usable dev run
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          # Only full matrix dev runs upload junit-integration-* artifacts
+          # (see the integration-tests job); the merge script re-checks
+          # coverage regardless.
+          # Newest-first candidates via their bucket-0 artifact. Fork PRs run
+          # their own ci.yml, so name and branch are spoofable; require
+          # same-repo. Assignment failures trip set -e and fail loudly.
+          candidates=$(
+            gh api "repos/${GITHUB_REPOSITORY}/actions/artifacts?name=junit-integration-0&per_page=100" \
+              --jq '.artifacts[] | select(.expired | not) | .workflow_run
+                    | select(.head_branch == "dev" and .head_repository_id != null
+                             and .head_repository_id == .repository_id)
+                    | .id'
+          )
+          # Green runs first, then the rest newest first; a run missing a
+          # bucket fails the coverage check and the next one is tried
+          green=""
+          rest=""
+          for id in ${candidates}; do
+            conclusion=$(gh api "repos/${GITHUB_REPOSITORY}/actions/runs/${id}" --jq '.conclusion // ""')
+            if [ "${conclusion}" = "success" ]; then
+              green="${green} ${id}"
+            elif [ -n "${conclusion}" ]; then
+              rest="${rest} ${id}"
+            fi
+          done
+          # helpers.py imports colorama; the script needs nothing else
+          pip install colorama
+          for id in ${green} ${rest}; do
+            rm -rf /tmp/junit
+            if ! gh run download "${id}" --repo "${GITHUB_REPOSITORY}" -p "junit-integration-*" -D /tmp/junit; then
+              echo "::warning::Could not download artifacts for run ${id}; trying the next"
+              continue
+            fi
+            status=0
+            python script/update_integration_test_durations.py /tmp/junit || status=$?
+            if [ "${status}" -eq 0 ]; then
+              echo "Refreshed from run ${id}"
+              exit 0
+            fi
+            # Only EXIT_LOW_COVERAGE (3) from the script advances to the next run
+            [ "${status}" -eq 3 ] || exit 1
+            echo "::warning::Run ${id} covers too few test files; trying the next"
+          done
+          echo "::error::No dev CI run with usable junit artifacts in range; the feed is starved"
+          exit 1
+
+      - name: Commit changes
+        uses: peter-evans/create-pull-request@5f6978faf089d4d20b00c7766989d076bb2fc7f1 # v8.1.1
+        with:
+          commit-message: "[ci] Refresh integration test durations"
+          committer: esphome[bot] <115708604+esphome[bot]@users.noreply.github.com>
+          author: esphome[bot] <115708604+esphome[bot]@users.noreply.github.com>
+          branch: sync/integration-durations
+          delete-branch: true
+          title: "[ci] Refresh integration test durations"
+          body-path: .github/PULL_REQUEST_TEMPLATE.md
+          token: ${{ steps.generate-token.outputs.token }}
diff --git a/docker/generate_tags.py b/docker/generate_tags.py
index 31f98c4614..a54205f1bf 100755
--- a/docker/generate_tags.py
+++ b/docker/generate_tags.py
@@ -1,5 +1,6 @@
 #!/usr/bin/env python3
 import argparse
+import os
 import re
 
 CHANNEL_DEV = "dev"
@@ -64,7 +65,8 @@ def main():
 
     suffix = f"-{args.suffix}" if args.suffix else ""
 
-    image_name = f"esphome/esphome{suffix}"
+    repository = (os.environ.get("GITHUB_REPOSITORY") or "esphome/esphome").strip().lower()
+    image_name = f"{repository}{suffix}"
 
     print(f"channel={channel}")
 
diff --git a/esphome/components/api/api_server.cpp b/esphome/components/api/api_server.cpp
index b5d7b96f2b..8effa29489 100644
--- a/esphome/components/api/api_server.cpp
+++ b/esphome/components/api/api_server.cpp
@@ -475,8 +475,10 @@ void APIServer::send_homeassistant_action(const HomeassistantActionRequest &call
     // Home Assistant subscribes to actions shortly *after* authenticating, so actions
     // fired right at connection time (on_client_connected, on_time_sync, ...) can
     // arrive before the subscription and are lost - warn instead of failing silently.
-    ESP_LOGW(TAG, "Home Assistant %s '%s' dropped; %s", call.is_event ? "event" : "action", call.service.c_str(),
-             this->is_connected() ? "client has not subscribed to actions (yet)" : "no client connected");
+    ESP_LOGW(TAG, "Home Assistant %s '%s' dropped; %s",
+             call.is_event ? LOG_STR_LITERAL("event") : LOG_STR_LITERAL("action"), call.service.c_str(),
+             this->is_connected() ? LOG_STR_LITERAL("client has not subscribed to actions (yet)")
+                                  : LOG_STR_LITERAL("no client connected"));
   }
 }
 #ifdef USE_API_HOMEASSISTANT_ACTION_RESPONSES
diff --git a/esphome/components/lvgl/defines.py b/esphome/components/lvgl/defines.py
index 61d15752be..1eee8041f9 100644
--- a/esphome/components/lvgl/defines.py
+++ b/esphome/components/lvgl/defines.py
@@ -628,10 +628,6 @@ OBJ_FLAGS = (
     "send_draw_task_events",
     "widget_1",
     "widget_2",
-    "user_1",
-    "user_2",
-    "user_3",
-    "user_4",
 )
 LV_OBJ_FLAG = LvConstant("LV_OBJ_FLAG_", *OBJ_FLAGS)
 
diff --git a/esphome/components/sen6x/sen6x.cpp b/esphome/components/sen6x/sen6x.cpp
index 2a6ea64735..ed6cb24c52 100644
--- a/esphome/components/sen6x/sen6x.cpp
+++ b/esphome/components/sen6x/sen6x.cpp
@@ -9,9 +9,11 @@ static const char *const TAG = "sen6x";
 
 static constexpr uint8_t POLL_RETRIES = 24;     // 24 attempts
 static constexpr uint32_t I2C_READ_DELAY = 20;  // 20 ms to wait for I2C read to complete
+static constexpr uint32_t CMD_EXEC_DELAY = 20;  // execution time of set commands (datasheet section 4.8)
 static constexpr uint32_t POLL_INTERVAL = 50;   // 50 ms between poll attempts
-// Single numeric timeout ID — the chain is sequential so only one is active at a time.
+// Numeric timeout IDs. Each chain is sequential, so only one timeout per ID is active at a time.
 static constexpr uint32_t TIMEOUT_POLL = 1;
+static constexpr uint32_t TIMEOUT_SETUP_STEP = 2;
 static constexpr uint16_t SEN6X_CMD_GET_DATA_READY_STATUS = 0x0202;
 static constexpr uint16_t SEN6X_CMD_GET_FIRMWARE_VERSION = 0xD100;
 static constexpr uint16_t SEN6X_CMD_GET_PRODUCT_NAME = 0xD014;
@@ -26,6 +28,8 @@ static constexpr uint16_t SEN6X_CMD_READ_MEASUREMENT_SEN69C = 0x04B5;
 
 static constexpr uint16_t SEN6X_CMD_START_MEASUREMENTS = 0x0021;
 static constexpr uint16_t SEN6X_CMD_RESET = 0xD304;
+static constexpr uint16_t SEN6X_CMD_VOC_ALGORITHM_TUNING = 0x60D0;
+static constexpr uint16_t SEN6X_CMD_NOX_ALGORITHM_TUNING = 0x60E1;
 
 static inline void set_read_command_and_words(SEN6XComponent::Sen6xType type, uint16_t &read_cmd, uint8_t &read_words) {
   read_cmd = SEN6X_CMD_READ_MEASUREMENT;
@@ -143,21 +147,76 @@ void SEN6XComponent::setup() {
           this->firmware_version_minor_ = raw_firmware_version & 0xFF;
           ESP_LOGI(TAG, "Firmware: %u.%u", this->firmware_version_major_, this->firmware_version_minor_);
 
-          if (!this->write_command(SEN6X_CMD_START_MEASUREMENTS)) {
-            ESP_LOGE(TAG, ESP_LOG_MSG_COMM_FAIL);
-            this->mark_failed(LOG_STR(ESP_LOG_MSG_COMM_FAIL));
-            return;
-          }
-
-          this->set_timeout(60000, [this]() { this->startup_complete_ = true; });
-          this->initialized_ = true;
-          ESP_LOGD(TAG, "Initialized");
+          // Step 4: write configuration commands one at a time, then start measurements.
+          // Delay the first step so it doesn't run in the same loop tick as the read above.
+          this->set_timeout(TIMEOUT_SETUP_STEP, CMD_EXEC_DELAY, [this]() { this->run_next_setup_step_(); });
         });
       });
     });
   });
 }
 
+// One configuration write per invocation, spaced by CMD_EXEC_DELAY. Cases without a
+// configured value fall through; each taken case must advance setup_step_index_ so the
+// next invocation resumes at the following step. These writes are optional, so a failure
+// only warns and the chain continues to the mandatory start-measurements write.
+void SEN6XComponent::run_next_setup_step_() {
+  switch (this->setup_step_index_) {
+    // Tuning writes are skipped when setup() disabled the sensor for this variant
+    case 0:
+      this->setup_step_index_++;
+      if (this->voc_sensor_ != nullptr && this->voc_tuning_params_.has_value()) {
+        this->write_tuning_parameters_(SEN6X_CMD_VOC_ALGORITHM_TUNING, this->voc_tuning_params_.value());
+        break;
+      }
+      [[fallthrough]];
+    case 1:
+      this->setup_step_index_++;
+      if (this->nox_sensor_ != nullptr && this->nox_tuning_params_.has_value()) {
+        this->write_tuning_parameters_(SEN6X_CMD_NOX_ALGORITHM_TUNING, this->nox_tuning_params_.value());
+        break;
+      }
+      [[fallthrough]];
+    default:
+      this->finish_setup_();
+      return;
+  }
+  this->set_timeout(TIMEOUT_SETUP_STEP, CMD_EXEC_DELAY, [this]() { this->run_next_setup_step_(); });
+}
+
+void SEN6XComponent::finish_setup_() {
+  if (!this->write_command(SEN6X_CMD_START_MEASUREMENTS)) {
+    ESP_LOGE(TAG, "Write 0x%04X failed, error %d", SEN6X_CMD_START_MEASUREMENTS, this->last_error_);
+    this->mark_failed(LOG_STR(ESP_LOG_MSG_COMM_FAIL));
+    return;
+  }
+
+  this->set_timeout(60000, [this]() { this->startup_complete_ = true; });
+  this->initialized_ = true;
+  ESP_LOGD(TAG, "Initialized");
+}
+
+// Writes one optional configuration command. A failure warns and returns false, but does
+// not stop setup: the sensor still measures with that setting left at its default.
+bool SEN6XComponent::write_config_words_(uint16_t i2c_command, const uint16_t *data, uint8_t len) {
+  if (!this->write_command(i2c_command, data, len)) {
+    ESP_LOGE(TAG, "Write 0x%04X failed, error %d", i2c_command, this->last_error_);
+    this->status_set_warning();
+    return false;
+  }
+  return true;
+}
+
+bool SEN6XComponent::write_tuning_parameters_(uint16_t i2c_command, const GasTuning &tuning) {
+  uint16_t params[6] = {tuning.index_offset,
+                        tuning.learning_time_offset_hours,
+                        tuning.learning_time_gain_hours,
+                        tuning.gating_max_duration_minutes,
+                        tuning.std_initial,
+                        tuning.gain_factor};
+  return this->write_config_words_(i2c_command, params, 6);
+}
+
 void SEN6XComponent::dump_config() {
   ESP_LOGCONFIG(TAG,
                 "sen6x:\n"
diff --git a/esphome/components/sen6x/sen6x.h b/esphome/components/sen6x/sen6x.h
index 041bf3b1aa..64ce3371fc 100644
--- a/esphome/components/sen6x/sen6x.h
+++ b/esphome/components/sen6x/sen6x.h
@@ -1,11 +1,25 @@
 #pragma once
 
 #include "esphome/core/component.h"
+#include "esphome/core/optional.h"
 #include "esphome/components/sensor/sensor.h"
 #include "esphome/components/sensirion_common/i2c_sensirion.h"
 
 namespace esphome::sen6x {
 
+// The NOx algorithm requires std_initial to stay at 50 (Sensirion datasheet)
+static constexpr uint16_t NOX_STD_INITIAL = 50;
+
+// Raw parameter block for the VOC/NOx algorithm tuning commands
+struct GasTuning {
+  uint16_t index_offset;
+  uint16_t learning_time_offset_hours;
+  uint16_t learning_time_gain_hours;
+  uint16_t gating_max_duration_minutes;
+  uint16_t std_initial;
+  uint16_t gain_factor;
+};
+
 class SEN6XComponent final : public PollingComponent, public sensirion_common::SensirionI2CDevice {
   SUB_SENSOR(pm_1_0)
   SUB_SENSOR(pm_2_5)
@@ -27,22 +41,46 @@ class SEN6XComponent final : public PollingComponent, public sensirion_common::S
   enum Sen6xType { SEN62, SEN63C, SEN65, SEN66, SEN68, SEN69C, UNKNOWN };
 
   void set_type(const std::string &type) { sen6x_type_ = infer_type_from_product_name_(type); }
+  void set_voc_algorithm_tuning(uint16_t index_offset, uint16_t learning_time_offset_hours,
+                                uint16_t learning_time_gain_hours, uint16_t gating_max_duration_minutes,
+                                uint16_t std_initial, uint16_t gain_factor) {
+    this->voc_tuning_params_ = GasTuning{
+        index_offset, learning_time_offset_hours, learning_time_gain_hours, gating_max_duration_minutes, std_initial,
+        gain_factor};
+  }
+  void set_nox_algorithm_tuning(uint16_t index_offset, uint16_t learning_time_offset_hours,
+                                uint16_t learning_time_gain_hours, uint16_t gating_max_duration_minutes,
+                                uint16_t gain_factor) {
+    this->nox_tuning_params_ = GasTuning{index_offset,
+                                         learning_time_offset_hours,
+                                         learning_time_gain_hours,
+                                         gating_max_duration_minutes,
+                                         NOX_STD_INITIAL,
+                                         gain_factor};
+  }
 
  protected:
   Sen6xType infer_type_from_product_name_(const std::string &product_name);
+  void run_next_setup_step_();
+  void finish_setup_();
+  bool write_config_words_(uint16_t i2c_command, const uint16_t *data, uint8_t len);
+  bool write_tuning_parameters_(uint16_t i2c_command, const GasTuning &tuning);
   void poll_data_ready_();
   void read_measurements_();
   void parse_and_publish_measurements_();
 
-  bool initialized_{false};
   std::string product_name_;
-  Sen6xType sen6x_type_{UNKNOWN};
   std::string serial_number_;
+  optional<GasTuning> voc_tuning_params_;
+  optional<GasTuning> nox_tuning_params_;
+  Sen6xType sen6x_type_{UNKNOWN};
   uint16_t read_cmd_{0};
+  uint8_t setup_step_index_{0};
   uint8_t firmware_version_major_{0};
   uint8_t firmware_version_minor_{0};
   uint8_t poll_retries_remaining_{0};
   uint8_t read_words_{0};
+  bool initialized_{false};
   bool startup_complete_{false};
 };
 
diff --git a/esphome/components/sen6x/sensor.py b/esphome/components/sen6x/sensor.py
index b0ffdc53a4..4c0242f2e3 100644
--- a/esphome/components/sen6x/sensor.py
+++ b/esphome/components/sen6x/sensor.py
@@ -3,15 +3,22 @@ from esphome.components import i2c, sensirion_common, sensor
 from esphome.components.const import CONF_NOX_INDEX, CONF_VOC_INDEX
 import esphome.config_validation as cv
 from esphome.const import (
+    CONF_ALGORITHM_TUNING,
     CONF_CO2,
     CONF_FORMALDEHYDE,
+    CONF_GAIN_FACTOR,
+    CONF_GATING_MAX_DURATION_MINUTES,
     CONF_HUMIDITY,
     CONF_ID,
+    CONF_INDEX_OFFSET,
+    CONF_LEARNING_TIME_GAIN_HOURS,
+    CONF_LEARNING_TIME_OFFSET_HOURS,
     CONF_NOX,
     CONF_PM_1_0,
     CONF_PM_2_5,
     CONF_PM_4_0,
     CONF_PM_10_0,
+    CONF_STD_INITIAL,
     CONF_TEMPERATURE,
     CONF_TYPE,
     CONF_VOC,
@@ -44,6 +51,42 @@ SEN6XComponent = sen6x_ns.class_(
 )
 
 
+def _gas_index_schema(
+    *,
+    index_offset: int,
+    gating_max_duration: int,
+    std_initial: int | None,
+) -> cv.Schema:
+    """Sensor schema for a gas index sensor with optional algorithm tuning.
+
+    std_initial is only configurable for VOC; the NOx algorithm requires 50.
+    """
+    tuning_schema = {
+        cv.Optional(CONF_INDEX_OFFSET, default=index_offset): cv.int_range(
+            min=1, max=250
+        ),
+        cv.Optional(CONF_LEARNING_TIME_OFFSET_HOURS, default=12): cv.int_range(
+            min=1, max=1000
+        ),
+        cv.Optional(CONF_LEARNING_TIME_GAIN_HOURS, default=12): cv.int_range(
+            min=1, max=1000
+        ),
+        cv.Optional(
+            CONF_GATING_MAX_DURATION_MINUTES, default=gating_max_duration
+        ): cv.int_range(min=0, max=3000),
+        cv.Optional(CONF_GAIN_FACTOR, default=230): cv.int_range(min=1, max=1000),
+    }
+    if std_initial is not None:
+        tuning_schema[cv.Optional(CONF_STD_INITIAL, default=std_initial)] = (
+            cv.int_range(min=10, max=5000)
+        )
+    return sensor.sensor_schema(
+        icon=ICON_RADIATOR,
+        accuracy_decimals=0,
+        state_class=STATE_CLASS_MEASUREMENT,
+    ).extend({cv.Optional(CONF_ALGORITHM_TUNING): cv.Schema(tuning_schema)})
+
+
 CONFIG_SCHEMA = cv.All(
     cv.rename_key(CONF_VOC, CONF_VOC_INDEX, removed_in="2027.2.0", component="sen6x"),
     cv.rename_key(CONF_NOX, CONF_NOX_INDEX, removed_in="2027.2.0", component="sen6x"),
@@ -94,15 +137,15 @@ CONFIG_SCHEMA = cv.All(
                 device_class=DEVICE_CLASS_HUMIDITY,
                 state_class=STATE_CLASS_MEASUREMENT,
             ),
-            cv.Optional(CONF_VOC_INDEX): sensor.sensor_schema(
-                icon=ICON_RADIATOR,
-                accuracy_decimals=0,
-                state_class=STATE_CLASS_MEASUREMENT,
+            cv.Optional(CONF_VOC_INDEX): _gas_index_schema(
+                index_offset=100,
+                gating_max_duration=180,
+                std_initial=50,
             ),
-            cv.Optional(CONF_NOX_INDEX): sensor.sensor_schema(
-                icon=ICON_RADIATOR,
-                accuracy_decimals=0,
-                state_class=STATE_CLASS_MEASUREMENT,
+            cv.Optional(CONF_NOX_INDEX): _gas_index_schema(
+                index_offset=1,
+                gating_max_duration=720,
+                std_initial=None,
             ),
             cv.Optional(CONF_CO2): sensor.sensor_schema(
                 unit_of_measurement=UNIT_PARTS_PER_MILLION,
@@ -149,3 +192,20 @@ async def to_code(config: ConfigType) -> None:
         if cfg := config.get(key):
             sens = await sensor.new_sensor(cfg)
             cg.add(getattr(var, func_name)(sens))
+
+    for key, setter in (
+        (CONF_VOC_INDEX, "set_voc_algorithm_tuning"),
+        (CONF_NOX_INDEX, "set_nox_algorithm_tuning"),
+    ):
+        if (tuning := config.get(key, {}).get(CONF_ALGORITHM_TUNING)) is not None:
+            args = [
+                tuning[CONF_INDEX_OFFSET],
+                tuning[CONF_LEARNING_TIME_OFFSET_HOURS],
+                tuning[CONF_LEARNING_TIME_GAIN_HOURS],
+                tuning[CONF_GATING_MAX_DURATION_MINUTES],
+            ]
+            # std_initial is in the schema for VOC only
+            if (std_initial := tuning.get(CONF_STD_INITIAL)) is not None:
+                args.append(std_initial)
+            args.append(tuning[CONF_GAIN_FACTOR])
+            cg.add(getattr(var, setter)(*args))
diff --git a/esphome/components/whynter/whynter.cpp b/esphome/components/whynter/whynter.cpp
index b8a8db4d7c..f5f304aa17 100644
--- a/esphome/components/whynter/whynter.cpp
+++ b/esphome/components/whynter/whynter.cpp
@@ -84,8 +84,8 @@ void Whynter::transmit_state() {
 
   if (fahrenheit_) {
     remote_state |= UNIT_MASK;
-    uint8_t temp =
-        (uint8_t) clamp<float>(esphome::celsius_to_fahrenheit(this->target_temperature), TEMP_MIN_F, TEMP_MAX_F);
+    uint8_t temp = (uint8_t) roundf(
+        clamp<float>(esphome::celsius_to_fahrenheit(this->target_temperature), TEMP_MIN_F, TEMP_MAX_F));
     temp = esphome::reverse_bits(temp);
     remote_state |= temp;
   } else {
diff --git a/esphome/core/scheduler.cpp b/esphome/core/scheduler.cpp
index e9c5bf2c04..afb323f78e 100644
--- a/esphome/core/scheduler.cpp
+++ b/esphome/core/scheduler.cpp
@@ -193,7 +193,7 @@ void HOT Scheduler::set_timer_common_(Component *component, SchedulerItem::Type
     }
 
 #ifdef ESPHOME_DEBUG_SCHEDULER
-    this->debug_log_timer_(item, name_type, static_name, hash_or_id, type, delay, now_64);
+    this->debug_log_timer_(item, name_type, static_name, hash_or_id, delay, now_64);
 #endif /* ESPHOME_DEBUG_SCHEDULER */
   }
 
@@ -438,9 +438,10 @@ uint32_t HOT Scheduler::call(uint32_t now) {
       SchedulerNameLog name_log;
       bool is_cancelled = is_item_removed_(item);
       ESP_LOGD(TAG, "  %s '%s/%s' interval=%" PRIu32 " next_execution in %" PRIu64 "ms at %" PRIu64 "%s",
-               item->get_type_str(), LOG_STR_ARG(item->get_source()),
+               LOG_STR_ARG(item->get_type_str()), LOG_STR_ARG(item->get_source()),
                name_log.format(item->get_name_type(), item->get_name(), item->get_name_hash_or_id()), item->interval,
-               item->get_next_execution() - now_64, item->get_next_execution(), is_cancelled ? " [CANCELLED]" : "");
+               item->get_next_execution() - now_64, item->get_next_execution(),
+               is_cancelled ? LOG_STR_LITERAL(" [CANCELLED]") : LOG_STR_LITERAL(""));
 
       old_items.push_back(item);
     }
@@ -512,7 +513,7 @@ uint32_t HOT Scheduler::call(uint32_t now) {
     {
       SchedulerNameLog name_log;
       ESP_LOGV(TAG, "Running %s '%s/%s' with interval=%" PRIu32 " next_execution=%" PRIu64 " (now=%" PRIu64 ")",
-               item->get_type_str(), LOG_STR_ARG(item->get_source()),
+               LOG_STR_ARG(item->get_type_str()), LOG_STR_ARG(item->get_source()),
                name_log.format(item->get_name_type(), item->get_name(), item->get_name_hash_or_id()), item->interval,
                item->get_next_execution(), now_64);
     }
@@ -794,7 +795,7 @@ void Scheduler::trim_freelist() {
 
 #ifdef ESPHOME_DEBUG_SCHEDULER
 void Scheduler::debug_log_timer_(const SchedulerItem *item, NameType name_type, const char *static_name,
-                                 uint32_t hash_or_id, SchedulerItem::Type type, uint32_t delay, uint64_t now) {
+                                 uint32_t hash_or_id, uint32_t delay, uint64_t now) {
   // Validate static strings in debug mode
   if (name_type == NameType::STATIC_STRING && static_name != nullptr) {
     validate_static_string(static_name);
@@ -802,8 +803,8 @@ void Scheduler::debug_log_timer_(const SchedulerItem *item, NameType name_type,
 
   // Debug logging
   SchedulerNameLog name_log;
-  const char *type_str = (type == SchedulerItem::TIMEOUT) ? "timeout" : "interval";
-  if (type == SchedulerItem::TIMEOUT) {
+  const char *type_str = LOG_STR_ARG(item->get_type_str());
+  if (item->type == SchedulerItem::TIMEOUT) {
     ESP_LOGD(TAG, "set_%s(name='%s/%s', %s=%" PRIu32 ")", type_str, LOG_STR_ARG(item->get_source()),
              name_log.format(name_type, static_name, hash_or_id), type_str, delay);
   } else {
diff --git a/esphome/core/scheduler.h b/esphome/core/scheduler.h
index 8ef3499a11..56fc83f12f 100644
--- a/esphome/core/scheduler.h
+++ b/esphome/core/scheduler.h
@@ -254,7 +254,7 @@ class Scheduler {
       // This is correct because millis_major_ that creates these values is also 16 bits.
       next_execution_high_ = static_cast<uint16_t>(value >> 32);
     }
-    constexpr const char *get_type_str() const { return (type == TIMEOUT) ? "timeout" : "interval"; }
+    const LogString *get_type_str() const { return (type == TIMEOUT) ? LOG_STR("timeout") : LOG_STR("interval"); }
     // The owning component, or nullptr for SELF_POINTER items (whose slot holds source_name instead).
     // All component access goes through this so SELF_POINTER items read as component-less.
     Component *get_component() const { return name_type_ == NameType::SELF_POINTER ? nullptr : component; }
@@ -404,7 +404,7 @@ class Scheduler {
 #ifdef ESPHOME_DEBUG_SCHEDULER
   // Helper for debug logging in set_timer_common_ - extracted to reduce code size
   void debug_log_timer_(const SchedulerItem *item, NameType name_type, const char *static_name, uint32_t hash_or_id,
-                        SchedulerItem::Type type, uint32_t delay, uint64_t now);
+                        uint32_t delay, uint64_t now);
 #endif /* ESPHOME_DEBUG_SCHEDULER */
 
 #ifndef ESPHOME_THREAD_SINGLE
diff --git a/esphome/platformio/prefetch.py b/esphome/platformio/prefetch.py
index ef8c27c9aa..1df0a4b328 100644
--- a/esphome/platformio/prefetch.py
+++ b/esphome/platformio/prefetch.py
@@ -16,8 +16,9 @@ name and promote with an atomic rename.
 
 from __future__ import annotations
 
+from collections.abc import Iterator
 from concurrent.futures import ThreadPoolExecutor
-from contextlib import suppress
+from contextlib import contextmanager, suppress
 import hashlib
 import json
 import logging
@@ -43,6 +44,17 @@ from esphome.helpers import get_bool_env, get_usable_cpu_count, rmtree
 
 _LOGGER = logging.getLogger(__name__)
 
+
+@contextmanager
+def _preserved_sys_path() -> Iterator[None]:
+    """Platform setup may rewrite sys.path (pioarduino's penv does); undo it."""
+    saved = list(sys.path)
+    try:
+        yield
+    finally:
+        sys.path[:] = saved
+
+
 # Concurrent registry resolutions / HEAD probes (each is network-bound)
 _RESOLVE_WORKERS = 8
 
@@ -96,6 +108,14 @@ class _Resolved(NamedTuple):
     cached: bool
 
 
+class _Group(NamedTuple):
+    """The installable ``(name, spec)`` entries of one package manager."""
+
+    manager: Any
+    entries: list[tuple[str, Any]]
+    is_platform: bool
+
+
 # Child records a no-work run; the parent skips the next spawn while valid
 _SENTINEL_NAME = ".esphome_prefetch.json"
 _SENTINEL_SCHEMA = 1
@@ -772,13 +792,9 @@ def _preinstall(
         # poison the next wave; pio run installs the rest cleanly
         _LOGGER.warning("Skipping the dependency wave")
         return
-    # The builtin probe may construct platforms whose setup rewrites
-    # sys.path (see _prefetch); restore it for later imports
-    saved_sys_path = list(sys.path)
-    try:
+    # The builtin probe may construct platforms
+    with _preserved_sys_path():
         next_entries = _dependency_entries(manager, installed, seen)
-    finally:
-        sys.path[:] = saved_sys_path
     if next_entries:
         # Terminates without a cap: every wave admits only never-seen
         # names, so a cycle yields an empty next wave
@@ -803,15 +819,13 @@ def _prefetch(build_dir: Path, env: str) -> None:
         return
 
     # The platform (manifest plus build scripts) installs first and
-    # resolves the rest. Its setup may rewrite sys.path (pioarduino's penv
-    # setup does); restore it so later imports here still resolve.
-    saved_sys_path = list(sys.path)
-    pm = PlatformPackageManager()
-    _sweep_stale_sidecars(Path(pm.get_download_dir()), pm.DOWNLOAD_CACHE_EXPIRE)
-    pkg = pm.install(platform_spec, skip_dependencies=True)
-    p = PlatformFactory.new(pkg)
-    p.configure_project_packages(env, ["run"])
-    sys.path[:] = saved_sys_path
+    # resolves the rest
+    with _preserved_sys_path():
+        pm = PlatformPackageManager()
+        _sweep_stale_sidecars(Path(pm.get_download_dir()), pm.DOWNLOAD_CACHE_EXPIRE)
+        pkg = pm.install(platform_spec, skip_dependencies=True)
+        p = PlatformFactory.new(pkg)
+        p.configure_project_packages(env, ["run"])
 
     specs = [
         p.get_package_spec(name)
@@ -851,9 +865,9 @@ def _prefetch(build_dir: Path, env: str) -> None:
 
     seen: set[str] = set()
     jobs: list[tuple[str, int, Any]] = []
-    groups: list[tuple[Any, list[tuple[str, Any]]]] = []
+    groups: list[_Group] = []
     unresolved = 0
-    for mgr, batch in ((p.pm, specs), (lm, lib_specs)):
+    for mgr, batch, is_platform in ((p.pm, specs, True), (lm, lib_specs, False)):
         entries: list[tuple[str, Any]] = []
         for build_jobs in (_registry_jobs, _uri_jobs):
             batch_jobs, failed, installable = build_jobs(mgr, batch, seen)
@@ -861,7 +875,7 @@ def _prefetch(build_dir: Path, env: str) -> None:
             unresolved += failed
             entries += installable
         if entries:
-            groups.append((mgr, entries))
+            groups.append(_Group(mgr, entries, is_platform))
 
     sentinel = build_dir / _SENTINEL_NAME
     if jobs or groups:
@@ -890,7 +904,8 @@ def _prefetch(build_dir: Path, env: str) -> None:
             encoding="utf-8",
         )
 
-    for mgr, entries in groups:
+    platform_packages_installed = False
+    for mgr, entries, is_platform in groups:
         # One install per destination: pio derives the directory from
         # the package name, so key on the name part
         to_install = {
@@ -901,6 +916,8 @@ def _prefetch(build_dir: Path, env: str) -> None:
         if to_install:
             try:
                 _preinstall(mgr, list(to_install.values()))
+                if is_platform:
+                    platform_packages_installed = True
             except Exception as err:  # noqa: BLE001  # pylint: disable=broad-exception-caught
                 # Each group degrades independently; pio run installs
                 # whatever this one did not
@@ -910,6 +927,17 @@ def _prefetch(build_dir: Path, env: str) -> None:
                     failure_reason(err),
                 )
                 _LOGGER.debug("Pre-install group failure detail", exc_info=True)
+    if platform_packages_installed:
+        # pioarduino installs its real toolchains from configure (the registry
+        # package is a stub); settle that here so pio run does not redo it
+        with _preserved_sys_path(), ThreadPoolExecutor(max_workers=1) as ex:
+            # A worker so SIGTERM joins it; exception() so a postinstall exit only warns
+            err = ex.submit(p.configure_project_packages, env, ["run"]).exception()
+        if err is not None:
+            _LOGGER.warning(
+                "Could not settle platform packages: %s", failure_reason(err)
+            )
+            _LOGGER.debug("Platform settle failure detail", exc_info=err)
 
 
 def _sigterm(_signum, _frame) -> None:
diff --git a/requirements.txt b/requirements.txt
index a065492dfa..f19559dca8 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,6 +1,6 @@
 # cryptography 49+ ships no Intel macOS wheels (arm64 only); esptool caps <49 there.
 # Keep 48.0.1, the last universal2 release, so esphome stays installable on Intel Macs.
-cryptography==50.0.0; platform_system != "Darwin" or platform_machine != "x86_64"
+cryptography==50.0.1; platform_system != "Darwin" or platform_machine != "x86_64"
 cryptography==48.0.1; platform_system == "Darwin" and platform_machine == "x86_64"
 voluptuous==0.16.0
 PyYAML==6.0.3
@@ -14,7 +14,7 @@ esptool==5.3.1
 click==8.3.3
 aioesphomeapi==46.3.0
 aiohappyeyeballs==2.7.1  # Happy Eyeballs for requests downloads; already pulled in by aioesphomeapi
-zeroconf==0.150.0
+zeroconf==0.150.4
 puremagic==2.2.0
 ruamel.yaml==0.19.1 # dashboard_import
 ruamel.yaml.clib==0.2.15 # dashboard_import
@@ -27,7 +27,7 @@ bleak==3.0.2
 smpclient==7.2.0
 requests==2.34.2
 py7zr==1.1.3
-platformdirs==4.11.4  # native esp-idf toolchain global cache dir
+platformdirs==4.11.5  # native esp-idf toolchain global cache dir
 ninja==1.13.0  # native esp8266 arduino toolchain build driver
 filelock==3.32.4  # inter-process locks (PlatformIO cache heal, git clone cache); >=3.32 for FileLock(fallback_to_soft=...), older versions silently drop the kwarg
 
diff --git a/requirements_test.txt b/requirements_test.txt
index cf4b028b0e..df37a10cb4 100644
--- a/requirements_test.txt
+++ b/requirements_test.txt
@@ -1,6 +1,6 @@
 pylint==4.0.7
 flake8==7.3.0  # also change in .pre-commit-config.yaml when updating
-ruff==0.16.4  # also change in .pre-commit-config.yaml when updating
+ruff==0.16.5  # also change in .pre-commit-config.yaml when updating
 pyupgrade==3.21.2  # also change in .pre-commit-config.yaml when updating
 prek==0.4.14  # also change in .github/workflows/ci.yml when updating
 
diff --git a/script/determine-jobs.py b/script/determine-jobs.py
index add1af5bba..f5412af21d 100755
--- a/script/determine-jobs.py
+++ b/script/determine-jobs.py
@@ -53,8 +53,10 @@ from collections import Counter
 from enum import StrEnum
 from functools import cache
 import json
+import math
 import os
 from pathlib import Path
+import statistics
 import sys
 from typing import Any
 
@@ -67,7 +69,9 @@ from clang_tidy_hash import (
 from helpers import (
     CPP_FILE_EXTENSIONS,
     ESPHOME_TESTS_COMPONENTS_PATH,
+    INTEGRATION_TESTS_PATH,
     PYTHON_FILE_EXTENSIONS,
+    all_integration_test_files,
     base_python_changed,
     changed_files,
     core_changed,
@@ -83,6 +87,8 @@ from helpers import (
     get_target_branch,
     git_ls_files,
     is_validate_only_file,
+    load_integration_durations,
+    lpt_partition,
     root_path,
 )
 from split_components_for_ci import create_intelligent_batches
@@ -96,10 +102,13 @@ CLANG_TIDY_SPLIT_THRESHOLD = 65
 # Isolated components count as 10x, groupable components count as 1x
 COMPONENT_TEST_BATCH_SIZE = 40
 
-# Integration test bucketing: when more than the threshold tests are scheduled,
-# fan out across this many parallel jobs. Below the threshold, a single job runs.
+# Above the threshold, fan out across up to this many jobs, balanced by the
+# recorded per-file durations. The target is serial junit-time weight per
+# bucket, not wall time (calibrated with the conftest compile cap); it
+# sizes the bucket count for small subsets.
 INTEGRATION_TESTS_SPLIT_THRESHOLD = 10
-INTEGRATION_TESTS_SPLIT_BUCKETS = 3
+INTEGRATION_TESTS_SPLIT_BUCKETS = 5
+INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT = 360.0
 
 # platformio and aioesphomeapi (requirements.txt), the pytest stack
 # (requirements_test.txt) and the fixture every session compiles; a change
@@ -113,27 +122,13 @@ INTEGRATION_TESTS_TRIGGER_FILES = frozenset(
 )
 
 
-def _split_list(items: list[str], n: int) -> list[list[str]]:
-    """Split a list into n roughly-equal contiguous parts (matches script/clang-tidy)."""
-    k, m = divmod(len(items), n)
-    return [items[i * k + min(i, m) : (i + 1) * k + min(i + 1, m)] for i in range(n)]
-
-
-def _all_integration_test_files() -> list[str]:
-    """Return all integration test file paths, sorted, relative to repo root."""
-    return sorted(
-        str(p.relative_to(root_path))
-        for p in (Path(root_path) / "tests" / "integration").glob("test_*.py")
-    )
-
-
 def _compute_integration_test_buckets(
     integration_run_all: bool,
     integration_test_files: list[str],
 ) -> tuple[bool, list[dict[str, Any]]]:
     """Compute (run_integration, buckets) from the determine_integration_tests result.
 
-    Pure function for unit testing — no I/O beyond `_all_integration_test_files`
+    Pure function for unit testing — no I/O beyond `all_integration_test_files`
     when `integration_run_all` is set.
 
     `buckets` is a list of `{name, tests}` dicts where `tests` is a JSON-friendly
@@ -141,7 +136,7 @@ def _compute_integration_test_buckets(
     shell word-splitting / glob hazards.
     """
     if integration_run_all:
-        files = _all_integration_test_files()
+        files = all_integration_test_files()
     else:
         files = sorted(integration_test_files)
 
@@ -152,12 +147,23 @@ def _compute_integration_test_buckets(
         return False, []
 
     if len(files) > INTEGRATION_TESTS_SPLIT_THRESHOLD:
-        parts = [
-            part for part in _split_list(files, INTEGRATION_TESTS_SPLIT_BUCKETS) if part
-        ]
+        durations = load_integration_durations()
+        # Unrecorded files weigh the recording's median; with no recording a
+        # file weighs a whole bucket, which keeps the full fan-out
+        default = (
+            statistics.median(durations.values())
+            if durations
+            else INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT
+        )
+        weights = {f: durations.get(f, default) for f in files}
+        count = min(
+            INTEGRATION_TESTS_SPLIT_BUCKETS,
+            math.ceil(sum(weights.values()) / INTEGRATION_TESTS_TARGET_BUCKET_WEIGHT),
+        )
+        # count <= SPLIT_BUCKETS < threshold < len(files): no group is empty
+        parts = [sorted(part) for part in lpt_partition(files, weights, count)]
         buckets = [
-            {"name": f"{i + 1}/{len(parts)}", "tests": part}
-            for i, part in enumerate(parts)
+            {"name": f"{i + 1}/{count}", "tests": part} for i, part in enumerate(parts)
         ]
     else:
         buckets = [{"name": "1/1", "tests": files}]
@@ -264,9 +270,9 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
     # If infrastructure Python files changed (conftest, utils, etc.), run all tests
     # Excludes test files (test_*.py), fixtures, and non-Python files (README.md)
     if any(
-        f.startswith("tests/integration/")
+        f.startswith(INTEGRATION_TESTS_PATH)
         and f.endswith(".py")
-        and not f.startswith("tests/integration/test_")
+        and not f.startswith(f"{INTEGRATION_TESTS_PATH}test_")
         and "/fixtures/" not in f
         for f in files
     ):
@@ -277,9 +283,9 @@ def determine_integration_tests(branch: str | None = None) -> tuple[bool, list[s
     fixture_to_test_files = get_fixture_to_test_files()
 
     for f in files:
-        if f.startswith("tests/integration/test_") and f.endswith(".py"):
+        if f.startswith(f"{INTEGRATION_TESTS_PATH}test_") and f.endswith(".py"):
             test_files.add(f)
-        elif f.startswith("tests/integration/fixtures/"):
+        elif f.startswith(f"{INTEGRATION_TESTS_PATH}fixtures/"):
             if f.endswith(".yaml"):
                 # Fixture YAML changed - add corresponding test file(s)
                 test_files.update(fixture_to_test_files.get(Path(f).stem, ()))
@@ -1415,6 +1421,7 @@ def main() -> None:
     output: dict[str, Any] = {
         "core_ci": run_core_ci,
         "integration_tests": run_integration,
+        "integration_run_all": integration_run_all,
         "integration_test_buckets": integration_test_buckets,
         "clang_tidy": run_clang_tidy,
         "clang_tidy_mode": clang_tidy_mode,
diff --git a/script/helpers.py b/script/helpers.py
index e648bb91bb..bf22e15808 100644
--- a/script/helpers.py
+++ b/script/helpers.py
@@ -43,6 +43,53 @@ ESPHOME_TESTS_COMPONENTS_PATH = "tests/components/"
 # Tuple of component and test paths for efficient startswith checks
 COMPONENT_AND_TESTS_PATHS = (ESPHOME_COMPONENTS_PATH, ESPHOME_TESTS_COMPONENTS_PATH)
 
+# Integration tests path prefix
+INTEGRATION_TESTS_PATH = "tests/integration/"
+
+# Per-file integration test durations from CI junit output; shared by the
+# reader (determine-jobs) and writer (update_integration_test_durations)
+INTEGRATION_TEST_DURATIONS_FILE = "tests/integration/integration_test_durations.json"
+
+
+def all_integration_test_files() -> list[str]:
+    """Return all integration test file paths, sorted, relative to repo root."""
+    return sorted(
+        p.relative_to(root_path).as_posix()
+        for p in (Path(root_path) / "tests" / "integration").glob("test_*.py")
+    )
+
+
+def load_integration_durations() -> dict[str, float]:
+    """Return recorded per-file pytest durations in seconds; empty when unavailable."""
+    try:
+        raw = json.loads(
+            (Path(root_path) / INTEGRATION_TEST_DURATIONS_FILE).read_text()
+        )
+        if not isinstance(raw, dict):
+            print(
+                f"integration durations unavailable: expected an object, "
+                f"got {type(raw).__name__}",
+                file=sys.stderr,
+            )
+            return {}
+    except (OSError, ValueError) as err:
+        # The file ships in the repo; degrade to unweighted bucketing, loudly
+        print(f"integration durations unavailable: {err}", file=sys.stderr)
+        return {}
+    durations = {
+        key: seconds
+        for key, value in raw.items()
+        if isinstance(value, (int, float)) and (seconds := float(value)) > 0
+    }
+    if len(durations) != len(raw):
+        # One bad entry must not discard the whole recording
+        print(
+            f"dropped {len(raw) - len(durations)} invalid duration entries",
+            file=sys.stderr,
+        )
+    return durations
+
+
 # Base bus components - these ARE the bus implementations and should not
 # be flagged as needing migration since they are the platform/base components
 BASE_BUS_COMPONENTS = {
@@ -1545,3 +1592,21 @@ def get_cpp_changed_components(files: list[str]) -> list[str]:
         if file.startswith(ESPHOME_COMPONENTS_PATH):
             affected.update(find_children_of_component(components_graph, component))
     return sorted(c for c in affected if has_cpp_unit_tests(c, tests_dir))
+
+
+def lpt_partition(
+    items: list[str], weights: dict[str, float], count: int
+) -> list[list[str]]:
+    """Partition items into `count` weight-balanced groups (LPT greedy).
+
+    Heaviest item first into the lightest group. Ties keep input order, so
+    pass pre-sorted items for deterministic output. script/clang-tidy's
+    split_list is the unweighted contiguous sibling.
+    """
+    groups: list[list[str]] = [[] for _ in range(count)]
+    group_weights = [0.0] * count
+    for item in sorted(items, key=lambda i: -weights[i]):
+        lightest = min(range(count), key=group_weights.__getitem__)
+        groups[lightest].append(item)
+        group_weights[lightest] += weights[item]
+    return groups
diff --git a/script/update_integration_test_durations.py b/script/update_integration_test_durations.py
new file mode 100755
index 0000000000..bbb959c0b2
--- /dev/null
+++ b/script/update_integration_test_durations.py
@@ -0,0 +1,119 @@
+#!/usr/bin/env python3
+"""Merge CI junit output into tests/integration/integration_test_durations.json.
+
+The integration-tests CI job uploads one junit XML artifact per bucket on
+full matrix dev runs. Download a run's artifacts and merge the per file
+durations into the recording used by script/determine-jobs.py:
+
+    gh run download <run-id> --repo esphome/esphome -p "junit-integration-*" -D /tmp/junit
+    script/update_integration_test_durations.py /tmp/junit
+
+Missing files keep their previous recording and deleted files drop out; a
+run covering under 90% of the test files aborts unless --allow-partial.
+"""
+
+from __future__ import annotations
+
+import argparse
+from collections import defaultdict
+import json
+from pathlib import Path
+import sys
+import xml.etree.ElementTree as ET
+
+from helpers import (
+    INTEGRATION_TEST_DURATIONS_FILE,
+    INTEGRATION_TESTS_PATH,
+    all_integration_test_files,
+    load_integration_durations,
+    root_path,
+)
+
+DURATIONS_FILE = Path(root_path) / INTEGRATION_TEST_DURATIONS_FILE
+MIN_COVERAGE = 0.9
+# Exit code for the expected "run covers too few files" refusal, so the
+# refresh workflow can move on to the next candidate run
+EXIT_LOW_COVERAGE = 3
+
+
+def collect_durations(junit_dir: Path, known_files: set[str]) -> dict[str, float]:
+    """Sum junit testcase times per integration test file, in seconds."""
+    durations: defaultdict[str, float] = defaultdict(float)
+    unmatched = 0
+    xml_files = sorted(junit_dir.rglob("*.xml"))
+    if not xml_files:
+        raise SystemExit(f"no junit XML files found under {junit_dir}")
+    for xml_file in xml_files:
+        for testcase in ET.parse(xml_file).getroot().iter("testcase"):
+            # Skipped/errored testcases carry time="0"; recording them would
+            # overwrite a good previous duration
+            if any(
+                testcase.find(tag) is not None
+                for tag in ("skipped", "error", "failure")
+            ):
+                continue
+            # classname is the dotted module plus any test class, e.g.
+            # tests.integration.test_x or tests.integration.test_x.TestFoo
+            parts = testcase.get("classname", "").split(".")
+            if parts[:2] != ["tests", "integration"] or len(parts) < 3:
+                unmatched += 1
+                continue
+            path = f"{INTEGRATION_TESTS_PATH}{parts[2]}.py"
+            if path not in known_files:
+                print(f"skipping unknown test module {path}", file=sys.stderr)
+                continue
+            durations[path] += float(testcase.get("time", "0"))
+    if unmatched:
+        # A junit naming change would otherwise shrink the recording silently
+        raise SystemExit(
+            f"{unmatched} testcases with unexpected classnames; the junit layout changed"
+        )
+    # An all-skipped file totals 0.0; let the merge keep its previous entry
+    return {k: v for k, v in durations.items() if v > 0}
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "junit_dir", type=Path, help="directory containing downloaded junit XML files"
+    )
+    parser.add_argument(
+        "--allow-partial",
+        action="store_true",
+        help="merge a run covering under 90%% of the test files",
+    )
+    args = parser.parse_args()
+
+    on_disk = set(all_integration_test_files())
+    if not on_disk:
+        raise SystemExit("no integration test files found; wrong checkout root?")
+    collected = collect_durations(args.junit_dir, on_disk)
+    coverage = len(collected.keys() & on_disk) / len(on_disk)
+    if coverage < MIN_COVERAGE and not args.allow_partial:
+        print(
+            f"artifacts cover only {coverage:.0%} of {len(on_disk)} test files; "
+            "use a full matrix run or pass --allow-partial to merge anyway",
+            file=sys.stderr,
+        )
+        return EXIT_LOW_COVERAGE
+
+    # Validated load: a bad previous entry cannot survive the round trip, and
+    # an unreadable file aborts rather than being overwritten
+    previous = load_integration_durations()
+    if DURATIONS_FILE.is_file() and not previous:
+        raise SystemExit(f"{DURATIONS_FILE} is unreadable; refusing to overwrite it")
+    # New recordings win, absent files keep theirs, deleted files drop out
+    merged = {
+        path: collected.get(path, previous.get(path))
+        for path in sorted(on_disk)
+        if path in collected or path in previous
+    }
+    DURATIONS_FILE.write_text(
+        json.dumps({k: round(v, 2) for k, v in merged.items()}, indent=2) + "\n"
+    )
+    print(f"wrote {len(merged)} entries to {DURATIONS_FILE} ({coverage:.0%} fresh)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/tests/components/lvgl/lvgl-package.yaml b/tests/components/lvgl/lvgl-package.yaml
index b457ec2c0b..07c492db35 100644
--- a/tests/components/lvgl/lvgl-package.yaml
+++ b/tests/components/lvgl/lvgl-package.yaml
@@ -188,6 +188,8 @@ lvgl:
     dark_mode: true
     obj:
       border_width: 1
+      user_1:
+        bg_color: black
 
   gradients:
     - id: color_bar
@@ -717,6 +719,30 @@ lvgl:
                 id: button_with_text
                 text: Clicked
 
+        # Exercises the LV_STATE_USER_1..USER_4 states: setting them at creation
+        # (both literal and lambda), styling each of them individually, and
+        # setting/clearing them at runtime with lvgl.widget.update.
+        - button:
+            id: user_flags_button
+            text: User flags
+            state:
+              user_1: true
+              user_2: !lambda return true;
+            user_1:
+              bg_color: 0xFF00FF
+            user_2:
+              bg_color: 0x00FFFF
+            user_3:
+              bg_color: 0xFFFF00
+            user_4:
+              bg_color: 0x808080
+            on_click:
+              - lvgl.widget.update:
+                  id: user_flags_button
+                  state:
+                    user_3: true
+                    user_4: !lambda return !lv_obj_has_state(id(user_flags_button), LV_STATE_USER_4);
+
         - button:
             layout: 2x1
             id: button_button
diff --git a/tests/components/sen6x/common.yaml b/tests/components/sen6x/common.yaml
index 859e012c4a..c9b6f22c0f 100644
--- a/tests/components/sen6x/common.yaml
+++ b/tests/components/sen6x/common.yaml
@@ -28,8 +28,21 @@ sensor:
       accuracy_decimals: 1
     nox_index:
       name: NOx Index
+      algorithm_tuning:
+        index_offset: 8
+        learning_time_offset_hours: 6
+        learning_time_gain_hours: 24
+        gating_max_duration_minutes: 900
+        gain_factor: 180
     voc_index:
       name: VOC Index
+      algorithm_tuning:
+        index_offset: 120
+        learning_time_offset_hours: 6
+        learning_time_gain_hours: 24
+        gating_max_duration_minutes: 240
+        std_initial: 75
+        gain_factor: 180
     co2:
       name: Carbon Dioxide
     formaldehyde:
diff --git a/tests/components/sen6x/validate.esp32-idf.yaml b/tests/components/sen6x/validate.esp32-idf.yaml
new file mode 100644
index 0000000000..3ae23af4ac
--- /dev/null
+++ b/tests/components/sen6x/validate.esp32-idf.yaml
@@ -0,0 +1,18 @@
+# Config-only: partial algorithm_tuning blocks, so the schema defaults fill in the
+# keys that are left out.
+packages:
+  i2c: !include ../../test_build_components/common/i2c/esp32-idf.yaml
+
+sensor:
+  - platform: sen6x
+    id: sen6x_partial_tuning
+    type: SEN65
+    i2c_id: i2c_bus
+    voc_index:
+      name: VOC Index
+      algorithm_tuning:
+        index_offset: 60
+    nox_index:
+      name: NOx Index
+      algorithm_tuning:
+        gain_factor: 45
diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py
index 12b1407fe1..6777e6cabc 100644
--- a/tests/integration/conftest.py
+++ b/tests/integration/conftest.py
@@ -23,6 +23,7 @@ import pytest_asyncio
 
 import esphome.config
 from esphome.core import CORE
+from esphome.helpers import get_usable_cpu_count
 from esphome.platformio.toolchain import get_idedata
 
 from .const import (
@@ -67,6 +68,14 @@ def _get_platformio_env(cache_dir: Path) -> dict[str, str]:
     env["PLATFORMIO_LIBDEPS_DIR"] = str(cache_dir / "libdeps" / worker)
     # Prevent cache cleaning during integration tests
     env["ESPHOME_SKIP_CLEAN_BUILD"] = "1"
+    # Cap each compile's -j so several xdist workers do not each spawn a
+    # full-width compiler fan-out on the same machine. An explicit env wins.
+    if "ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT" not in os.environ:
+        workers = int(os.environ.get("PYTEST_XDIST_WORKER_COUNT", "1"))
+        # Floor of 2 keeps a lone tail compile from running fully serial
+        env["ESPHOME_DEFAULT_COMPILE_PROCESS_LIMIT"] = str(
+            max(2, get_usable_cpu_count() // workers)
+        )
     # Compile with THIS tree's esphome sources, not wherever the venv's editable
     # install points (which may be a different git worktree or checkout).
     repo_root = str(Path(__file__).resolve().parent.parent.parent)
diff --git a/tests/integration/integration_test_durations.json b/tests/integration/integration_test_durations.json
new file mode 100644
index 0000000000..9bada5cd36
--- /dev/null
+++ b/tests/integration/integration_test_durations.json
@@ -0,0 +1,142 @@
+{
+  "tests/integration/test_action_concurrent_reentry.py": 45.23,
+  "tests/integration/test_addressable_light_transition.py": 74.47,
+  "tests/integration/test_alarm_control_panel_state_transitions.py": 74.1,
+  "tests/integration/test_api_action_metadata.py": 62.1,
+  "tests/integration/test_api_action_responses.py": 71.08,
+  "tests/integration/test_api_action_timeout.py": 21.64,
+  "tests/integration/test_api_conditional_memory.py": 13.72,
+  "tests/integration/test_api_custom_services.py": 24.16,
+  "tests/integration/test_api_get_time_response_timezone.py": 23.48,
+  "tests/integration/test_api_homeassistant.py": 37.87,
+  "tests/integration/test_api_homeassistant_action_no_subscriber.py": 14.38,
+  "tests/integration/test_api_list_entities_backpressure.py": 26.85,
+  "tests/integration/test_api_message_size_batching.py": 33.36,
+  "tests/integration/test_api_reboot_timeout.py": 13.63,
+  "tests/integration/test_api_string_lambda.py": 25.04,
+  "tests/integration/test_api_vv_logging.py": 16.6,
+  "tests/integration/test_api_zero_psk_provisioning.py": 43.14,
+  "tests/integration/test_areas_and_devices.py": 25.98,
+  "tests/integration/test_automation_wait_actions.py": 21.91,
+  "tests/integration/test_automations.py": 42.43,
+  "tests/integration/test_batch_delay_zero_rapid_transitions.py": 16.65,
+  "tests/integration/test_binary_sensor_autorepeat_filter.py": 28.67,
+  "tests/integration/test_binary_sensor_invalidate_state.py": 23.69,
+  "tests/integration/test_blocking_warning_log_time_not_charged_to_next_operation.py": 22.99,
+  "tests/integration/test_build_info.py": 24.96,
+  "tests/integration/test_camera_mock.py": 14.47,
+  "tests/integration/test_climate_control_action.py": 31.07,
+  "tests/integration/test_climate_custom_modes.py": 28.59,
+  "tests/integration/test_continuation_actions.py": 14.96,
+  "tests/integration/test_cover_control_action.py": 26.14,
+  "tests/integration/test_crc8_helper.py": 10.92,
+  "tests/integration/test_device_id_in_state.py": 64.97,
+  "tests/integration/test_duplicate_entities.py": 30.81,
+  "tests/integration/test_entity_icon.py": 32.85,
+  "tests/integration/test_fan_turn_on_action.py": 24.91,
+  "tests/integration/test_fnv1_hash_object_id.py": 12.54,
+  "tests/integration/test_fnv1a_hash.py": 21.8,
+  "tests/integration/test_gpio_expander_cache.py": 5.2,
+  "tests/integration/test_host_logger_thread_safety.py": 21.7,
+  "tests/integration/test_host_mode_basic.py": 13.62,
+  "tests/integration/test_host_mode_batch_delay.py": 14.56,
+  "tests/integration/test_host_mode_climate_basic_state.py": 30.95,
+  "tests/integration/test_host_mode_climate_control.py": 29.06,
+  "tests/integration/test_host_mode_empty_string_options.py": 27.22,
+  "tests/integration/test_host_mode_entity_fields.py": 30.95,
+  "tests/integration/test_host_mode_fan_preset.py": 14.44,
+  "tests/integration/test_host_mode_many_entities.py": 54.13,
+  "tests/integration/test_host_mode_many_entities_multiple_connections.py": 32.17,
+  "tests/integration/test_host_mode_noise_encryption.py": 42.77,
+  "tests/integration/test_host_mode_reconnect.py": 4.06,
+  "tests/integration/test_host_mode_sensor.py": 13.47,
+  "tests/integration/test_host_ota.py": 21.4,
+  "tests/integration/test_host_preferences.py": 25.43,
+  "tests/integration/test_host_preferences_suspend_resume.py": 19.2,
+  "tests/integration/test_improv_serial_uart.py": 31.52,
+  "tests/integration/test_large_message_batching.py": 15.64,
+  "tests/integration/test_legacy_area.py": 22.63,
+  "tests/integration/test_legacy_climate_compat.py": 26.13,
+  "tests/integration/test_legacy_fan_compat.py": 24.05,
+  "tests/integration/test_light_automations.py": 30.86,
+  "tests/integration/test_light_binary_effect_off_phase.py": 23.19,
+  "tests/integration/test_light_calls.py": 32.35,
+  "tests/integration/test_light_constant_brightness.py": 29.89,
+  "tests/integration/test_light_control_action.py": 29.06,
+  "tests/integration/test_light_dim_relative_action.py": 29.61,
+  "tests/integration/test_light_effect_zero_brightness.py": 18.68,
+  "tests/integration/test_light_initial_state.py": 24.49,
+  "tests/integration/test_light_toggle_action.py": 26.46,
+  "tests/integration/test_lock_automations.py": 23.28,
+  "tests/integration/test_logger_buffered_recursion_guard.py": 24.29,
+  "tests/integration/test_loop_disable_enable.py": 45.28,
+  "tests/integration/test_loop_interval_decoupling.py": 28.35,
+  "tests/integration/test_loop_interval_default_not_pulled_forward.py": 21.97,
+  "tests/integration/test_micros_to_millis.py": 20.79,
+  "tests/integration/test_multi_click_trigger.py": 26.2,
+  "tests/integration/test_multi_device_preferences.py": 16.87,
+  "tests/integration/test_noise_encryption_key_protection.py": 77.05,
+  "tests/integration/test_object_id_api_verification.py": 73.51,
+  "tests/integration/test_object_id_friendly_name_no_mac_suffix.py": 62.33,
+  "tests/integration/test_object_id_no_friendly_name.py": 43.47,
+  "tests/integration/test_online_image_auto_detects_image_bmp_mime.py": 32.21,
+  "tests/integration/test_online_image_auto_detects_redirected_image_bmp_mime.py": 56.86,
+  "tests/integration/test_online_image_bmp.py": 50.9,
+  "tests/integration/test_oversized_payloads.py": 53.2,
+  "tests/integration/test_preference_key_stability.py": 26.09,
+  "tests/integration/test_runtime_stats.py": 18.34,
+  "tests/integration/test_safe_mode_loop_runs.py": 10.07,
+  "tests/integration/test_scheduler_blocking_warning.py": 40.91,
+  "tests/integration/test_scheduler_bulk_cleanup.py": 23.14,
+  "tests/integration/test_scheduler_defer_cancel.py": 24.54,
+  "tests/integration/test_scheduler_defer_cancel_regular.py": 13.48,
+  "tests/integration/test_scheduler_defer_fifo_simple.py": 26.86,
+  "tests/integration/test_scheduler_defer_stress.py": 27.23,
+  "tests/integration/test_scheduler_heap_stress.py": 24.02,
+  "tests/integration/test_scheduler_internal_id_no_collision.py": 24.57,
+  "tests/integration/test_scheduler_interval_reschedule.py": 13.12,
+  "tests/integration/test_scheduler_interval_zero_coerced.py": 22.91,
+  "tests/integration/test_scheduler_null_name.py": 23.46,
+  "tests/integration/test_scheduler_numeric_id_test.py": 24.54,
+  "tests/integration/test_scheduler_pool.py": 25.0,
+  "tests/integration/test_scheduler_rapid_cancellation.py": 14.68,
+  "tests/integration/test_scheduler_recursive_timeout.py": 25.35,
+  "tests/integration/test_scheduler_removed_item_race.py": 26.19,
+  "tests/integration/test_scheduler_self_keyed.py": 23.43,
+  "tests/integration/test_scheduler_simultaneous_callbacks.py": 22.16,
+  "tests/integration/test_scheduler_string_test.py": 15.22,
+  "tests/integration/test_script_array_params.py": 14.67,
+  "tests/integration/test_script_delay_params.py": 15.65,
+  "tests/integration/test_script_queued.py": 24.93,
+  "tests/integration/test_script_queued_idle_loop.py": 5.04,
+  "tests/integration/test_script_wait_on_boot.py": 13.08,
+  "tests/integration/test_select_stringref_trigger.py": 29.6,
+  "tests/integration/test_sensor_filters_delta.py": 28.01,
+  "tests/integration/test_sensor_filters_ring_buffer.py": 25.04,
+  "tests/integration/test_sensor_filters_sliding_window.py": 71.5,
+  "tests/integration/test_sensor_filters_value_list.py": 16.94,
+  "tests/integration/test_sensor_timeout_filter.py": 29.48,
+  "tests/integration/test_socket_wake_gate_tcp.py": 20.36,
+  "tests/integration/test_status_flags.py": 37.42,
+  "tests/integration/test_strftime_to.py": 22.61,
+  "tests/integration/test_syslog.py": 16.34,
+  "tests/integration/test_template_alarm_control_panel_many_sensors.py": 29.81,
+  "tests/integration/test_template_text_save.py": 25.43,
+  "tests/integration/test_text_command.py": 23.34,
+  "tests/integration/test_text_sensor_raw_state.py": 69.57,
+  "tests/integration/test_uart_mock_ld2410.py": 37.95,
+  "tests/integration/test_uart_mock_ld2412.py": 93.22,
+  "tests/integration/test_uart_mock_ld2420.py": 43.24,
+  "tests/integration/test_uart_mock_ld2450.py": 31.75,
+  "tests/integration/test_uart_mock_modbus.py": 667.4,
+  "tests/integration/test_udp.py": 9.38,
+  "tests/integration/test_use_address_runtime.py": 37.05,
+  "tests/integration/test_valve_control_action.py": 24.47,
+  "tests/integration/test_varint_five_byte_device_id.py": 25.03,
+  "tests/integration/test_wait_until_mid_loop_timing.py": 23.73,
+  "tests/integration/test_wait_until_on_boot.py": 9.16,
+  "tests/integration/test_wait_until_ordering.py": 13.3,
+  "tests/integration/test_wait_until_reentrant_restart.py": 25.23,
+  "tests/integration/test_wake_loop_forces_phase_b.py": 23.34,
+  "tests/integration/test_water_heater_template.py": 17.67
+}
diff --git a/tests/script/test_determine_jobs.py b/tests/script/test_determine_jobs.py
index 7b641e275e..4971821969 100644
--- a/tests/script/test_determine_jobs.py
+++ b/tests/script/test_determine_jobs.py
@@ -151,9 +151,14 @@ def test_main_all_tests_should_run(
         patch.object(determine_jobs, "_is_clang_tidy_full_scan", return_value=False),
         patch.object(
             determine_jobs,
-            "_all_integration_test_files",
+            "all_integration_test_files",
             return_value=fake_test_files,
         ),
+        patch.object(
+            determine_jobs,
+            "load_integration_durations",
+            return_value=dict.fromkeys(fake_test_files, 200.0),
+        ),
         patch.object(
             determine_jobs,
             "get_changed_components",
@@ -189,24 +194,12 @@ def test_main_all_tests_should_run(
     output = json.loads(captured.out)
 
     assert output["integration_tests"] is True
-    # run_all=True expands to the full glob and pre-buckets into 3 parts.
-    # Each bucket's `tests` is a JSON list of file paths.
+    assert output["integration_run_all"] is True
+    # run_all=True expands to the full glob; balance and naming are pinned
+    # by the unit tests, main() only needs to round-trip the structure
     assert isinstance(output["integration_test_buckets"], list)
-    assert len(output["integration_test_buckets"]) == 3
-    assert [b["name"] for b in output["integration_test_buckets"]] == [
-        "1/3",
-        "2/3",
-        "3/3",
-    ]
-    for bucket in output["integration_test_buckets"]:
-        assert isinstance(bucket["tests"], list)
-        for path in bucket["tests"]:
-            assert isinstance(path, str)
     bucket_files = [f for b in output["integration_test_buckets"] for f in b["tests"]]
-    assert bucket_files == fake_test_files
-    # Bucket sizes are balanced (max-min difference at most 1).
-    sizes = [len(b["tests"]) for b in output["integration_test_buckets"]]
-    assert max(sizes) - min(sizes) <= 1
+    assert sorted(bucket_files) == fake_test_files
     assert output["clang_tidy"] is True
     assert output["clang_tidy_mode"] in ["nosplit", "split"]
     assert output["clang_format"] is True
@@ -509,14 +502,24 @@ def test_compute_integration_test_buckets_at_threshold_stays_single() -> None:
 
 
 def test_compute_integration_test_buckets_just_over_threshold_splits() -> None:
-    """One file over the threshold triggers the 3-bucket fan-out, balanced."""
+    """One file over the threshold fans out fully when the weights demand it."""
     n = determine_jobs.INTEGRATION_TESTS_SPLIT_THRESHOLD + 1
     files = [f"tests/integration/test_{i:02d}.py" for i in range(n)]
-    run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    with patch.object(
+        determine_jobs,
+        "load_integration_durations",
+        return_value=dict.fromkeys(files, 200.0),
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
     assert run is True
-    assert [b["name"] for b in buckets] == ["1/3", "2/3", "3/3"]
-    union = [path for b in buckets for path in b["tests"]]
+    # threshold+1 files x 200s caps at the maximum bucket count.
+    n_buckets = determine_jobs.INTEGRATION_TESTS_SPLIT_BUCKETS
+    assert [b["name"] for b in buckets] == [
+        f"{i + 1}/{n_buckets}" for i in range(n_buckets)
+    ]
+    union = sorted(path for b in buckets for path in b["tests"])
     assert union == sorted(files)
+    # Equal weights => bucket sizes are balanced (difference at most 1).
     sizes = [len(b["tests"]) for b in buckets]
     assert max(sizes) - min(sizes) <= 1
 
@@ -526,7 +529,7 @@ def test_compute_integration_test_buckets_run_all_with_empty_glob_disables_run()
 ):
     """run_all=True but glob returns no files => run suppressed (otherwise
     pytest would collect tests outside tests/integration/)."""
-    with patch.object(determine_jobs, "_all_integration_test_files", return_value=[]):
+    with patch.object(determine_jobs, "all_integration_test_files", return_value=[]):
         run, buckets = determine_jobs._compute_integration_test_buckets(True, [])
     assert run is False
     assert buckets == []
@@ -3146,3 +3149,86 @@ def test_memory_impact_elf_layouts_are_found(tmp_path: Path) -> None:
         elf.write_text("")
 
         assert find_elf_path(build_path) == elf, f"{platform} ELF not found"
+
+
+def test_compute_integration_test_buckets_no_durations_full_fanout() -> None:
+    """Without recorded durations the fan-out stays at the maximum."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(15)]
+    with patch.object(determine_jobs, "load_integration_durations", return_value={}):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    assert len(buckets) == determine_jobs.INTEGRATION_TESTS_SPLIT_BUCKETS
+    assert sorted(f for b in buckets for f in b["tests"]) == files
+
+
+def test_compute_integration_test_buckets_adaptive_count() -> None:
+    """A small recorded total weight collapses to one bucket above the threshold."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(15)]
+    with patch.object(
+        determine_jobs,
+        "load_integration_durations",
+        return_value=dict.fromkeys(files, 10.0),
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    # 15 files x 10s recorded = 150s, under the per-bucket weight target.
+    assert [b["name"] for b in buckets] == ["1/1"]
+    assert buckets[0]["tests"] == files
+
+
+def test_compute_integration_test_buckets_duration_weighted() -> None:
+    """Heavy files spread across buckets instead of clustering by sorted name."""
+    files = [f"tests/integration/test_{i:03d}.py" for i in range(12)]
+    durations = dict.fromkeys(files, 10.0)
+    durations[files[0]] = 600.0
+    durations[files[1]] = 600.0
+    with patch.object(
+        determine_jobs, "load_integration_durations", return_value=durations
+    ):
+        run, buckets = determine_jobs._compute_integration_test_buckets(False, files)
+    assert run is True
+    assert len(buckets) >= 2
+    heavy_buckets = [b for b in buckets if set(files[:2]) & set(b["tests"])]
+    assert len(heavy_buckets) == 2, "heavy files should land in different buckets"
+    assert sorted(f for b in buckets for f in b["tests"]) == files
+
+
+def test_load_integration_durations_missing_or_corrupt(tmp_path: Path) -> None:
+    """Missing or unparsable durations data degrades to an empty mapping."""
+    with patch.object(helpers, "root_path", str(tmp_path)):
+        assert determine_jobs.load_integration_durations() == {}
+        durations_file = tmp_path / helpers.INTEGRATION_TEST_DURATIONS_FILE
+        durations_file.parent.mkdir(parents=True)
+        durations_file.write_text("not json")
+        assert determine_jobs.load_integration_durations() == {}
+        durations_file.write_text('{"tests/integration/test_a.py": 12.5}')
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # Non-positive entries are dropped, valid ones survive
+        durations_file.write_text(
+            '{"tests/integration/test_a.py": 12.5, "tests/integration/test_b.py": -1}'
+        )
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # One non-numeric entry cannot discard the whole recording
+        durations_file.write_text(
+            '{"tests/integration/test_a.py": 12.5, "tests/integration/test_b.py": null}'
+        )
+        assert determine_jobs.load_integration_durations() == {
+            "tests/integration/test_a.py": 12.5
+        }
+        # A non-dict top level degrades to empty
+        durations_file.write_text("[12.5]")
+        assert determine_jobs.load_integration_durations() == {}
+
+
+def test_committed_integration_durations_are_sane() -> None:
+    """The committed recording itself holds positive bounded floats."""
+    raw = json.loads(
+        (Path(helpers.root_path) / helpers.INTEGRATION_TEST_DURATIONS_FILE).read_text()
+    )
+    assert raw, "committed durations file missing or empty"
+    assert all(isinstance(v, (int, float)) and 0 < v < 86400 for v in raw.values())
+    assert all(k.startswith("tests/integration/test_") for k in raw)
diff --git a/tests/script/test_helpers.py b/tests/script/test_helpers.py
index 38b8c57368..7d4059da2f 100644
--- a/tests/script/test_helpers.py
+++ b/tests/script/test_helpers.py
@@ -2120,3 +2120,30 @@ def test_get_cpp_changed_components_independent_of_cwd(
     assert helpers.get_cpp_changed_components(
         ["tests/components/time/__init__.py"]
     ) == ["time"]
+
+
+def test_lpt_partition_balances_skewed_weights() -> None:
+    """Heavy items spread across groups instead of clustering."""
+    items = [f"i{n}" for n in range(6)]
+    weights = {"i0": 100.0, "i1": 90.0, "i2": 10.0, "i3": 10.0, "i4": 5.0, "i5": 5.0}
+    groups = helpers.lpt_partition(items, weights, 2)
+    group_weights = sorted(sum(weights[i] for i in g) for g in groups)
+    # Contiguous split would give 200 vs 20; LPT lands at 110 vs 110
+    assert group_weights == [110.0, 110.0]
+    assert sorted(i for g in groups for i in g) == items
+
+
+def test_lpt_partition_more_groups_than_items() -> None:
+    """Surplus groups come back empty; every item still lands somewhere."""
+    items = ["a", "b"]
+    groups = helpers.lpt_partition(items, {"a": 1.0, "b": 1.0}, 4)
+    assert len(groups) == 4
+    assert sorted(i for g in groups for i in g) == items
+    assert sum(not g for g in groups) == 2
+
+
+def test_lpt_partition_tie_determinism() -> None:
+    """Equal weights assign in input order, so output is reproducible."""
+    items = [f"i{n}" for n in range(4)]
+    weights = dict.fromkeys(items, 1.0)
+    assert helpers.lpt_partition(items, weights, 2) == [["i0", "i2"], ["i1", "i3"]]
diff --git a/tests/script/test_update_integration_test_durations.py b/tests/script/test_update_integration_test_durations.py
new file mode 100644
index 0000000000..f2f373d4bb
--- /dev/null
+++ b/tests/script/test_update_integration_test_durations.py
@@ -0,0 +1,130 @@
+"""Unit tests for script/update_integration_test_durations.py."""
+
+import json
+from pathlib import Path
+import sys
+from unittest.mock import patch
+
+import pytest
+
+# Add the script directory to Python path so we can import the module
+script_dir = str((Path(__file__).parent / ".." / ".." / "script").resolve())
+sys.path.insert(0, script_dir)
+
+import helpers  # noqa: E402
+import update_integration_test_durations as uitd  # noqa: E402
+
+JUNIT_TEMPLATE = """<?xml version="1.0" encoding="utf-8"?>
+<testsuites><testsuite>{testcases}</testsuite></testsuites>
+"""
+
+KNOWN = {
+    "tests/integration/test_a.py",
+    "tests/integration/test_b.py",
+}
+
+
+def _write_junit(path: Path, testcases: str) -> None:
+    path.write_text(JUNIT_TEMPLATE.format(testcases=testcases), encoding="utf-8")
+
+
+def test_collect_durations_sums_per_file(tmp_path: Path) -> None:
+    """Testcases from the same module sum."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t1" time="1.5"/>'
+        '<testcase classname="tests.integration.test_a" name="t2" time="2.0"/>'
+        '<testcase classname="tests.integration.test_b" name="t1" time="4.0"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {
+        "tests/integration/test_a.py": 3.5,
+        "tests/integration/test_b.py": 4.0,
+    }
+
+
+def test_collect_durations_class_based_testcase(tmp_path: Path) -> None:
+    """A class-based classname still maps to its module file."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a.TestFoo" name="t" time="2.5"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {
+        "tests/integration/test_a.py": 2.5
+    }
+
+
+def test_collect_durations_unknown_module_skipped(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    """A classname that maps to no known file is skipped with a warning."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_gone" name="t" time="2.5"/>',
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {}
+    assert "test_gone" in capsys.readouterr().err
+
+
+def test_collect_durations_skips_skipped_testcases(tmp_path: Path) -> None:
+    """Skipped testcases do not record a bogus zero duration."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t" time="0">'
+        "<skipped/></testcase>",
+    )
+    assert uitd.collect_durations(tmp_path, KNOWN) == {}
+
+
+def test_collect_durations_unexpected_classname_aborts(tmp_path: Path) -> None:
+    """A classname outside tests.integration means the junit layout changed."""
+    _write_junit(
+        tmp_path / "a.xml",
+        '<testcase classname="tests.unit_tests.test_x" name="t" time="9.0"/>',
+    )
+    with pytest.raises(SystemExit):
+        uitd.collect_durations(tmp_path, KNOWN)
+
+
+def test_collect_durations_empty_dir_aborts(tmp_path: Path) -> None:
+    """No junit XML at all is a hard error, not an empty recording."""
+    with pytest.raises(SystemExit):
+        uitd.collect_durations(tmp_path, KNOWN)
+
+
+def test_main_merges_partial_run(tmp_path: Path) -> None:
+    """A partial run merges over the previous data instead of truncating it."""
+    tests_dir = tmp_path / "tests" / "integration"
+    tests_dir.mkdir(parents=True)
+    for name in ("test_a", "test_b", "test_c"):
+        (tests_dir / f"{name}.py").write_text("", encoding="utf-8")
+    durations_file = tmp_path / helpers.INTEGRATION_TEST_DURATIONS_FILE
+    durations_file.write_text(
+        json.dumps(
+            {
+                "tests/integration/test_a.py": 5.0,
+                "tests/integration/test_b.py": 7.0,
+                "tests/integration/test_gone.py": 9.0,
+            }
+        ),
+        encoding="utf-8",
+    )
+    junit_dir = tmp_path / "junit"
+    junit_dir.mkdir()
+    _write_junit(
+        junit_dir / "a.xml",
+        '<testcase classname="tests.integration.test_a" name="t" time="6.0"/>',
+    )
+    with (
+        patch.object(helpers, "root_path", str(tmp_path)),
+        patch.object(uitd, "DURATIONS_FILE", durations_file),
+    ):
+        # 1 of 3 files covered: refused without --allow-partial
+        with patch.object(sys, "argv", ["uitd", str(junit_dir)]):
+            assert uitd.main() == uitd.EXIT_LOW_COVERAGE
+        with patch.object(sys, "argv", ["uitd", str(junit_dir), "--allow-partial"]):
+            assert uitd.main() == 0
+    # test_a updated, test_b kept, deleted test_gone dropped
+    assert json.loads(durations_file.read_text()) == {
+        "tests/integration/test_a.py": 6.0,
+        "tests/integration/test_b.py": 7.0,
+    }
diff --git a/tests/unit_tests/test_platformio_prefetch.py b/tests/unit_tests/test_platformio_prefetch.py
index 91fb78c6af..d0785d2724 100644
--- a/tests/unit_tests/test_platformio_prefetch.py
+++ b/tests/unit_tests/test_platformio_prefetch.py
@@ -1417,6 +1417,76 @@ def test_prefetch_installs_cached_archives_without_downloads(
     assert not (tmp_path / pf._SENTINEL_NAME).exists()
 
 
+@pytest.mark.parametrize(
+    ("platform_group", "lib_group", "expected"),
+    [
+        (
+            [("toolchain-x@1", _FakeSpec(name="toolchain-x"))],
+            [],
+            ["configure", "install", "configure"],
+        ),
+        ([], [("noise-c@1.0", _FakeSpec(name="noise-c"))], ["configure", "install"]),
+    ],
+)
+def test_prefetch_reconfigures_only_after_platform_installs(
+    tmp_path: Path, platform_group: list, lib_group: list, expected: list[str]
+) -> None:
+    """Installed platform packages get a second configure pass; libraries do not."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    order: list[str] = []
+    fake_platform = MagicMock()
+    fake_platform.packages = {}
+    fake_platform.configure_project_packages.side_effect = lambda env, targets: (
+        order.append("configure")
+    )
+    config = _fake_config(
+        tmp_path, {"platform": "fake/p@1", "lib_deps": ["esphome/noise-c@1.0"]}
+    )
+    modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(
+            pf,
+            "_registry_jobs",
+            side_effect=[([], 0, platform_group), ([], 0, lib_group)],
+        ),
+        patch.object(pf, "_uri_jobs", return_value=([], 0, [])),
+        patch.object(pf, "_preinstall", side_effect=lambda *_: order.append("install")),
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    assert order == expected
+
+
+@pytest.mark.parametrize(
+    "err", [RuntimeError("idf_tools.py failed"), SystemExit("postinstall exited")]
+)
+def test_prefetch_settle_failure_warns_and_continues(
+    tmp_path: Path, caplog: pytest.LogCaptureFixture, err: BaseException
+) -> None:
+    """A failing second configure pass only costs the speedup."""
+    _write_ini(tmp_path, "[env:testenv]\nplatform = fake/p@1\n")
+    fake_platform = MagicMock()
+    fake_platform.packages = {}
+    fake_platform.configure_project_packages.side_effect = [None, err]
+    config = _fake_config(tmp_path, {"platform": "fake/p@1"})
+    modules = _pio_modules(tmp_path, fake_platform, MagicMock(), config)
+    with (
+        patch.dict("sys.modules", modules),
+        patch.object(
+            pf,
+            "_registry_jobs",
+            side_effect=[
+                ([], 0, [("toolchain-x@1", _FakeSpec(name="toolchain-x"))]),
+                ([], 0, []),
+            ],
+        ),
+        patch.object(pf, "_uri_jobs", return_value=([], 0, [])),
+        patch.object(pf, "_preinstall"),
+    ):
+        pf._prefetch(tmp_path, "testenv")
+    assert f"Could not settle platform packages: {err}" in caplog.text
+
+
 def test_preinstall_extracts_in_parallel_under_one_lock(tmp_path: Path) -> None:
     """The manager lock wraps the whole batch; per-thread managers share
     its package dir; one failing install leaves the rest alone."""