diff --git a/homeassistant/components/timer/icons.json b/homeassistant/components/timer/icons.json
index 5641020f8318..769fdfd10f88 100644
--- a/homeassistant/components/timer/icons.json
+++ b/homeassistant/components/timer/icons.json
@@ -45,6 +45,9 @@
     },
     "started": {
       "trigger": "mdi:timer-play"
+    },
+    "time_remaining": {
+      "trigger": "mdi:timer-alert-outline"
     }
   }
 }
diff --git a/homeassistant/components/timer/strings.json b/homeassistant/components/timer/strings.json
index 9e4111281050..ddc6e9db44f6 100644
--- a/homeassistant/components/timer/strings.json
+++ b/homeassistant/components/timer/strings.json
@@ -183,6 +183,15 @@
         }
       },
       "name": "Timer started"
+    },
+    "time_remaining": {
+      "description": "Triggers when one or more timers reach a specific remaining time.",
+      "fields": {
+        "remaining": {
+          "name": "Time remaining"
+        }
+      },
+      "name": "Timer time remaining"
     }
   }
 }
diff --git a/homeassistant/components/timer/trigger.py b/homeassistant/components/timer/trigger.py
index 8f79391769b4..3107fdeeff5f 100644
--- a/homeassistant/components/timer/trigger.py
+++ b/homeassistant/components/timer/trigger.py
@@ -2,11 +2,161 @@
 
 from __future__ import annotations
 
