merge: fix/audit-b3-sse-events (audit batch 3: sse heartbeat, task idempotency, write deadline, fencing log)

deploy/postgres/goose/00001_baseline.sql

@@ -2,6 +2,9 @@
 -- =============================================================================
 -- 医疗器械商务与研发智能参谋系统 - 专属记忆与多级上下文数据库初始化脚本
 -- 运行环境：开发环境 (5433:5432) | 镜像：pgvector/pgvector:pg16
+-- 【guard 2026-09-19 审计 H7】baseline 无 down 迁移（不可回退，属预期）；
+-- 00001–00009 均为早期基线，已应用库请勿修改本文件（goose 校验和不可变）。
+-- 新增长型大表索引迁移须使用 -- +goose NO TRANSACTION + CREATE INDEX CONCURRENTLY。
 -- 涵盖范围：
 --   1. 单用户偏好记忆 (User Preferences & Habits)
 --   2. 多用户与项目级单一事实黑板 (Project Blackboards & Decision Audit)

deploy/postgres/goose/00012_2026-09-11-seed-ten-projects-chat.sql

@@ -1,6 +1,10 @@
 -- +goose Up
 -- ==============================================================================
 -- 迁移脚本: 10 个评审项目会话与种子消息持久化导入
+-- 【guard 2026-09-19 审计 H7】本迁移为 seed 导入，含对 30 个 session_id 的
+-- 无条件 DELETE（chat_messages/chat_read_positions）。仅可在全新/演示库执行；
+-- 若这些 session 已有真实用户消息将被静默清空。goose 已应用库请勿修改本文件
+-- （校验和不可变）；新库执行前确认目标 session 无真实数据。
 -- ==============================================================================
 BEGIN;
 

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