local-agent/server/internal/api/handlers/camera_handler.go

@@ -1,14 +1,12 @@
 package handlers
 
 import (
-	"context"
 	"fmt"
 	"io"
 	"os"
 	"path/filepath"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 
 	"visiona-agent/server/internal/api/ws"
@@ -35,33 +33,8 @@ type CameraHandler struct {
 	videoFPS       float64          // target FPS
 	videoInfo      camera.VideoInfo // duration, total frames
 	activeDeviceID string           // device ID for current video session
-
-	// pendingStartCancel 取消「等 WS join room 才開跑 pipeline」的 gated-start goroutine
-	// （video-inference-stuck 修法 A2）。stopActivePipeline 會呼叫它，確保下一次上傳 /
-	// 停止時，還沒開跑的舊 pipeline 不會事後才 Start()（避免 pipeline 洩漏）。
-	pendingStartCancel context.CancelFunc
-
-	// startMu 保護「gated-start 的 check-then-act」與 handler 端 stop 對 pendingStartCancel /
-	// pipeline 的併發存取（Reviewer Major-1）。
-	//
-	// 為什麼需要：gated goroutine「檢查 startCtx.Err() → pipeline.Start()」這兩步跨 goroutine
-	// 非原子；若 handler goroutine 恰在中間呼叫 stopActivePipeline() → cancel()，舊 gated
-	// goroutine 仍可能 Start 一個已被換掉的舊 pipeline。單一 Run() goroutine 只保護 Hub 內部，
-	// 管不到 handler 端這段。用這把鎖把「二次檢查 + Start」原子化、並讓 stop 端的
-	// cancel + 換 pipeline 也在鎖內，兩者互斥。
-	//
-	// 範圍刻意只涵蓋 pendingStartCancel / pipeline 這組跨 gated-goroutine 與 handler 的共享狀態，
-	// 不擴大到 videoPath / activeSource 等其他欄位（維持原有請求序列化假設，避免無關擴大）。
-	startMu sync.Mutex
 }
 
-// waitRoomJoinTimeout 是 A2 gated-start 等待「結果 WS join inference room」的上限。
-//
-// 逾時仍會開跑 pipeline（degrade 成舊行為），確保就算 WS 因故一直沒連上，
-// 影片推論也不會永久卡住（後續有 B 的 replay 緩存兜底早期結果）。
-// 15s 足夠涵蓋 tunnel WS 握手 + 雲端 forward 的正常延遲。
-const waitRoomJoinTimeout = 15 * time.Second
-
 func NewCameraHandler(
 	cameraMgr *camera.Manager,
 	deviceMgr *device.Manager,
@@ -118,9 +91,6 @@ func (h *CameraHandler) StartPipeline(c *gin.Context) {
 		return
 	}
 
-	// 新 pipeline 前清掉此 room 的 replay 緩存，避免上一次 session 的殘留結果補送給 client。
-	h.wsHub.ClearRoomReplay("inference:" + req.DeviceID)
-
 	// Create inference result channel
 	resultCh := make(chan *driver.InferenceResult, 10)
 
@@ -136,8 +106,6 @@ func (h *CameraHandler) StartPipeline(c *gin.Context) {
 	// Start pipeline with camera as source
 	h.activeSource = h.cameraMgr
 	h.sourceType = camera.SourceCamera
-	// Major-1：h.pipeline 由 stop 端在 startMu 內存取，這裡設定 + Start 也在鎖內保持一致。
-	h.startMu.Lock()
 	h.pipeline = camera.NewInferencePipeline(
 		h.cameraMgr,
 		camera.SourceCamera,
@@ -146,7 +114,6 @@ func (h *CameraHandler) StartPipeline(c *gin.Context) {
 		resultCh,
 	)
 	h.pipeline.Start()
-	h.startMu.Unlock()
 
 	streamURL := "/api/camera/stream"
 	c.JSON(200, gin.H{
@@ -220,10 +187,6 @@ func (h *CameraHandler) UploadImage(c *gin.Context) {
 		return
 	}
 
-	// 新 pipeline 前清掉此 room 的 replay 緩存。image 只推論一次，replay 讓晚連的 WS
-	// client 仍能補到那唯一一筆結果（順帶修 image 路徑同類的早期丟棄）。
-	h.wsHub.ClearRoomReplay("inference:" + deviceID)
-
 	resultCh := make(chan *driver.InferenceResult, 10)
 
 	go func() {
@@ -236,23 +199,18 @@ func (h *CameraHandler) UploadImage(c *gin.Context) {
 
 	h.activeSource = imgSource
 	h.sourceType = camera.SourceImage
-	imgPipeline := camera.NewInferencePipeline(
+	h.pipeline = camera.NewInferencePipeline(
 		imgSource,
 		camera.SourceImage,
 		session.Driver,
 		h.streamer.FrameChannel(),
 		resultCh,
 	)
-	// Major-1：h.pipeline 由 stop 端在 startMu 內存取，這裡設定 + Start 也在鎖內保持一致。
+	h.pipeline.Start()
-	h.startMu.Lock()
-	h.pipeline = imgPipeline
-	imgPipeline.Start()
-	h.startMu.Unlock()
 
 	// Clean up result channel after pipeline completes
-	// 用 local imgPipeline（非 h.pipeline）避免 goroutine 讀共享欄位。
 	go func() {
-		<-imgPipeline.Done()
+		<-h.pipeline.Done()
 		close(resultCh)
 	}()
 
@@ -339,81 +297,29 @@ func (h *CameraHandler) UploadVideo(c *gin.Context) {
 		}
 	}()
 
-	room := "inference:" + deviceID
-	// 新一輪上傳：清掉舊的 replay 緩存，避免上一支影片的早期結果殘留補送給這次的 client。
-	h.wsHub.ClearRoomReplay(room)
-
-	pipeline := camera.NewInferencePipeline(
-		videoSource,
-		camera.SourceVideo,
-		session.Driver,
-		h.streamer.FrameChannel(),
-		resultCh,
-	)
-
 	h.activeSource = videoSource
 	h.sourceType = camera.SourceVideo
 	h.videoPath = tmpFile.Name()
 	h.videoFPS = 15
 	h.videoInfo = videoInfo
 	h.activeDeviceID = deviceID
+	h.pipeline = camera.NewInferencePipeline(
+		videoSource,
+		camera.SourceVideo,
+		session.Driver,
+		h.streamer.FrameChannel(),
+		resultCh,
+	)
+	h.pipeline.Start()
 
-	// A2（主修）：解耦「回 200」與「pipeline 開跑」。
+	// Notify frontend when video playback completes
-	//
-	// 存檔完成即可回 200，但不立刻廣播推論結果——先在背景等結果 WS join
-	// inference room，join 後（或逾時 degrade）才 pipeline.Start()。這樣影片
-	// 上傳走 localhost（極快）與結果訂閱走 tunnel WS（較慢）時序解耦後，
-	// 早期結果不會在 Hub 因 room 無 client 被靜默丟棄（root cause §2）。
-	//
-	// gated-start goroutine 用 startCtx 控制：stopActivePipeline 會 cancel 它，
-	// 確保下一次上傳 / 停止時，這個還沒開跑的 pipeline 不會事後才 Start()。
-	//
-	// Major-1：在 startMu 鎖內原子設定 pipeline + pendingStartCancel，讓後續可能併發的
-	// stop 看到一致的一對（pipeline 與其 cancel），不會讀到半設定狀態。
-	startCtx, cancelStart := context.WithCancel(context.Background())
-	h.startMu.Lock()
-	h.pipeline = pipeline
-	h.pendingStartCancel = cancelStart
-	h.startMu.Unlock()
-
 	go func() {
-		waitCtx, waitCancel := context.WithTimeout(startCtx, waitRoomJoinTimeout)
+		<-h.pipeline.Done()
-		defer waitCancel()
+		close(resultCh)
-		// 等到 room 有 client（true）或逾時（false, degrade 開跑）。
+		h.wsHub.BroadcastToRoom("inference:"+deviceID, map[string]interface{}{
-		// startCtx 被 cancel（stopActivePipeline）→ WaitForRoomClient 回 false 且
+			"type":       "pipeline_complete",
-		// startCtx.Err()!=nil，此時不可開跑（pipeline 已被換掉 / 停止）。
+			"sourceType": "video",
-		_ = h.wsHub.WaitForRoomClient(waitCtx, room)
+		})
-
-		// Major-1：把「二次檢查 startCtx.Err() → Start()」原子化。
-		// 取 startMu 後再檢查一次：若 stop 端已在等待與此刻之間 cancel 並換掉 pipeline，
-		// startCtx.Err()!=nil，放棄開跑；否則在鎖內 Start，並清掉 pendingStartCancel
-		// （已成功開跑，之後的 stop 改由 pipeline.Stop() 負責，不再靠 cancel）。
-		h.startMu.Lock()
-		if startCtx.Err() != nil {
-			h.startMu.Unlock()
-			// 已被 stopActivePipeline 取消，放棄開跑。
-			// 必須關 resultCh，否則上面的 forwarder goroutine（range resultCh）永久阻塞洩漏。
-			// pipeline 從未 Start()，不會有人寫 resultCh，close 安全。
-			close(resultCh)
-			return
-		}
-		pipeline.Start()
-		// 這個 gated goroutine 的任務已完成：清掉自己登記的 cancel。
-		// startCtx.Err()==nil 保證期間沒有 stop 介入過（stop 會 cancel），故 pendingStartCancel
-		// 必仍是自己登記的 cancelStart，直接清成 nil——之後的 stop 改由 pipeline.Stop() 負責。
-		h.pendingStartCancel = nil
-		h.startMu.Unlock()
-
-		// pipeline 跑完 → 關 resultCh、通知前端。放在開跑之後才註冊，
-		// 避免「還沒 Start 就等 Done()」永久阻塞（NewInferencePipeline 的 doneCh 尚未 close）。
-		go func() {
-			<-pipeline.Done()
-			close(resultCh)
-			h.wsHub.BroadcastToRoom(room, map[string]interface{}{
-				"type":       "pipeline_complete",
-				"sourceType": "video",
-			})
-		}()
 	}()
 
 	streamURL := "/api/camera/stream"
@@ -515,9 +421,6 @@ func (h *CameraHandler) UploadBatchImages(c *gin.Context) {
 		return
 	}
 
-	// 新 pipeline 前清掉此 room 的 replay 緩存（避免上一批殘留補送給 client）。
-	h.wsHub.ClearRoomReplay("inference:" + deviceID)
-
 	batchID := fmt.Sprintf("batch-%d", time.Now().UnixNano())
 	resultCh := make(chan *driver.InferenceResult, 10)
 
@@ -531,23 +434,18 @@ func (h *CameraHandler) UploadBatchImages(c *gin.Context) {
 
 	h.activeSource = batchSource
 	h.sourceType = camera.SourceBatchImage
-	batchPipeline := camera.NewInferencePipeline(
+	h.pipeline = camera.NewInferencePipeline(
 		batchSource,
 		camera.SourceBatchImage,
 		session.Driver,
 		h.streamer.FrameChannel(),
 		resultCh,
 	)
-	// Major-1：h.pipeline 由 stop 端在 startMu 內存取，這裡設定 + Start 也在鎖內保持一致。
+	h.pipeline.Start()
-	h.startMu.Lock()
-	h.pipeline = batchPipeline
-	batchPipeline.Start()
-	h.startMu.Unlock()
 
 	// Notify frontend when batch completes
-	// 用 local batchPipeline（非 h.pipeline）避免 goroutine 讀共享欄位。
 	go func() {
-		<-batchPipeline.Done()
+		<-h.pipeline.Done()
 		close(resultCh)
 		h.wsHub.BroadcastToRoom("inference:"+deviceID, map[string]interface{}{
 			"type":       "pipeline_complete",
@@ -605,35 +503,12 @@ func (h *CameraHandler) GetBatchImageFrame(c *gin.Context) {
 	c.Data(200, "image/jpeg", jpegData)
 }
 
-// cancelPendingStartAndStopPipeline 在 startMu 鎖內原子地：
+// stopPipelineForSeek stops the pipeline and ffmpeg process but keeps the video file.
-//  1. 取消尚未開跑的 gated-start goroutine（pendingStartCancel）
+func (h *CameraHandler) stopPipelineForSeek() {
-//  2. Stop 並清掉 h.pipeline
-//
-// 這把鎖與 gated goroutine 的「二次檢查 + Start」共用，兩者互斥（Reviewer Major-1）：
-//   - 若此函式先取鎖：cancel startCtx + 清 pipeline → gated goroutine 之後取鎖時
-//     startCtx.Err()!=nil，放棄開跑。
-//   - 若 gated goroutine 先取鎖：Start 已完成、pendingStartCancel 已清 nil → 此函式的
-//     pipeline.Stop() 負責停掉已開跑的 pipeline。
-//
-// 兩種情況都不會發生「stop 後 gated goroutine 又 Start 舊 pipeline」的洩漏。
-func (h *CameraHandler) cancelPendingStartAndStopPipeline() {
-	h.startMu.Lock()
-	defer h.startMu.Unlock()
-	if h.pendingStartCancel != nil {
-		h.pendingStartCancel()
-		h.pendingStartCancel = nil
-	}
 	if h.pipeline != nil {
 		h.pipeline.Stop()
 		h.pipeline = nil
 	}
-}
-
-// stopPipelineForSeek stops the pipeline and ffmpeg process but keeps the video file.
-func (h *CameraHandler) stopPipelineForSeek() {
-	// A2：seek 前也要原子地取消尚未開跑的 gated-start goroutine + 停 pipeline
-	// （極端情況：上傳後 WS 還沒 join 就 seek）。cancel 後該 goroutine 自行 close 原 resultCh。
-	h.cancelPendingStartAndStopPipeline()
 	if h.activeSource != nil {
 		if vs, ok := h.activeSource.(*camera.VideoSource); ok {
 			vs.CloseWithoutRemove()
@@ -644,13 +519,9 @@ func (h *CameraHandler) stopPipelineForSeek() {
 
 // stopActivePipeline stops the current pipeline and cleans up resources.
 func (h *CameraHandler) stopActivePipeline() {
-	// A2 + Major-1：先原子地取消「等 WS join 才開跑」的 gated-start goroutine + 停 pipeline，
+	if h.pipeline != nil {
-	// 確保尚未開跑的舊 pipeline 不會在此之後才 Start()。cancel 後該 goroutine 會自行
+		h.pipeline.Stop()
-	// close resultCh，不需在此處理。
+		h.pipeline = nil
-	h.cancelPendingStartAndStopPipeline()
-	// 清掉 inference room 的 replay 緩存（若有 active 影片 session）。
-	if h.activeDeviceID != "" {
-		h.wsHub.ClearRoomReplay("inference:" + h.activeDeviceID)
 	}
 	// Only close non-camera sources (camera is managed by cameraMgr)
 	if h.activeSource != nil && h.sourceType != camera.SourceCamera {
@@ -699,9 +570,6 @@ func (h *CameraHandler) SeekVideo(c *gin.Context) {
 
 	// Stop current pipeline without deleting the video file
 	h.stopPipelineForSeek()
-	// 清掉 seek 前的 replay 緩存，避免舊位置的結果被補送給 seek 後才 late-join 的 client。
-	// seek 不需 gated-start：WS client 早已 join（能觸發 seek 代表已在收結果），直接開跑。
-	h.wsHub.ClearRoomReplay("inference:" + h.activeDeviceID)
 
 	// Create new VideoSource with seek position
 	videoSource, err := camera.NewVideoSourceWithSeek(h.videoPath, h.videoFPS, req.TimeSeconds)
@@ -734,7 +602,7 @@ func (h *CameraHandler) SeekVideo(c *gin.Context) {
 	}()
 
 	h.activeSource = videoSource
-	seekPipeline := camera.NewInferencePipelineWithOffset(
+	h.pipeline = camera.NewInferencePipelineWithOffset(
 		videoSource,
 		camera.SourceVideo,
 		session.Driver,
@@ -742,16 +610,10 @@ func (h *CameraHandler) SeekVideo(c *gin.Context) {
 		resultCh,
 		frameOffset,
 	)
-	// Major-1：h.pipeline 由 stop 端在 startMu 內存取，這裡設定 + Start 也在鎖內保持一致。
+	h.pipeline.Start()
-	// seek 不走 gated-start（WS 早已 join），故不設 pendingStartCancel。
-	h.startMu.Lock()
-	h.pipeline = seekPipeline
-	seekPipeline.Start()
-	h.startMu.Unlock()
 
-	// 用 local seekPipeline（非 h.pipeline）避免 goroutine 讀共享欄位。
 	go func() {
-		<-seekPipeline.Done()
+		<-h.pipeline.Done()
 		close(resultCh)
 		h.wsHub.BroadcastToRoom("inference:"+h.activeDeviceID, map[string]interface{}{
 			"type":       "pipeline_complete",

local-agent/server/internal/api/handlers/camera_handler_gatedstart_test.go

@@ -1,168 +0,0 @@
-package handlers
-
-// camera_handler_gatedstart_test.go — video-inference-stuck 修法 A2 的協調邏輯測試
-//
-// UploadVideo 完整路徑需要真實 ffmpeg + 影片檔（VideoSource 會 spawn ffmpeg），不適合
-// 快速確定性單元測試。這裡聚焦驗證 A2 的核心「gated-start 協調契約」：
-//
-//  1. gated-start goroutine 等 inference room 有 client join 後才觸發「開跑」
-//  2. 在 client join 前呼叫 stopActivePipeline（cancel pendingStartCancel）→ 不開跑
-//
-// 用真實 ws.Hub + 與 UploadVideo 相同的 gated-start pattern（context + WaitForRoomClient）
-// 驗證行為，不牽涉 camera / ffmpeg / driver。
-
-import (
-	"context"
-	"sync"
-	"sync/atomic"
-	"testing"
-	"time"
-
-	"visiona-agent/server/internal/api/ws"
-)
-
-// runGatedStart 複製 UploadVideo 裡 gated-start goroutine 的協調骨架（不含真實 pipeline）。
-// started 在「決定開跑」時設為 1；aborted 在「因 cancel 放棄開跑」時設為 1。
-func runGatedStart(hub *ws.Hub, room string, startCtx context.Context, started, aborted *int32) {
-	go func() {
-		waitCtx, cancel := context.WithTimeout(startCtx, waitRoomJoinTimeout)
-		defer cancel()
-		_ = hub.WaitForRoomClient(waitCtx, room)
-		if startCtx.Err() != nil {
-			atomic.StoreInt32(aborted, 1)
-			return
-		}
-		atomic.StoreInt32(started, 1)
-	}()
-}
-
-func TestGatedStart_StartsAfterClientJoins(t *testing.T) {
-	hub := ws.NewHub()
-	go hub.Run()
-
-	room := "inference:DEVA"
-	var started, aborted int32
-	startCtx, cancel := context.WithCancel(context.Background())
-	defer cancel()
-
-	runGatedStart(hub, room, startCtx, &started, &aborted)
-
-	// 尚無 client → 不該開跑
-	time.Sleep(100 * time.Millisecond)
-	if atomic.LoadInt32(&started) != 0 {
-		t.Fatal("client join 前就開跑了（A2 gate 失效）")
-	}
-
-	// client join → 應開跑
-	c := &ws.Client{Send: make(chan []byte, 4)}
-	hub.RegisterSync(&ws.Subscription{Client: c, Room: room})
-
-	deadline := time.After(time.Second)
-	for atomic.LoadInt32(&started) == 0 {
-		select {
-		case <-deadline:
-			t.Fatal("client join 後 gated-start 未開跑")
-		case <-time.After(10 * time.Millisecond):
-		}
-	}
-	if atomic.LoadInt32(&aborted) != 0 {
-		t.Fatal("正常 join 不該被標記為 aborted")
-	}
-}
-
-func TestGatedStart_AbortsWhenCancelledBeforeJoin(t *testing.T) {
-	hub := ws.NewHub()
-	go hub.Run()
-
-	room := "inference:DEVB"
-	var started, aborted int32
-	startCtx, cancel := context.WithCancel(context.Background())
-
-	runGatedStart(hub, room, startCtx, &started, &aborted)
-	time.Sleep(50 * time.Millisecond)
-
-	// 模擬 stopActivePipeline：client join 前 cancel pendingStartCancel
-	cancel()
-
-	deadline := time.After(time.Second)
-	for atomic.LoadInt32(&aborted) == 0 {
-		select {
-		case <-deadline:
-			t.Fatal("cancel 後 gated-start 未放棄開跑（會洩漏 pipeline）")
-		case <-time.After(10 * time.Millisecond):
-		}
-	}
-	if atomic.LoadInt32(&started) != 0 {
-		t.Fatal("被 cancel 後不該開跑")
-	}
-}
-
-// TestGatedStart_CheckThenAct_Atomic_RealHandler — Reviewer Major-1 修復驗證。
-//
-// 用「真實 CameraHandler.startMu + pendingStartCancel + cancelPendingStartAndStopPipeline」
-// 復現原 race：gated goroutine 的「檢查 startCtx.Err() → 開跑」與 handler 端 stop
-// （鎖內 cancel）併發。
-//
-// 核心不變式（原子性保證）：gated 在鎖內若決定「開跑」（started=1），則它檢查當下
-// startCtx.Err() 必為 nil；而 stop 的 cancel 也在同一把鎖內。兩者互斥後，
-// 「gated 觀察到 Err()==nil 卻仍被 cancel 搶先」這種狀態不可能出現。
-//   - 修好前（check 與 act 之間放掉鎖 / 無鎖）：race detector 會抓到 pendingStartCancel /
-//     startCtx 的無同步併發存取；且可能出現 started=1 但 pendingStartCancel 未被正確清理。
-//   - 修好後：兩段都在 startMu 內，started 與 aborted 互斥、pendingStartCancel 狀態一致。
-//
-// 跑很多輪 + go test -race，任何原子性破綻都會被 race detector 抓到。
-func TestGatedStart_CheckThenAct_Atomic_RealHandler(t *testing.T) {
-	const iterations = 3000
-
-	for i := 0; i < iterations; i++ {
-		h := &CameraHandler{} // 只用到 startMu / pendingStartCancel / pipeline，其餘不需初始化
-
-		startCtx, cancelStart := context.WithCancel(context.Background())
-		h.startMu.Lock()
-		h.pendingStartCancel = cancelStart
-		h.startMu.Unlock()
-
-		var started, aborted int32
-
-		var wg sync.WaitGroup
-		wg.Add(2)
-
-		// gated goroutine：忠實複製生產端 UploadVideo 的鎖內原子區塊
-		//（取 startMu → 二次檢查 startCtx.Err() → 開跑 → 清 pendingStartCancel）。
-		go func() {
-			defer wg.Done()
-			h.startMu.Lock()
-			defer h.startMu.Unlock()
-			if startCtx.Err() != nil {
-				atomic.StoreInt32(&aborted, 1)
-				return // 已被 stop 取消，放棄開跑（正確）
-			}
-			atomic.StoreInt32(&started, 1)
-			// 開跑成功：清掉自己登記的 cancel（生產端相同語意）。
-			h.pendingStartCancel = nil
-		}()
-
-		// stop goroutine：走真實的 cancelPendingStartAndStopPipeline（鎖內 cancel + 清 pipeline）。
-		go func() {
-			defer wg.Done()
-			h.cancelPendingStartAndStopPipeline()
-		}()
-
-		wg.Wait()
-
-		// 不變式 1：started 與 aborted 互斥（不可能同時、也不可能都沒發生）。
-		s, a := atomic.LoadInt32(&started), atomic.LoadInt32(&aborted)
-		if s == a { // 兩者相等 → 同為 0（都沒跑）或同為 1（同時發生），都代表原子性被破壞
-			t.Fatalf("iter %d：started(%d)/aborted(%d) 非互斥，check-then-act 原子性被破壞", i, s, a)
-		}
-
-		// 不變式 2：無論哪條路徑，最終 pendingStartCancel 都應是 nil
-		//（started → gated 清 nil；aborted → stop 清 nil）。殘留非 nil 代表狀態不一致。
-		h.startMu.Lock()
-		leftover := h.pendingStartCancel != nil
-		h.startMu.Unlock()
-		if leftover {
-			t.Fatalf("iter %d：pendingStartCancel 未被清乾淨（狀態不一致）", i)
-		}
-	}
-}

local-agent/server/internal/api/ws/hub.go

@@ -1,34 +1,16 @@
 package ws
 
 import (
-	"context"
 	"encoding/json"
 	"fmt"
 	"os"
 	"path/filepath"
-	"strings"
 	"sync"
 	"time"
 
 	"github.com/gorilla/websocket"
 )
 
-// replayPrefix 決定哪些 room 啟用「late-join replay 緩存」。
-//
-// 只對推論結果 room（"inference:<serial>"）緩存最近 N 筆結果，理由：
-//   - 推論結果面走雲端 tunnel（ADR-019 混合路徑）、比上傳的 localhost 慢，
-//     上傳完成到 WS join room 之間有時間窗口（見 video-inference-stuck-rootcause.md §2）。
-//   - flash / system / device-events / server-logs 等 room 沒有這種「早期訊息會被丟」的問題，
-//     不緩存以免無謂佔記憶體。
-const replayPrefix = "inference:"
-
-// replayBufferSize 是每個 inference room 緩存的最近結果筆數上限。
-//
-// 30 筆 ≈ 15fps 影片的 2 秒；足夠覆蓋「上傳完成 → tunnel WS join room」的窗口，
-// 又不會讓記憶體膨脹（單筆 InferenceResult JSON 通常 < 幾 KB）。緩存在
-// ClearRoomReplay（pipeline 停止 / 切換時）被清掉，跟 stopActivePipeline 生命週期對齊。
-const replayBufferSize = 30
-
 type Client struct {
 	Conn *websocket.Conn
 	Send chan []byte
@@ -45,16 +27,6 @@ type RoomMessage struct {
 	Message []byte
 }
 
-// roomWaiter 由 WaitForRoomClient 註冊、在 Run() loop 內處理，
-// 確保「檢查 room 是否有 client」與「register 事件」在同一 goroutine 序列化、無 race。
-//
-// 若註冊當下 room 已有 client → 立即 close(ch)；否則存進 h.waiters，
-// 待該 room 有 client register 時 close(ch) 喚醒。
-type roomWaiter struct {
-	room string
-	ch   chan struct{}
-}
-
 // Hub 管理 WebSocket client 訂閱與訊息廣播。
 //
 // M8-4b：Hub 額外負責「第一個 client 連上時寫 sentinel file」，
@@ -64,37 +36,25 @@ type roomWaiter struct {
 // dataDir 由 main.go 在初始化 Hub 後透過 SetStartupSentinel(dataDir) 注入。
 // 若 dataDir 為空，sentinel 寫入會被跳過（單元測試或缺少資料目錄時的安全行為）。
 type Hub struct {
-	rooms       map[string]map[*Client]bool
+	rooms      map[string]map[*Client]bool
-	register    chan *Subscription
+	register   chan *Subscription
-	unregister  chan *Subscription
+	unregister chan *Subscription
-	broadcast   chan *RoomMessage
+	broadcast  chan *RoomMessage
-	waitReq     chan *roomWaiter // WaitForRoomClient 的等待請求（在 Run() 內序列化處理）
+	mu         sync.RWMutex
-	clearReplay chan string      // ClearRoomReplay 的清除請求
-	mu          sync.RWMutex
 
 	// M8-4b: 啟動 sentinel file
 	sentinelDataDir string    // <dataDir>，由 SetStartupSentinel 設定
 	sentinelOnce    sync.Once // 確保只在「第一個」client 連上時寫一次
 	bootID          string    // 寫入 sentinel 內容供 debug
-
-	// video-inference-stuck 修法 A2/B（均在 Run() goroutine 內存取、無需額外鎖）：
-	//   waiters — WaitForRoomClient 尚未被喚醒的等待者，key=room
-	//   replay  — 每個 inference room 的最近 replayBufferSize 筆訊息 ring（late-join replay）
-	waiters map[string][]*roomWaiter
-	replay  map[string][][]byte
 }
 
 func NewHub() *Hub {
 	return &Hub{
-		rooms:       make(map[string]map[*Client]bool),
+		rooms:      make(map[string]map[*Client]bool),
-		register:    make(chan *Subscription, 10),
+		register:   make(chan *Subscription, 10),
-		unregister:  make(chan *Subscription, 10),
+		unregister: make(chan *Subscription, 10),
-		broadcast:   make(chan *RoomMessage, 100),
+		broadcast:  make(chan *RoomMessage, 100),
-		waitReq:     make(chan *roomWaiter, 10),
+		bootID:     fmt.Sprintf("boot-%d", time.Now().UnixNano()),
-		clearReplay: make(chan string, 10),
-		bootID:      fmt.Sprintf("boot-%d", time.Now().UnixNano()),
-		waiters:     make(map[string][]*roomWaiter),
-		replay:      make(map[string][][]byte),
 	}
 }
 
@@ -146,30 +106,7 @@ func (h *Hub) Run() {
 				h.rooms[sub.Room] = make(map[*Client]bool)
 			}
 			h.rooms[sub.Room][sub.Client] = true
-			// B（late-join replay）：inference room 若有緩存的早期結果，
-			// 在此把它們補送給剛 join 的 client，解「WS 稍慢也不丟第一筆」。
-			// 在鎖內取出 replay 快照、鎖外送出，避免 client.Send 阻塞時卡住 Run() 。
-			var pending [][]byte
-			if buffered, ok := h.replay[sub.Room]; ok && len(buffered) > 0 {
-				pending = make([][]byte, len(buffered))
-				copy(pending, buffered)
-			}
 			h.mu.Unlock()
-			for _, msg := range pending {
-				select {
-				case sub.Client.Send <- msg:
-				default:
-					// client buffer 已滿（極罕見：replay 30 筆 > send buffer 20）→ 停止補送，
-					// 後續 live 廣播仍會照常送達，不因 replay 溢出而 drop client。
-				}
-			}
-			// A2：喚醒等待「此 room 有 client」的 waiter（WaitForRoomClient）。
-			if ws := h.waiters[sub.Room]; len(ws) > 0 {
-				for _, w := range ws {
-					close(w.ch)
-				}
-				delete(h.waiters, sub.Room)
-			}
 			// M8-4b：第一次有 client 加入任何 room → 寫 sentinel file
 			// （sync.Once 保證後續呼叫 no-op）
 			h.writeStartupSentinel()
@@ -177,26 +114,6 @@ func (h *Hub) Run() {
 				close(sub.done)
 			}
 
-		case w := <-h.waitReq:
-			// A2：WaitForRoomClient 的請求。若 room 當下已有 client → 立即喚醒；
-			// 否則存進 waiters，待 register 時喚醒。與 register 在同一 goroutine
-			// 序列化處理，故「檢查 + 掛等待」對 register 事件無 race。
-			h.mu.RLock()
-			hasClient := len(h.rooms[w.room]) > 0
-			h.mu.RUnlock()
-			if hasClient {
-				close(w.ch)
-			} else {
-				h.waiters[w.room] = append(h.waiters[w.room], w)
-			}
-
-		case room := <-h.clearReplay:
-			// pipeline 停止 / 切換時清掉該 room 的 replay 緩存（跟 stopActivePipeline 對齊），
-			// 防止上一支影片的結果殘留給下一次 join 的 client。
-			h.mu.Lock()
-			delete(h.replay, room)
-			h.mu.Unlock()
-
 		case sub := <-h.unregister:
 			h.mu.Lock()
 			if clients, ok := h.rooms[sub.Room]; ok {
@@ -208,15 +125,7 @@ func (h *Hub) Run() {
 			h.mu.Unlock()
 
 		case msg := <-h.broadcast:
-			h.mu.Lock()
+			h.mu.RLock()
-			// B：inference room 的訊息先進 replay ring（不論當下有無 client）。
-			if strings.HasPrefix(msg.Room, replayPrefix) {
-				buf := append(h.replay[msg.Room], msg.Message)
-				if len(buf) > replayBufferSize {
-					buf = buf[len(buf)-replayBufferSize:]
-				}
-				h.replay[msg.Room] = buf
-			}
 			if clients, ok := h.rooms[msg.Room]; ok {
 				for client := range clients {
 					select {
@@ -227,7 +136,7 @@ func (h *Hub) Run() {
 					}
 				}
 			}
-			h.mu.Unlock()
+			h.mu.RUnlock()
 		}
 	}
 }
@@ -255,45 +164,3 @@ func (h *Hub) BroadcastToRoom(room string, data interface{}) {
 	}
 	h.broadcast <- &RoomMessage{Room: room, Message: jsonData}
 }
-
-// HasClients 回報指定 room 當下是否有至少一個 client。
-// 用於快速判斷、不阻塞（WaitForRoomClient 才是「等到有 client」的阻塞版）。
-func (h *Hub) HasClients(room string) bool {
-	h.mu.RLock()
-	defer h.mu.RUnlock()
-	return len(h.rooms[room]) > 0
-}
-
-// WaitForRoomClient 阻塞直到 room 有至少一個 client join、或 ctx 被取消。
-//
-// 回傳 true 代表「room 已有 client」；false 代表 ctx 先結束（timeout / 上游取消）。
-//
-// A2 主修用途：UploadVideo 存檔後即回 200，但 pipeline 廣播延到「結果 WS 已 join
-// inference room」才開跑——避免上傳（localhost，快）與結果訂閱（tunnel WS，慢）
-// 時序解耦後，早期結果在 Hub 因 room 無 client 被靜默丟棄（root cause §2）。
-//
-// 實作：把等待請求丟進 Run() goroutine 序列化處理（waitReq case），
-// 確保「檢查 room 是否有 client」與「register 事件」無 race。
-// ctx 先結束時仍會殘留一個 waiter 在 h.waiters，但 register 喚醒它只是 close 一個
-// 沒人收的 channel（無害）；replay/room 生命週期短，不會累積。
-func (h *Hub) WaitForRoomClient(ctx context.Context, room string) bool {
-	w := &roomWaiter{room: room, ch: make(chan struct{})}
-	select {
-	case h.waitReq <- w:
-	case <-ctx.Done():
-		return false
-	}
-	select {
-	case <-w.ch:
-		return true
-	case <-ctx.Done():
-		return false
-	}
-}
-
-// ClearRoomReplay 清除指定 room 的 late-join replay 緩存。
-// 由 CameraHandler 在切換 / 停止 pipeline 時呼叫，跟 stopActivePipeline 生命週期對齊，
-// 防止上一支影片的結果殘留給下一次 join 的 client。非 inference room 呼叫也安全（no-op）。
-func (h *Hub) ClearRoomReplay(room string) {
-	h.clearReplay <- room
-}

local-agent/server/internal/api/ws/hub_video_race_test.go

@@ -1,202 +0,0 @@
-package ws
-
-// hub_video_race_test.go — video-inference-stuck 修法 A2/B 的 Hub 行為測試
-//
-// 涵蓋：
-//  A2  WaitForRoomClient：room 有 client 時立即返回、無 client 時阻塞到 register、ctx 取消返回 false
-//  B   late-join replay：inference room 緩存最近 N 筆、join 時補送、ClearRoomReplay 清除
-//  隔離 非 inference room（如 flash:）不緩存 replay
-
-import (
-	"context"
-	"encoding/json"
-	"testing"
-	"time"
-)
-
-// drainN 從 client.Send 收 n 筆訊息，逾時 fail。
-func drainN(t *testing.T, c *Client, n int, timeout time.Duration) [][]byte {
-	t.Helper()
-	out := make([][]byte, 0, n)
-	deadline := time.After(timeout)
-	for len(out) < n {
-		select {
-		case msg := <-c.Send:
-			out = append(out, msg)
-		case <-deadline:
-			t.Fatalf("只收到 %d/%d 筆訊息就逾時", len(out), n)
-		}
-	}
-	return out
-}
-
-func TestHub_WaitForRoomClient_ReturnsWhenClientAlreadyPresent(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	makeRegisteredClient(hub, "inference:DEV1", 4)
-
-	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
-	defer cancel()
-	if !hub.WaitForRoomClient(ctx, "inference:DEV1") {
-		t.Fatal("room 已有 client，WaitForRoomClient 應立即回 true")
-	}
-}
-
-func TestHub_WaitForRoomClient_BlocksUntilRegister(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	got := make(chan bool, 1)
-	go func() {
-		ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
-		defer cancel()
-		got <- hub.WaitForRoomClient(ctx, "inference:DEV2")
-	}()
-
-	// 確保 waiter 已掛上（尚未有 client）
-	select {
-	case <-got:
-		t.Fatal("尚無 client 時 WaitForRoomClient 不該返回")
-	case <-time.After(100 * time.Millisecond):
-	}
-
-	// 現在 register 一個 client → 應喚醒 waiter
-	makeRegisteredClient(hub, "inference:DEV2", 4)
-
-	select {
-	case ok := <-got:
-		if !ok {
-			t.Fatal("client join 後 WaitForRoomClient 應回 true")
-		}
-	case <-time.After(time.Second):
-		t.Fatal("client join 後 WaitForRoomClient 未在時限內返回")
-	}
-}
-
-func TestHub_WaitForRoomClient_ReturnsFalseOnCtxCancel(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
-	defer cancel()
-	if hub.WaitForRoomClient(ctx, "inference:NEVER") {
-		t.Fatal("無 client 且 ctx 逾時，應回 false")
-	}
-}
-
-func TestHub_LateJoinReplay_DeliversBufferedResults(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	room := "inference:DEV3"
-	// 在無 client 時廣播 3 筆（模擬「WS 連上前的早期結果」）
-	for i := 0; i < 3; i++ {
-		hub.BroadcastToRoom(room, map[string]int{"frame": i})
-	}
-	// 讓 broadcast 都被 Run() 處理完（進 replay ring）
-	time.Sleep(50 * time.Millisecond)
-
-	// 現在 client late-join → 應補送到那 3 筆
-	c := makeRegisteredClient(hub, room, 16)
-	msgs := drainN(t, c, 3, time.Second)
-
-	for i, m := range msgs {
-		var got map[string]int
-		if err := json.Unmarshal(m, &got); err != nil {
-			t.Fatalf("replay 第 %d 筆 bad json: %v", i, err)
-		}
-		if got["frame"] != i {
-			t.Errorf("replay 順序錯：第 %d 筆 frame=%d，預期 %d", i, got["frame"], i)
-		}
-	}
-}
-
-func TestHub_LateJoinReplay_CapsAtBufferSize(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	room := "inference:DEV4"
-	total := replayBufferSize + 10
-	for i := 0; i < total; i++ {
-		hub.BroadcastToRoom(room, map[string]int{"frame": i})
-	}
-	time.Sleep(80 * time.Millisecond)
-
-	c := makeRegisteredClient(hub, room, replayBufferSize+8)
-	msgs := drainN(t, c, replayBufferSize, time.Second)
-
-	// 應只保留最後 replayBufferSize 筆，第一筆 frame 應為 total-replayBufferSize
-	var first map[string]int
-	if err := json.Unmarshal(msgs[0], &first); err != nil {
-		t.Fatalf("bad json: %v", err)
-	}
-	if first["frame"] != total-replayBufferSize {
-		t.Errorf("ring 未正確截斷：首筆 frame=%d，預期 %d", first["frame"], total-replayBufferSize)
-	}
-	// 不應再有第 replayBufferSize+1 筆
-	select {
-	case extra := <-c.Send:
-		t.Errorf("replay 超出 buffer 上限，仍收到多餘訊息: %s", extra)
-	case <-time.After(150 * time.Millisecond):
-	}
-}
-
-func TestHub_ClearRoomReplay_DropsBuffer(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	room := "inference:DEV5"
-	hub.BroadcastToRoom(room, map[string]int{"frame": 0})
-	time.Sleep(50 * time.Millisecond)
-
-	hub.ClearRoomReplay(room)
-	time.Sleep(50 * time.Millisecond)
-
-	// clear 之後 late-join 不該收到任何 replay
-	c := makeRegisteredClient(hub, room, 4)
-	select {
-	case msg := <-c.Send:
-		t.Errorf("ClearRoomReplay 後仍補送 replay: %s", msg)
-	case <-time.After(150 * time.Millisecond):
-	}
-}
-
-func TestHub_NonInferenceRoom_NoReplay(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	room := "flash:DEV6" // 非 inference 前綴 → 不緩存
-	hub.BroadcastToRoom(room, map[string]string{"type": "progress"})
-	time.Sleep(50 * time.Millisecond)
-
-	c := makeRegisteredClient(hub, room, 4)
-	select {
-	case msg := <-c.Send:
-		t.Errorf("非 inference room 不應緩存 replay，卻補送: %s", msg)
-	case <-time.After(150 * time.Millisecond):
-	}
-}
-
-// TestHub_ReplayAndLiveBroadcast_Ordering：late-join client 先收 replay、再收後續 live 訊息。
-func TestHub_ReplayAndLiveBroadcast_Ordering(t *testing.T) {
-	hub := NewHub()
-	go hub.Run()
-
-	room := "inference:DEV7"
-	hub.BroadcastToRoom(room, map[string]int{"frame": 0}) // 早期（進 replay）
-	time.Sleep(50 * time.Millisecond)
-
-	c := makeRegisteredClient(hub, room, 8)
-	// join 後再來一筆 live
-	hub.BroadcastToRoom(room, map[string]int{"frame": 1})
-
-	msgs := drainN(t, c, 2, time.Second)
-	for i, m := range msgs {
-		var got map[string]int
-		_ = json.Unmarshal(m, &got)
-		if got["frame"] != i {
-			t.Errorf("順序錯：第 %d 筆 frame=%d，預期 %d（replay 應在 live 之前）", i, got["frame"], i)
-		}
-	}
-}

local-agent/start-agent.sh

@@ -35,18 +35,12 @@ case "$MODE" in
     CLOUD_API_URL="https://$HOST"
     RELAY_URL="wss://$HOST/tunnel/connect"
     WS_SCHEME="https"
-    # ADR-019 §2.5 CORS 白名單（stage 雙入口）：影片分頁 localhost 直連時，
-    # 瀏覽器可能從公網 HTTPS 或內網純 HTTP 任一入口開啟，兩者都要放行。
-    CLOUD_ORIGINS="https://stage-9527.innovedus.com:9527,http://192.168.0.130:9527"
     ;;
   internal)
     HOST="192.168.0.130:9527"
     CLOUD_API_URL="http://$HOST"
     RELAY_URL="ws://$HOST/tunnel/connect"
     WS_SCHEME="http"
-    # ADR-019 §2.5 CORS 白名單：internal 走內網純 HTTP 入口；一併帶公網入口
-    # 以便同一台 agent 兩種路徑都能開影片分頁（多帶白名單不會放寬安全性）。
-    CLOUD_ORIGINS="http://192.168.0.130:9527,https://stage-9527.innovedus.com:9527"
     ;;
   *)
     echo "用法：$0 [public|internal]" >&2
@@ -111,16 +105,6 @@ echo ""
 export VISIONA_CLOUD_API_URL="$CLOUD_API_URL"
 export VISIONA_RELAY_URL="$RELAY_URL"
 
-# ADR-019 localhost 直連的 CORS 白名單（server 讀 middleware.go:46 os.Getenv）。
-# 漏設會導致影片分頁的 preflight（OPTIONS）被 local-agent 回 403 → 前端 port 探測
-# 掃不到同機 agent → 影片分頁顯示 LOCAL_AGENT_NOT_FOUND（「需在同一台電腦操作」）。
-# 格式：逗號分隔的完整 origin（scheme+host+port 一字不差、結尾無斜線），與
-# middleware.go 的「完整 origin 精確比對」相容。
-# 尊重外部覆寫（與上方優先級 env > 內建預設一致）：若使用者已設則沿用其值。
-export VISIONA_CLOUD_ORIGINS="${VISIONA_CLOUD_ORIGINS:-$CLOUD_ORIGINS}"
-echo "  CORS 白名單（VISIONA_CLOUD_ORIGINS）：$VISIONA_CLOUD_ORIGINS"
-echo ""
-
 # public 模式：stage 用自簽憑證，agent 需明確 opt-in 跳過 TLS 驗證
 # （pairing exchange + tunnel WSS + 設定頁「測試連線」三條路徑共用此開關），
 # 否則全部 x509 失敗，與本腳本的 demo 目的自相矛盾。