diff --git a/deploy/postgres/goose/00001_baseline.sql b/deploy/postgres/goose/00001_baseline.sql index 7e006961..ca896035 100644 --- a/deploy/postgres/goose/00001_baseline.sql +++ b/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) diff --git a/deploy/postgres/goose/00012_2026-09-11-seed-ten-projects-chat.sql b/deploy/postgres/goose/00012_2026-09-11-seed-ten-projects-chat.sql index a7ce2956..3d417202 100644 --- a/deploy/postgres/goose/00012_2026-09-11-seed-ten-projects-chat.sql +++ b/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; diff --git a/internal/events/worker.go b/internal/events/worker.go index 4ec0cd7d..56df8ad3 100644 --- a/internal/events/worker.go +++ b/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) diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 692e50d1..a8c6b16a 100644 --- a/internal/server/handlers.go +++ b/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{ diff --git a/internal/server/sse.go b/internal/server/sse.go index 84d066c9..e079983c 100644 --- a/internal/server/sse.go +++ b/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