fix(sse): bound heartbeat lifetime and reuse idempotent task streams

- Wait for the chat SSE heartbeat goroutine to exit before the handler
  returns, so it cannot write a recycled connection
- Extend /tasks/run write deadline to 75m so long sync runs are not cut
  by the global 120s WriteTimeout before the response is written
- Reuse an existing task run on the SSE path when the idempotency key
  already belongs to the same user/session/project
- Log stale fencing on event ack so unbounded replay is observable

Co-authored-by: Cursor <cursoragent@cursor.com>

internal/events/worker.go

@@ -3,6 +3,7 @@ package events
 import (
 	"context"
 	"errors"
+	"log/slog"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -261,6 +262,9 @@ func (d *Dispatcher) handle(ctx context.Context, lease Lease) {
 		_ = tx.Rollback(hctx)
 		if errors.Is(err, ErrStaleFencing) {
 			d.Metrics.Fenced.Add(1)
+			slog.Warn("events_stale_fencing_ack",
+				"event_id", lease.EventID, "consumer", lease.Consumer,
+				"owner", lease.Owner, "attempt", lease.Attempt)
 			return
 		}
 		_ = d.fail(ctx, lease, "TRANSIENT", true)

internal/server/handlers.go

@@ -270,6 +270,10 @@ func (s *Server) handleRunTask(w http.ResponseWriter, r *http.Request) {
 		})
 	}
 
+	// 同步任务可能远超全局 WriteTimeout（120s）。先把本响应的写期限拉长，
+	// 否则业务已执行完但客户端收不到结果、重试会表现为重复执行。
+	extendSyncTaskWriteDeadline(w)
+
 	// 1. 如果指定了 stage，则执行成熟度阶段编排
 	if req.Stage != "" {
 		stageRes, err := s.taskUseCase().ExecuteStage(r.Context(), bizdev.StageRunRequest{

internal/server/sse.go

@@ -54,6 +54,15 @@ func extendSSEWriteDeadline(w http.ResponseWriter) {
 	_ = rc.SetWriteDeadline(time.Now().Add(sseWriteDeadlineExtend))
 }
 
+// syncTaskWriteDeadline 覆盖同步 /tasks/run 的长执行窗口。全局 WriteTimeout=120s
+// 会在响应写出前截断数十分钟的阶段/任务；此处只延长该请求的写期限，不改全局值。
+const syncTaskWriteDeadline = 75 * time.Minute
+
+func extendSyncTaskWriteDeadline(w http.ResponseWriter) {
+	rc := http.NewResponseController(w)
+	_ = rc.SetWriteDeadline(time.Now().Add(syncTaskWriteDeadline))
+}
+
 func writeSSEHeartbeat(w http.ResponseWriter, flusher http.Flusher) error {
 	extendSSEWriteDeadline(w)
 	if err := (transport.Encoder{W: w}).Heartbeat(time.Now()); err != nil {
@@ -251,7 +260,10 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
 	}
 
 	heartbeatDone := make(chan struct{})
+	var heartbeatWg sync.WaitGroup
+	heartbeatWg.Add(1)
 	go func() {
+		defer heartbeatWg.Done()
 		ticker := time.NewTicker(sseHeartbeatInterval)
 		defer ticker.Stop()
 		for {
@@ -270,7 +282,11 @@ func (s *Server) handleChatStream(w http.ResponseWriter, r *http.Request) {
 			}
 		}
 	}()
-	defer close(heartbeatDone)
+	// handler 返回前先停心跳再等待其退出，避免心跳 goroutine 在连接回收后残留写
+	defer func() {
+		close(heartbeatDone)
+		heartbeatWg.Wait()
+	}()
 
 	clientTurnID := req.ClientTurnID
 	if clientTurnID == "" {
@@ -410,10 +426,36 @@ func (s *Server) handleRunTaskStream(w http.ResponseWriter, r *http.Request) {
 	taskReq := s.buildBizdevTaskRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID)
 	stageReq := s.buildBizdevStageRequest(req.TaskRunRequestDTO, req.SessionID, req.ClientTurnID, runID)
 
-	created, err := s.taskRuns.CreateAndStart(r.Context(), runID, record, taskReq, stageReq, req.WorkDir)
-	if err != nil {
-		writeSafeError(w, r, http.StatusInternalServerError, "start task run failed: "+err.Error())
-		return
+	// 幂等前置：与 REST 路径对齐，EventSource 断线重连命中已有 run 时直接复用，
+	// 避免重复执行同一 client_turn_id 的任务。
+	var reused *taskrun.RunRecord
+	if idemKey := taskrun.BuildTaskRunIdempotencyKey(req.SessionID, req.ClientTurnID); idemKey != "" && s.taskRuns != nil {
+		if existing, findErr := s.taskRuns.FindByIdempotencyKey(r.Context(), idemKey); findErr != nil {
+			writeSafeError(w, r, http.StatusInternalServerError, findErr.Error())
+			return
+		} else if existing != nil {
+			if existing.UserID != req.UserID || existing.SessionID != req.SessionID ||
+				existing.ProjectID != req.ProjectID || existing.ClientTurnID != req.ClientTurnID {
+				writeJSONResponse(w, http.StatusForbidden, map[string]string{
+					"error_code": "ERR_FORBIDDEN",
+					"message":    "幂等键已存在但不属于当前用户/会话/项目",
+				})
+				return
+			}
+			reused = existing
+		}
+	}
+
+	var created *taskrun.RunRecord
+	if reused != nil {
+		created = reused
+	} else {
+		var err error
+		created, err = s.taskRuns.CreateAndStart(r.Context(), runID, record, taskReq, stageReq, req.WorkDir)
+		if err != nil {
+			writeSafeError(w, r, http.StatusInternalServerError, "start task run failed: "+err.Error())
+			return
+		}
 	}
 	_ = includeDetails
 	_ = requestID