derive recording paths from the cache segment timestamp

Recording paths carry one second of resolution, but since sub stream recording start times are resolved to fractional wall clock, anchored to the cache file mtime and chained to the previous segment's end. A stream cutting segments faster than once a second resolves consecutive segments into the same second, so two rows collide on the unique path index and the batch insert fails. The cache segment name is unique per camera stream and second by construction because ffmpeg names segments with strftime, so the recording path is now built from that timestamp while the row keeps the resolved start time. This also restores the path semantics from before sub stream recording, when start times came straight from the cache filename.

Nothing derives times from recording paths: playback offsets, stream switching, and export all use the row's start time, which is unchanged, and the recordings sync matches files by exact path string.

frigate/record/maintainer.py

@@ -79,6 +79,28 @@ def parse_cache_segment_name(basename: str) -> tuple[str, str, str] | None:
     return (prefix, STREAM_TYPE_MAIN, date)
 
 
+def segment_path_time(cache_path: str) -> datetime.datetime | None:
+    """Timestamp a segment's recording path is built from, or None if unparsable.
+
+    Recording paths carry one second of resolution, and so does ffmpeg's cache
+    segment template, which makes a cache file name unique per camera stream
+    and second. Resolved start times are not: a stream cutting segments faster
+    than once a second resolves consecutive segments into the same second, and
+    building the path from those collides on the unique path index.
+    """
+    parsed = parse_cache_segment_name(Path(cache_path).stem)
+
+    if parsed is None:
+        return None
+
+    try:
+        return datetime.datetime.strptime(parsed[2], CACHE_SEGMENT_FORMAT).astimezone(
+            datetime.UTC
+        )
+    except ValueError:
+        return None
+
+
 class SegmentInfo:
     def __init__(
         self,
@@ -865,18 +887,20 @@ class RecordingMaintainer(threading.Thread):
         video_codec: str | None = None,
         keyframes: list[int] | None = None,
     ) -> dict[str, Any] | None:
-        # directory will be in utc due to start_time being in utc
+        path_time = segment_path_time(cache_path) or start_time
+
+        # directory will be in utc due to path_time being in utc
         # sub segments get a tagged directory to avoid filename collisions
         directory = os.path.join(
             RECORD_DIR,
-            start_time.strftime("%Y-%m-%d/%H"),
+            path_time.strftime("%Y-%m-%d/%H"),
             camera if stream_type == STREAM_TYPE_MAIN else f"{camera}{SUB_CACHE_TAG}",
         )
 
         os.makedirs(directory, exist_ok=True)
 
-        # file will be in utc due to start_time being in utc
-        file_name = f"{start_time.strftime('%M.%S.mp4')}"
+        # file will be in utc due to path_time being in utc
+        file_name = f"{path_time.strftime('%M.%S.mp4')}"
         file_path = os.path.join(directory, file_name)
 
         try:

frigate/test/test_record_sub_maintainer.py

@@ -15,6 +15,7 @@ from frigate.record.maintainer import (
     RecordingMaintainer,
     SegmentInfo,
     parse_cache_segment_name,
+    segment_path_time,
 )
 
 
@@ -349,6 +350,98 @@ class TestSegmentAudioPresence(unittest.IsolatedAsyncioTestCase):
                 self.assertEqual(result[Recordings.video_codec.name], video_codec)
 
 
+class TestSegmentPathTime(unittest.IsolatedAsyncioTestCase):
+    """The recording path must stay unique when segments are shorter than a second."""
+
+    def _build_maintainer(self) -> RecordingMaintainer:
+        camera_config = MagicMock()
+        camera_config.record.enabled = True
+        camera_config.record.continuous.days = 1
+        camera_config.record.motion.days = 0
+
+        config = MagicMock()
+        config.cameras = {"test_cam": camera_config}
+
+        maintainer = RecordingMaintainer.__new__(RecordingMaintainer)
+        maintainer.config = config
+        maintainer.end_time_cache = {}
+        maintainer.object_recordings_info = defaultdict(list)
+        maintainer.audio_recordings_info = defaultdict(list)
+        maintainer.recordings_publisher = MagicMock()
+        maintainer.last_segment_end = {("test_cam", "main"): 0.0}
+        return maintainer
+
+    def test_parses_main_and_sub_names(self):
+        expected = datetime.datetime(2026, 6, 10, 14, 30, 22, tzinfo=datetime.UTC)
+        self.assertEqual(
+            segment_path_time("/tmp/cache/test_cam@20260610143022+0000.mp4"), expected
+        )
+        self.assertEqual(
+            segment_path_time("/tmp/cache/test_cam@sub@20260610143022+0000.mp4"),
+            expected,
+        )
+
+    def test_returns_none_for_unparsable_names(self):
+        self.assertIsNone(segment_path_time("/tmp/cache/garbage.mp4"))
+        self.assertIsNone(segment_path_time("/tmp/cache/test_cam@notadate.mp4"))
+
+    async def test_sub_second_segments_get_distinct_paths(self):
+        # two cache files a second apart whose resolved starts both land in
+        # second 22; deriving the path from the resolved start collides
+        segments = [
+            ("test_cam@20260610143022+0000.mp4", 100_000),
+            ("test_cam@20260610143023+0000.mp4", 980_000),
+        ]
+        paths = []
+
+        with tempfile.TemporaryDirectory() as tmpdir:
+            for name, microsecond in segments:
+                maintainer = self._build_maintainer()
+                maintainer.config.ffmpeg.ffmpeg_path = "ffmpeg"
+
+                start_time = datetime.datetime(
+                    2026, 6, 10, 14, 30, 22, microsecond, tzinfo=datetime.UTC
+                )
+                cache_path = os.path.join(tmpdir, name)
+                with open(cache_path, "wb") as f:
+                    f.write(b"\x00" * 16)
+
+                proc = MagicMock()
+                proc.returncode = 0
+                proc.wait = AsyncMock(return_value=0)
+
+                with (
+                    patch(
+                        "frigate.record.maintainer.RECORD_DIR",
+                        os.path.join(tmpdir, "recordings"),
+                    ),
+                    patch(
+                        "frigate.record.maintainer.asyncio.create_subprocess_exec",
+                        AsyncMock(return_value=proc),
+                    ),
+                ):
+                    result = await maintainer.move_segment(
+                        "test_cam",
+                        "main",
+                        start_time,
+                        start_time + datetime.timedelta(seconds=0.96),
+                        0.96,
+                        cache_path,
+                        SegmentInfo(0, 0, 0, 0),
+                    )
+
+                self.assertIsNotNone(result)
+                paths.append(result[Recordings.path.name])
+                # the row keeps the resolved start even though the path doesn't
+                self.assertEqual(
+                    result[Recordings.start_time.name], start_time.timestamp()
+                )
+
+        self.assertEqual(len(set(paths)), 2, paths)
+        self.assertTrue(paths[0].endswith("30.22.mp4"), paths[0])
+        self.assertTrue(paths[1].endswith("30.23.mp4"), paths[1])
+
+
 class TestSegmentStartChaining(unittest.IsolatedAsyncioTestCase):
     """Contiguous segments must chain start times across filename truncation.