-from homeassistant.core import HomeAssistant
-from homeassistant.helpers.automation import DomainSpec
-from homeassistant.helpers.trigger import Trigger, make_entity_target_state_trigger
+from datetime import datetime, timedelta
+from typing import cast, override
+
+import voluptuous as vol
+
+from homeassistant.const import ATTR_ENTITY_ID, CONF_OPTIONS
+from homeassistant.core import CALLBACK_TYPE, Context, HomeAssistant, State, callback
+from homeassistant.helpers import config_validation as cv
+from homeassistant.helpers.automation import DomainSpec, filter_by_domain_specs
+from homeassistant.helpers.event import async_track_point_in_utc_time
+from homeassistant.helpers.target import (
+    TargetStateChangedData,
+    async_track_target_selector_state_change_event,
+)
+from homeassistant.helpers.trigger import (
+    ENTITY_STATE_TRIGGER_SCHEMA,
+    Trigger,
+    TriggerActionRunner,
+    TriggerConfig,
+    make_entity_target_state_trigger,
+)
+from homeassistant.helpers.typing import ConfigType
+from homeassistant.util import dt as dt_util
+
+from . import ATTR_FINISHES_AT, ATTR_LAST_TRANSITION, DOMAIN, STATUS_ACTIVE
+
+CONF_REMAINING = "remaining"
+
+TIME_REMAINING_TRIGGER_SCHEMA = ENTITY_STATE_TRIGGER_SCHEMA.extend(
+    {
+        vol.Required(CONF_OPTIONS): {
+            vol.Required(CONF_REMAINING): cv.positive_time_period_dict,
+        },
+    }
+)
+
+
+class TimeRemainingTrigger(Trigger):
+    """Trigger when a timer has a specific amount of time remaining."""
+
+    _domain_specs: dict[str, DomainSpec] = {DOMAIN: DomainSpec()}
+    _schema = TIME_REMAINING_TRIGGER_SCHEMA
+
+    @override
+    @classmethod
+    async def async_validate_config(
+        cls, hass: HomeAssistant, config: ConfigType
+    ) -> ConfigType:
+        """Validate config."""
+        return cast(ConfigType, cls._schema(config))
+
+    def __init__(self, hass: HomeAssistant, config: TriggerConfig) -> None:
+        """Initialize the time remaining trigger."""
+        super().__init__(hass, config)
+        assert config.target is not None
+        self._target = config.target
+        options = config.options or {}
+        self._remaining: timedelta = options[CONF_REMAINING]
+
+    def entity_filter(self, entities: set[str]) -> set[str]:
+        """Filter entities to timer domain."""
+        return filter_by_domain_specs(self._hass, self._domain_specs, entities)
+
+    @override
+    async def async_attach_runner(
+        self, run_action: TriggerActionRunner
+    ) -> CALLBACK_TYPE:
+        """Attach the trigger to an action runner."""
+        scheduled: dict[str, CALLBACK_TYPE] = {}
+
+        @callback
+        def schedule_for_state(
+            entity_id: str,
+            to_state: State | None,
+            context: Context | None,
+        ) -> None:
+            """Schedule a fire for an active timer state, if applicable."""
+            if to_state is None:
+                return
+            if to_state.state != STATUS_ACTIVE:
+                return
+
+            finishes_at_str = to_state.attributes.get(ATTR_FINISHES_AT)
+            if finishes_at_str is None:
+                return
+
+            finishes_at = dt_util.parse_datetime(finishes_at_str)
+            if finishes_at is None:
+                return
+
+            fire_at = finishes_at - self._remaining
+            if fire_at <= dt_util.utcnow():
+                return
+
+            @callback
+            def fire_trigger(now: datetime) -> None:
+                """Fire the trigger."""
+                scheduled.pop(entity_id, None)
+                run_action(
+                    {
+                        ATTR_ENTITY_ID: entity_id,
+                        "to_state": to_state,
+                        "remaining": self._remaining,
+                    },
+                    f"time remaining of {entity_id}",
+                    context,
+                )
+
+            scheduled[entity_id] = async_track_point_in_utc_time(
+                self._hass, fire_trigger, fire_at
+            )
+
+        @callback
+        def state_change_listener(
+            target_state_change_data: TargetStateChangedData,
+        ) -> None:
+            """Listen for state changes and schedule trigger."""
+            event = target_state_change_data.state_change_event
+            entity_id: str = event.data["entity_id"]
+            to_state = event.data["new_state"]
+
+            # Cancel any previously scheduled callback for this entity
+            if entity_id in scheduled:
+                scheduled.pop(entity_id)()
+
+            schedule_for_state(entity_id, to_state, event.context)
+
+        @callback
+        def on_entities_update(added: set[str], removed: set[str]) -> None:
+            """Handle changes to the tracked entity set."""
+            for entity_id in removed:
+                if entity_id in scheduled:
+                    scheduled.pop(entity_id)()
+            for entity_id in added:
+                state = self._hass.states.get(entity_id)
+                schedule_for_state(entity_id, state, state.context if state else None)
+
+        unsub = async_track_target_selector_state_change_event(
+            self._hass,
+            self._target,
+            state_change_listener,
+            self.entity_filter,
+            on_entities_update,
+        )
+
+        @callback
+        def async_remove() -> None:
+            """Remove state listeners."""
+            unsub()
+            for cancel in scheduled.values():
+                cancel()
+            scheduled.clear()
+
+        return async_remove
 
