update cam id

backend/app/api.py

@@ -141,10 +141,21 @@ async def health() -> HealthResponse | JSONResponse:
     summary="Demo 离线 batch（链路 3）是否可用",
     description="供 demo 页探测；始终注册，不依赖 DEMO_ORCHESTRATOR_ENABLED。",
 )
-async def recording_modes_status() -> dict:
+async def recording_modes_status(
+    surgery_id: Annotated[
+        str | None,
+        Query(
+            min_length=6,
+            max_length=6,
+            pattern=r"^\d{6}$",
+            description="可选：查询该手术号是否仍有链路 1 实时开录占用。",
+        ),
+    ] = None,
+    pipeline: Annotated[SurgeryPipeline, Depends(get_surgery_pipeline)] = ...,
+) -> dict:
     f = (settings.or_site_config_json_file or "").strip()
     enabled = bool(settings.demo_orchestrator_enabled)
-    return {
+    body: dict = {
         "demo_recording_modes_enabled": enabled,
         "orchestrator_enabled": enabled,
         "offline_batch_method": "POST",
@@ -167,6 +178,12 @@ async def recording_modes_status() -> dict:
         "rtsp_segment_ttl_hours": float(settings.rtsp_segment_ttl_hours),
         "video_batch_vis_ttl_hours": float(bp.VIDEO_BATCH_VIS_TTL_HOURS),
     }
+    if surgery_id is not None:
+        phase = pipeline.live_recording_phase(surgery_id)
+        body["surgery_id"] = surgery_id
+        body["live_recording_phase"] = phase
+        body["live_recording_active"] = phase is not None
+    return body
 
 
 class HlsPreviewEnsureRequest(BaseModel):

backend/app/resources/or_site_config.sample.json

@@ -1,8 +1,8 @@
 {
   "video_rtsp_urls": {
     "or-cam-01": "rtsp://admin:Aa183137@192.168.3.2:554/Streaming/Channels/101",
-    "or-cam-02": "rtsp://admin:Aa183137@192.168.3.3:554/Streaming/Channels/101",
-    "or-cam-03": "rtsp://admin:Aa183137@192.168.3.4:554/Streaming/Channels/101",
+    "or-cam-03": "rtsp://admin:Aa183137@192.168.3.3:554/Streaming/Channels/101",
+    "or-cam-02": "rtsp://admin:Aa183137@192.168.3.4:554/Streaming/Channels/101",
     "or-cam-04": "rtsp://admin:Aa183137@192.168.3.5:554/Streaming/Channels/101"
   },
   "voice_or_room_bindings": [

backend/app/routers/recording_demo.py

@@ -148,6 +148,21 @@ async def offline_batch(
         )
     candidates = normalize_candidate_consumables_raw(candidates)
 
+    live_phase = pipeline.live_recording_phase(surgery_id)
+    if live_phase is not None:
+        raise HTTPException(
+            status_code=status.HTTP_409_CONFLICT,
+            detail={
+                "code": "LIVE_RECORDING_ACTIVE",
+                "message": (
+                    "该手术号仍有链路 1 实时开录（RTSP 切片）进行中，与离线精确互斥。"
+                    "请先 POST /client/surgeries/end 结束实时会话，或更换 surgery_id 后再上传。"
+                ),
+                "surgery_id": surgery_id,
+                "live_recording_phase": live_phase,
+            },
+        )
+
     raw = await video1.read()
     if not raw:
         raise HTTPException(

backend/app/services/surgery_pipeline.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import base64
+from typing import Literal
 
 from sqlalchemy.ext.asyncio import async_sessionmaker
 
@@ -84,6 +85,10 @@ class SurgeryPipeline:
         except SurgeryPipelineError:
             raise
 
+    def live_recording_phase(self, surgery_id: str) -> Literal["starting", "recording"] | None:
+        """链路 1 实时会话是否占用该 surgery_id（starting=开录中，recording=已开录）。"""
+        return self._sessions.active_recording_phase(surgery_id)
+
     async def save_video_batch_result(
         self,
         surgery_id: str,

backend/tests/test_algo_host_batch.py

@@ -626,6 +626,10 @@ def test_demo_video_batch_endpoint_writes_queryable_result(
         def __init__(self) -> None:
             self.rows: dict[str, list[SurgeryConsumptionStored]] = {}
 
+        def live_recording_phase(self, surgery_id: str) -> None:
+            del surgery_id
+            return None
+
         async def save_video_batch_result(
             self,
             surgery_id: str,
@@ -729,6 +733,10 @@ def test_demo_video_batch_endpoint_stages_vis_and_purges_cache_when_requested(
             vis_calls.append((surgery_id, video_path, result_path))
 
     class _FakePipeline:
+        def live_recording_phase(self, surgery_id: str) -> None:
+            del surgery_id
+            return None
+
         async def save_video_batch_result(
             self,
             surgery_id: str,

backend/tests/test_recording_modes_status.py

@@ -1,14 +1,30 @@
 """GET /internal/demo/recording-modes-status 契约。"""
 
+from __future__ import annotations
+
+from typing import Literal
+
 from fastapi import FastAPI
 from fastapi.testclient import TestClient
 
 from app.api import router as api_router
+from app.dependencies import get_surgery_pipeline
+from app.routers import recording_demo
+
+
+class _FakePipeline:
+    def __init__(self, *, phase: Literal["starting", "recording"] | None = None) -> None:
+        self._phase = phase
+
+    def live_recording_phase(self, surgery_id: str) -> Literal["starting", "recording"] | None:
+        del surgery_id
+        return self._phase
 
 
 def test_recording_modes_status_paths() -> None:
     app = FastAPI()
     app.include_router(api_router)
+    app.dependency_overrides[get_surgery_pipeline] = lambda: _FakePipeline()
     client = TestClient(app)
     res = client.get("/internal/demo/recording-modes-status")
     assert res.status_code == 200
@@ -19,3 +35,34 @@ def test_recording_modes_status_paths() -> None:
     assert body["rtsp_segment_ttl_hours"] == 24.0
     assert body["rtsp_segment_duration_sec"] == 120.0
     assert body["video_batch_vis_ttl_hours"] == 24.0
+    assert "live_recording_active" not in body
+
+
+def test_recording_modes_status_with_surgery_id() -> None:
+    app = FastAPI()
+    app.include_router(api_router)
+    app.dependency_overrides[get_surgery_pipeline] = lambda: _FakePipeline(phase="recording")
+    client = TestClient(app)
+    res = client.get("/internal/demo/recording-modes-status?surgery_id=100101")
+    assert res.status_code == 200
+    body = res.json()
+    assert body["surgery_id"] == "100101"
+    assert body["live_recording_phase"] == "recording"
+    assert body["live_recording_active"] is True
+
+
+def test_offline_batch_rejects_active_live_session(monkeypatch) -> None:
+    monkeypatch.setattr(recording_demo.settings, "demo_orchestrator_enabled", True)
+
+    app = FastAPI()
+    app.include_router(recording_demo.router)
+    app.dependency_overrides[get_surgery_pipeline] = lambda: _FakePipeline(phase="recording")
+
+    client = TestClient(app)
+    res = client.post(
+        "/internal/demo/offline-batch",
+        data={"surgery_id": "100101", "candidate_consumables_json": "[]"},
+        files={"video1": ("case.mp4", b"video-bytes", "video/mp4")},
+    )
+    assert res.status_code == 409, res.text
+    assert res.json()["detail"]["code"] == "LIVE_RECORDING_ACTIVE"