fix(media): prevent looping local video tasks from stopping at EOF

src/flow/channel/AlgChannelDemux.cc

@@ -4,6 +4,7 @@
 
 #include <algorithm>
 
+#include "flow/channel/VideoEofPolicy.h"
 #include "service/detail/ServiceRegistry.h"
 #include "service/event/IEventNotifier.h"
 #include "service/system/IConfigReadService.h"
@@ -218,36 +219,36 @@ void AlgChannelDemux::HandleStream() {
     duration_stat_.EndSample();
     if (media::ReadFrameStatus::Success != ret) {
         if (media::ReadFrameStatus::StreamEnd == ret) {
-            if ((video_read_count_ < video_repeat_count_) || (video_repeat_count_ <= 0)) {
-                LOG_INFO("{}:{} Stream End Now RepeatCount:{} Max RepeatCount:{}", kTag, channel_id_,
-                         video_read_count_, video_repeat_count_);
-                is_need_repeat_ = true;
-            } else {
-                // Do not report completion for live streams.
-                if ((!is_have_report_) && (!IsLiveStream())) {
-                    is_have_report_ = true;
-                    NotifyOnComplete();
-                }
-            }
-            if (IsLiveStream()) {
-                LOG_INFO("{}:{} Live Stream End  Need ReOpen Or Request Url", kTag, channel_id_);
-                // Live stream has no definitive end — request new URL or reopen.
-                SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadFailed);
+            const bool is_live_stream = IsLiveStream();
+            const auto disposition =
+                flow::DecideVideoEof(is_live_stream, video_read_count_, video_repeat_count_);
+            if (flow::VideoEofDisposition::Reopen == disposition) {
                 is_need_repeat_ = true;
-                action_status_  = util::ErrorEnum::DemuxReadStreamFail;
-            } else {
-                SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadEnd);
-                if (!is_need_repeat_) {
-                    action_status_ = util::ErrorEnum::DemuxStreamClosed;
+                if (is_live_stream) {
+                    LOG_INFO("{}:{} Live Stream End Need ReOpen Or Request Url", kTag, channel_id_);
+                    SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadFailed);
+                    action_status_ = util::ErrorEnum::DemuxReadStreamFail;
+                    return;
                 }
-                // When is_need_repeat_ is true the stream is about to loop —
-                // keep action_status_ as Success from the last frame read to
-                // prevent transient "取流无数据" during the loop transition.
 
-                // Signal unfinished recording tasks to finalize.
+                LOG_INFO("{}:{} Stream End Now RepeatCount:{} Max RepeatCount:{}", kTag, channel_id_,
+                         video_read_count_, video_repeat_count_);
+                // A loop boundary is not terminal. Keep the last Reading state
+                // and successful action status until OpenStream publishes the
+                // next Opened/Reading transition.
                 frame_packet->index = -1;
                 recorder_->TaskFrame(frame_packet);
+                return;
+            }
+
+            if (!is_have_report_) {
+                is_have_report_ = true;
+                NotifyOnComplete();
             }
+            SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadEnd);
+            action_status_      = util::ErrorEnum::DemuxStreamClosed;
+            frame_packet->index = -1;
+            recorder_->TaskFrame(frame_packet);
             return;
         }
         SetStatusInfo(service::camera::AlgDemuxStatus::AlgDemuxReadFailed);

src/flow/channel/AlgChannelDemuxStream.cc

@@ -145,7 +145,8 @@ bool AlgChannelDemux::GetAttr(MsgCameraAttr& attr) {
             attr.channelStatus = ChannelStatus::ChannelStatusResolusionUnSupport;
         }
     }
-    attr.dataStatus = static_cast<int>(status_.status);
+    attr.dataStatus    = static_cast<int>(status_.status);
+    attr.repeatPending = is_need_repeat_.load();
     return true;
 }
 

src/flow/channel/VideoEofPolicy.h

@@ -0,0 +1,28 @@
+// Local-video EOF lifecycle policy shared by the demuxer and monitor guard.
+
+#pragma once
+
+namespace cosmo::flow {
+
+enum class VideoEofDisposition {
+    Reopen,
+    Complete,
+};
+
+// video_read_count is the number of successful opens, including the current
+// playback. A non-positive repeat count means infinite playback.
+constexpr VideoEofDisposition DecideVideoEof(bool is_live_stream, int video_read_count,
+                                             int video_repeat_count) {
+    if (is_live_stream || video_repeat_count <= 0 || video_read_count < video_repeat_count) {
+        return VideoEofDisposition::Reopen;
+    }
+    return VideoEofDisposition::Complete;
+}
+
+// Defense for the camera monitor: a transient ReadEnd can never be terminal
+// while the demuxer has already committed to reopening the stream.
+constexpr bool IsTerminalOfflineReadEnd(bool is_read_end, bool repeat_pending) {
+    return is_read_end && !repeat_pending;
+}
+
+}  // namespace cosmo::flow