-from . import ATTR_LAST_TRANSITION, DOMAIN
 
 TRIGGERS: dict[str, type[Trigger]] = {
     "cancelled": make_entity_target_state_trigger(
@@ -24,6 +174,7 @@ TRIGGERS: dict[str, type[Trigger]] = {
     "started": make_entity_target_state_trigger(
         {DOMAIN: DomainSpec(value_source=ATTR_LAST_TRANSITION)}, "started"
     ),
+    "time_remaining": TimeRemainingTrigger,
 }
 
 
diff --git a/homeassistant/components/timer/triggers.yaml b/homeassistant/components/timer/triggers.yaml
index 110acc4f5b8f..a94482202b89 100644
--- a/homeassistant/components/timer/triggers.yaml
+++ b/homeassistant/components/timer/triggers.yaml
@@ -20,3 +20,13 @@ finished: *trigger_common
 paused: *trigger_common
 restarted: *trigger_common
 started: *trigger_common
+
+time_remaining:
+  target:
+    entity:
+      domain: timer
+  fields:
+    remaining:
+      required: true
+      selector:
+        duration:
diff --git a/tests/components/timer/test_trigger.py b/tests/components/timer/test_trigger.py
index ddc58f9564f4..d8d094609592 100644
--- a/tests/components/timer/test_trigger.py
+++ b/tests/components/timer/test_trigger.py
@@ -1,18 +1,38 @@
 """Test timer triggers."""
 
+from datetime import timedelta
+import logging
 from typing import Any
 
+from freezegun.api import FrozenDateTimeFactory
 import pytest
+import voluptuous as vol
 
 from homeassistant.components.timer import (
+    ATTR_FINISHES_AT,
     ATTR_LAST_TRANSITION,
     DOMAIN,
     STATUS_ACTIVE,
     STATUS_IDLE,
     STATUS_PAUSED,
 )
-from homeassistant.core import HomeAssistant
+from homeassistant.const import (
+    ATTR_LABEL_ID,
+    CONF_ENTITY_ID,
+    CONF_OPTIONS,
+    CONF_PLATFORM,
+    CONF_TARGET,
+)
+from homeassistant.core import Context, HomeAssistant, callback
+from homeassistant.helpers import entity_registry as er, label_registry as lr
+from homeassistant.helpers.trigger import (
+    async_initialize_triggers,
+    async_validate_trigger_config,
+)
+from homeassistant.helpers.typing import TemplateVarsType
+from homeassistant.util import dt as dt_util
 
+from tests.common import async_fire_time_changed
 from tests.components.common import (
     TriggerStateDescription,
     assert_trigger_behavior_any,
@@ -40,6 +60,7 @@ async def target_timers(hass: HomeAssistant) -> dict[str, list[str]]:
         "timer.paused",
         "timer.restarted",
         "timer.started",
+        "timer.time_remaining",
     ],
 )
 async def test_timer_triggers_gated_by_labs_flag(
@@ -58,6 +79,7 @@ async def test_timer_triggers_gated_by_labs_flag(
         ("timer.paused", {}, True, True),
         ("timer.restarted", {}, True, True),
         ("timer.started", {}, True, True),
+        ("timer.time_remaining", {"remaining": {"hours": 1}}, False, False),
     ],
 )
 async def test_timer_trigger_options_validation(
@@ -249,3 +271,490 @@ async def test_timer_trigger_behavior_last(
         trigger_options=trigger_options,
         states=states,
     )
+
+
+# --- time_remaining trigger tests ---
+
+
+async def _arm_time_remaining_trigger(
+    hass: HomeAssistant,
+    entity_id: str,
+    remaining: dict[str, int],
+    calls: list[dict[str, Any]],
+    *,
+    target: dict[str, Any] | None = None,
+) -> None:
+    """Arm the time_remaining trigger."""
+    trigger_config = await async_validate_trigger_config(
+        hass,
+        [
+            {
+                CONF_PLATFORM: "timer.time_remaining",
+                CONF_TARGET: target or {CONF_ENTITY_ID: entity_id},
+                CONF_OPTIONS: {"remaining": remaining},
+            }
+        ],
+    )
+
+    @callback
+    def action(run_variables: TemplateVarsType, context: Context | None = None) -> None:
+        calls.append(run_variables["trigger"])
+
+    logger = logging.getLogger(__name__)
+
+    def log_cb(level: int, msg: str, **kwargs: Any) -> None:
+        logger._log(level, "%s", msg, **kwargs)
+
+    await async_initialize_triggers(
+        hass,
+        trigger_config,
+        action,
+        domain="test",
+        name="test_trigger",
+        log_cb=log_cb,
+    )
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_validation(hass: HomeAssistant) -> None:
+    """Test time_remaining trigger config validation."""
+    # Valid config
+    await async_validate_trigger_config(
+        hass,
+        [
+            {
+                CONF_PLATFORM: "timer.time_remaining",
+                CONF_TARGET: {CONF_ENTITY_ID: "timer.test"},
+                CONF_OPTIONS: {"remaining": {"seconds": 30}},
+            }
+        ],
+    )
+
+    # Missing remaining option
+    with pytest.raises(vol.Invalid):
+        await async_validate_trigger_config(
+            hass,
+            [
+                {
+                    CONF_PLATFORM: "timer.time_remaining",
+                    CONF_TARGET: {CONF_ENTITY_ID: "timer.test"},
+                    CONF_OPTIONS: {},
+                }
+            ],
+        )
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_fires(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test time_remaining trigger fires at the right time."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # Start timer with 60 second duration
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    # Advance to 25 seconds - 35 seconds remaining, should not fire
+    freezer.move_to(now + timedelta(seconds=25))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    # Advance to 30 seconds - 30 seconds remaining, should fire
+    freezer.move_to(now + timedelta(seconds=30))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert calls[0]["entity_id"] == "timer.test"
+    assert calls[0]["remaining"] == timedelta(seconds=30)
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_paused_before_threshold(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test time_remaining trigger does not fire when timer is paused before threshold."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # Start timer with 60 second duration
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    # Pause timer at 10 seconds (before the 30-second threshold)
+    freezer.move_to(now + timedelta(seconds=10))
+    hass.states.async_set(
+        "timer.test",
+        STATUS_PAUSED,
+        {ATTR_LAST_TRANSITION: "paused"},
+    )
+    await hass.async_block_till_done()
+
+    # Advance past the original fire time - should not fire since paused
+    freezer.move_to(now + timedelta(seconds=35))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_cancelled_before_threshold(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test time_remaining trigger does not fire when timer is cancelled before threshold."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # Start timer with 60 second duration
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    # Cancel timer at 10 seconds
+    freezer.move_to(now + timedelta(seconds=10))
+    hass.states.async_set(
+        "timer.test",
+        STATUS_IDLE,
+        {ATTR_LAST_TRANSITION: "cancelled"},
+    )
+    await hass.async_block_till_done()
+
+    # Advance past the original fire time - should not fire since cancelled
+    freezer.move_to(now + timedelta(seconds=35))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_restarted(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test time_remaining trigger reschedules when timer is restarted."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # Start timer with 60 second duration
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    # Restart timer at 10 seconds with a new 60-second duration
+    freezer.move_to(now + timedelta(seconds=10))
+    new_finishes_at = now + timedelta(seconds=70)  # 10s elapsed + 60s new
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {
+            ATTR_LAST_TRANSITION: "restarted",
+            ATTR_FINISHES_AT: new_finishes_at.isoformat(),
+        },
+    )
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    # Original fire time (30s) should not fire since rescheduled
+    freezer.move_to(now + timedelta(seconds=30))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    # New fire time: new_finishes_at - 30s = now + 40s
+    freezer.move_to(now + timedelta(seconds=40))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_short_timer(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test time_remaining trigger does not fire when timer duration is shorter than remaining threshold."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # Start timer with only 20 second duration (less than 30s threshold)
+    finishes_at = now + timedelta(seconds=20)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    # fire_at = now + 20 - 30 = now - 10 (in the past), should not schedule
+    # Advance past the timer's end time
+    freezer.move_to(now + timedelta(seconds=25))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_already_active_at_attach(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test trigger schedules for timers already active when the trigger attaches."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    # Timer is already active before the trigger is armed
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # No fire yet
+    assert len(calls) == 0
+
+    # Before fire_at (finishes_at - 30s = now + 30s) — should not fire
+    freezer.move_to(now + timedelta(seconds=25))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    # At fire_at — should fire even though no state change occurred
+    freezer.move_to(now + timedelta(seconds=30))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert calls[0]["entity_id"] == "timer.test"
+    assert calls[0]["remaining"] == timedelta(seconds=30)
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_already_active_past_threshold_at_attach(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test trigger does not schedule for timers already past the fire point at attach."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    # Timer is active but only 20 seconds remain — past the 30s threshold already
+    finishes_at = now + timedelta(seconds=20)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # Advance past the timer's finishing time — should never fire
+    freezer.move_to(now + timedelta(seconds=25))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_idle_at_attach(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test trigger does not schedule for non-active timers at attach time."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    hass.states.async_set("timer.test", STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # Even far in the future, no fire because timer never started
+    freezer.move_to(now + timedelta(seconds=120))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_active_on_first_state_event(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+) -> None:
+    """Test trigger schedules when first observed state event has no from_state.
+
+    This simulates a timer entity that is created/restored after the trigger
+    is attached and appears directly in active state (e.g., RestoreEntity on
+    restart), where the initial state-change event has from_state=None.
+    """
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    await _arm_time_remaining_trigger(hass, "timer.test", {"seconds": 30}, calls)
+
+    # First state event for the entity has no old_state
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        "timer.test",
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+    # Advance to fire time — should still fire even though from_state was None
+    freezer.move_to(now + timedelta(seconds=30))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert calls[0]["entity_id"] == "timer.test"
+    assert calls[0]["remaining"] == timedelta(seconds=30)
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_entity_removed_from_target(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Test trigger cancels scheduled fire when entity is removed from the target."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    label_reg = lr.async_get(hass)
+    label = label_reg.async_create("Test Time Remaining")
+
+    entry = entity_registry.async_get_or_create(
+        domain=DOMAIN, platform="test", unique_id="time_remaining_remove"
+    )
+    entity_registry.async_update_entity(entry.entity_id, labels={label.label_id})
+
+    hass.states.async_set(entry.entity_id, STATUS_IDLE, {ATTR_LAST_TRANSITION: None})
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(
+        hass,
+        entry.entity_id,
+        {"seconds": 30},
+        calls,
+        target={ATTR_LABEL_ID: label.label_id},
+    )
+
+    # Start the timer — this schedules a fire via the state-change path
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        entry.entity_id,
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    # Remove the entity from the target by stripping its label
+    freezer.move_to(now + timedelta(seconds=10))
+    entity_registry.async_update_entity(entry.entity_id, labels=set())
+    await hass.async_block_till_done()
+
+    # Advance past the original fire time — should not fire since cancelled
+    freezer.move_to(now + timedelta(seconds=35))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 0
+
+
+@pytest.mark.usefixtures("enable_labs_preview_features")
+async def test_time_remaining_trigger_entity_added_to_target(
+    hass: HomeAssistant,
+    freezer: FrozenDateTimeFactory,
+    entity_registry: er.EntityRegistry,
+) -> None:
+    """Test trigger schedules a fire for an active timer added to the target later."""
+    now = dt_util.utcnow()
+    calls: list[dict[str, Any]] = []
+
+    label_reg = lr.async_get(hass)
+    label = label_reg.async_create("Test Time Remaining Add")
+
+    entry = entity_registry.async_get_or_create(
+        domain=DOMAIN, platform="test", unique_id="time_remaining_add"
+    )
+
+    # Timer is active, but not in the target yet
+    finishes_at = now + timedelta(seconds=60)
+    hass.states.async_set(
+        entry.entity_id,
+        STATUS_ACTIVE,
+        {ATTR_LAST_TRANSITION: "started", ATTR_FINISHES_AT: finishes_at.isoformat()},
+    )
+    await hass.async_block_till_done()
+
+    await _arm_time_remaining_trigger(
+        hass,
+        entry.entity_id,
+        {"seconds": 30},
+        calls,
+        target={ATTR_LABEL_ID: label.label_id},
+    )
+
+    # Now label the entity so it joins the target
+    entity_registry.async_update_entity(entry.entity_id, labels={label.label_id})
+    await hass.async_block_till_done()
+
+    # Advance to the fire time — should fire even though no state change occurred
+    freezer.move_to(now + timedelta(seconds=30))
+    async_fire_time_changed(hass)
+    await hass.async_block_till_done()
+    assert len(calls) == 1
+    assert calls[0]["entity_id"] == entry.entity_id
+    assert calls[0]["remaining"] == timedelta(seconds=30)