diff --git a/homeassistant/components/stream/hls.py b/homeassistant/components/stream/hls.py
index a88ee465e25..e1a0e6a8f67 100644
--- a/homeassistant/components/stream/hls.py
+++ b/homeassistant/components/stream/hls.py
@@ -18,7 +18,14 @@ from .const import (
     MAX_SEGMENTS,
     NUM_PLAYLIST_SEGMENTS,
 )
-from .core import PROVIDERS, IdleTimer, StreamOutput, StreamSettings, StreamView
+from .core import (
+    PROVIDERS,
+    IdleTimer,
+    Segment,
+    StreamOutput,
+    StreamSettings,
+    StreamView,
+)
 from .fmp4utils import get_codec_string
 
 if TYPE_CHECKING:
@@ -44,7 +51,7 @@ class HlsStreamOutput(StreamOutput):
         """Initialize HLS output."""
         super().__init__(hass, idle_timer, deque_maxlen=MAX_SEGMENTS)
         self.stream_settings: StreamSettings = hass.data[DOMAIN][ATTR_SETTINGS]
-        self._target_duration = 0.0
+        self._target_duration = self.stream_settings.min_segment_duration
 
     @property
     def name(self) -> str:
@@ -53,20 +60,23 @@ class HlsStreamOutput(StreamOutput):
 
     @property
     def target_duration(self) -> float:
-        """
-        Return the target duration.
-
-        The target duration is calculated as the max duration of any given segment,
-        and it is calculated only one time to avoid changing during playback.
-        """
-        if self._target_duration:
-            return self._target_duration
-        durations = [s.duration for s in self._segments if s.complete]
-        if len(durations) < 2:
-            return self.stream_settings.min_segment_duration
-        self._target_duration = max(durations)
+        """Return the target duration."""
         return self._target_duration
 
+    @callback
+    def _async_put(self, segment: Segment) -> None:
+        """Async put and also update the target duration.
+
+        The target duration is calculated as the max duration of any given segment.
+        Technically it should not change per the hls spec, but some cameras adjust
+        their GOPs periodically so we need to account for this change.
+        """
+        super()._async_put(segment)
+        self._target_duration = (
+            max((s.duration for s in self._segments), default=segment.duration)
+            or self.stream_settings.min_segment_duration
+        )
+
 
 class HlsMasterPlaylistView(StreamView):
     """Stream view used only for Chromecast compatibility."""
diff --git a/tests/components/stream/test_ll_hls.py b/tests/components/stream/test_ll_hls.py
index 1156832ada9..9a0d94136b9 100644
--- a/tests/components/stream/test_ll_hls.py
+++ b/tests/components/stream/test_ll_hls.py
@@ -224,7 +224,9 @@ async def test_ll_hls_stream(hass, hls_stream, stream_worker_sync):
                     datetimes[-1] - datetimes.popleft()
                 ).total_seconds()
                 if segment_duration:
-                    assert datetime_duration == segment_duration
+                    assert math.isclose(
+                        datetime_duration, segment_duration, rel_tol=1e-3
+                    )
                     tested[datetime_re] = True
             continue
         match = inf_re.match(line)
@@ -232,7 +234,7 @@ async def test_ll_hls_stream(hass, hls_stream, stream_worker_sync):
             segment_duration = float(match.group("segment_duration"))
             # Check that segment durations are consistent with part durations
             if len(part_durations) > 1:
-                assert math.isclose(sum(part_durations), segment_duration)
+                assert math.isclose(sum(part_durations), segment_duration, rel_tol=1e-3)
                 tested[inf_re] = True
                 part_durations.clear()
     # make sure all playlist tests were performed
diff --git a/tests/components/stream/test_recorder.py b/tests/components/stream/test_recorder.py
index ba35b5a4b72..fc6b48d273b 100644
--- a/tests/components/stream/test_recorder.py
+++ b/tests/components/stream/test_recorder.py
@@ -194,7 +194,7 @@ async def test_record_stream_audio(
     await async_setup_component(hass, "stream", {"stream": {}})
 
     # Generate source video with no audio
-    source = generate_h264_video(container_format="mov")
+    orig_source = generate_h264_video(container_format="mov")
 
     for a_codec, expected_audio_streams in (
         ("aac", 1),  # aac is a valid mp4 codec
@@ -202,9 +202,8 @@ async def test_record_stream_audio(
         ("empty", 0),  # audio stream with no packets
         (None, 0),  # no audio stream
     ):
-
         # Remux source video with new audio
-        source = remux_with_audio(source, "mov", a_codec)  # mov can store PCM
+        source = remux_with_audio(orig_source, "mov", a_codec)  # mov can store PCM
 
         record_worker_sync.reset()
         stream_worker_sync.pause()