src/service/camera/impl/CameraServiceImpl.cc

@@ -20,6 +20,7 @@
 #include <regex>
 
 #include "flow/channel/AlgChannel.h"
+#include "flow/channel/VideoEofPolicy.h"
 #include "flow/common/AlgDataRecord.h"
 #include "flow/common/FlowTaskUtil.h"
 #include "service/algorithm/IAlgorithmQuery.h"
@@ -581,11 +582,17 @@ void CameraServiceImpl::MonitorCameraEntity(const CameraEntityPtr& camera, bool
                 MsgCameraAttr attr;
                 if (ServiceRegistry::Instance().Get<ITaskChannel>().GetChannelAttr(camera->videoChannelId,
                                                                                    attr)) {
-                    bool isReadEnd =
+                    const bool isReadEnd =
                         (attr.dataStatus == static_cast<int>(camera::AlgDemuxStatus::AlgDemuxReadEnd));
+                    const bool isTerminalReadEnd =
+                        flow::IsTerminalOfflineReadEnd(isReadEnd, attr.repeatPending);
+                    if (isReadEnd && attr.repeatPending) {
+                        LOG_ERRO("[{}/{}] Refusing task auto-stop: local video is reopening",
+                                 camera->videoChannelId, task->task_id_);
+                    }
                     // Channel has finished reading and no active data remains (queue fully consumed)
-                    if (isReadEnd && !ServiceRegistry::Instance().Get<ITaskChannel>().TaskDataActive(
-                                         camera->videoChannelId)) {
+                    if (isTerminalReadEnd && !ServiceRegistry::Instance().Get<ITaskChannel>().TaskDataActive(
+                                                 camera->videoChannelId)) {
                         LOG_INFO("[{}/{}] Offline video completed, auto-stopping task to release resources",
                                  camera->videoChannelId, task->task_id_);
                         ServiceRegistry::Instance().Get<ITaskLifecycle>().TaskStop(task->task_id_);

src/util/dto/CameraMsgTypes.h

@@ -31,6 +31,9 @@ struct MsgCameraAttr {
     float fps{0.0};
     ChannelStatus channelStatus{ChannelStatus::ChannelStatusOffline};  // 0: offline, 1: online, 2: auth error
     int dataStatus{0};  // service::camera::AlgDemuxStatus status
+    // Runtime-only guard. It is intentionally not serialized into the public
+    // camera DTO because it only coordinates the demuxer and task monitor.
+    bool repeatPending{false};
     friend void to_json(nlohmann::json& j, const MsgCameraAttr& v);
     friend void from_json(const nlohmann::json& j, MsgCameraAttr& v);
 };

test/test_video_eof_policy.cc

@@ -0,0 +1,33 @@
+#include "catch_amalgamated.hpp"
+#include "flow/channel/VideoEofPolicy.h"
+
+using cosmo::flow::DecideVideoEof;
+using cosmo::flow::IsTerminalOfflineReadEnd;
+using cosmo::flow::VideoEofDisposition;
+
+TEST_CASE("Infinite local video EOF always reopens", "[video-eof][repeat]") {
+    for (int read_count = 1; read_count <= 1000; ++read_count) {
+        CHECK(DecideVideoEof(false, read_count, 0) == VideoEofDisposition::Reopen);
+    }
+}
+
+TEST_CASE("Finite local video EOF completes only after the configured playback count",
+          "[video-eof][repeat]") {
+    CHECK(DecideVideoEof(false, 1, 1) == VideoEofDisposition::Complete);
+
+    CHECK(DecideVideoEof(false, 1, 3) == VideoEofDisposition::Reopen);
+    CHECK(DecideVideoEof(false, 2, 3) == VideoEofDisposition::Reopen);
+    CHECK(DecideVideoEof(false, 3, 3) == VideoEofDisposition::Complete);
+}
+
+TEST_CASE("Live stream EOF remains a reopen condition", "[video-eof][live]") {
+    CHECK(DecideVideoEof(true, 1, 1) == VideoEofDisposition::Reopen);
+    CHECK(DecideVideoEof(true, 1000, 3) == VideoEofDisposition::Reopen);
+}
+
+TEST_CASE("Camera monitor rejects terminal interpretation while reopen is pending", "[video-eof][monitor]") {
+    CHECK_FALSE(IsTerminalOfflineReadEnd(false, false));
+    CHECK_FALSE(IsTerminalOfflineReadEnd(false, true));
+    CHECK_FALSE(IsTerminalOfflineReadEnd(true, true));
+    CHECK(IsTerminalOfflineReadEnd(true, false));
+}