package server import ( "context" "fmt" "net" "net/http" "os" "path/filepath" "strconv" "sync" "time" "chat-agent-system/internal/app" "chat-agent-system/internal/app/chatflow" "chat-agent-system/internal/app/intentflow" "chat-agent-system/internal/app/stageflow" "chat-agent-system/internal/artifactvault" "chat-agent-system/internal/auth" "chat-agent-system/internal/bizdev" "chat-agent-system/internal/chat" "chat-agent-system/internal/config" "chat-agent-system/internal/coordinator" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/handoff" "chat-agent-system/internal/intent" "chat-agent-system/internal/maintenance" "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/taskrun" "chat-agent-system/internal/workspace" ) // Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务。 // Pool 与领域服务装配归属 app.Runtime；本结构仅保留 HTTP 适配所需句柄。 // 残留具体领域字段见 short-term 任务记录（增量收口，本批先去掉 *pgxpool.Pool）。 type Server struct { rt *app.Runtime cfg *config.Config sessions map[string]*chat.Session swbootRT swbootRuntime sessionsMu sync.RWMutex supplementImpact SupplementImpactService // intentPlansOverride 仅测试注入用（SetIntentPlansForTest）；为空回落 rt.IntentPlans。 intentPlansOverride intentflow.Service // testOverrides 仅供测试注入替换件；生产路径恒为 nil。 testOverrides *serverTestOverrides // 后台循环停止控制（backgroundLoop，修 Shutdown 与 loop 的 DATA RACE）。 wsGCLoop backgroundLoop reminderLoop backgroundLoop vaultLoop backgroundLoop chatTurnSweepLoop backgroundLoop chatLog chatflow.ChatLog assembler chatflow.ContextAssembler listener net.Listener lastMaintToken string wsReg *workspace.FileRegistry httpServer *http.Server planAccessFn func(ctx context.Context, projectID, userID int64) error planSessionOwnerFn func(ctx context.Context, sessionID string) (int64, error) // FAV stubs (U7 incremental): nil / off unless FAV_PUBLISHER_ENABLED. vaultReg artifactvault.Registry vaultStore *artifactvault.FilesystemStore vaultPub *artifactvault.Publisher vaultIngest *artifactvault.Ingest vaultPublisherStarted int32 loginLimiter *loginRateLimiter healthMu sync.Mutex healthCache healthSnapshot eventsAlive func() bool } // SetEventsAlive wires an optional in-process events dispatcher liveness probe for /ready. func (s *Server) SetEventsAlive(fn func() bool) { if s == nil { return } s.eventsAlive = fn } // NewFromRuntime attaches the HTTP adapter to an already-wired composition root. func NewFromRuntime(rt *app.Runtime) *Server { if rt == nil { panic("server: nil runtime") } s := &Server{ rt: rt, cfg: rt.Cfg, sessions: make(map[string]*chat.Session), wsReg: workspaceRegistryFor(rt.Cfg), // 登录/注册限流：10 次 / 5 分钟 / (IP+username) 滑动窗口 loginLimiter: newLoginRateLimiter(10, 5*time.Minute), } s.chatLog = chatflow.NewAdapter(s.ctxMgr(), s.resolveStageAccess, s.failClosedAuthz()) s.assembler = chatflow.NewAssembler(coordinator.NewContextAssembler(), s.chatLog) rt.ChatLog = s.chatLog rt.Assembler = s.assembler if s.instSvc() != nil && s.instSvc().Available() && s.svc() != nil { s.svc().SetStageAccessGate(newBizdevStageGate(s.instSvc(), s.pwSvc())) } s.initTaskRunExecutors() s.initArtifactVault() s.wireFinalArtifactHooks() s.startVaultPublisher() s.startChatTurnSweeper() // Durable notify is live via events/docnotify consumer; do not assign OnOutbox here // (see internal/app.bindDocumentNotify and unhook_test). return s } // newMinimalRuntime 构造只带测试所需字段的 app.Runtime（task_finalize 系列专用）。 func newMinimalRuntime(cfg *config.Config, svc *bizdev.Service, pwSvc *projectworkflow.Service, ctxMgr *ctxmgr.ContextManager) *app.Runtime { return &app.Runtime{ Cfg: cfg, Biz: svc, PW: pwSvc, Ctx: ctxMgr, TaskRuns: taskrun.NewManager(pwSvc, ctxMgr), } } // NewServer 构造函数（测试兼容：内部走 app.Assemble） func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { rt, err := app.Assemble(cfg, svc) if err != nil { panic("server: runtime init failed: " + err.Error()) } return NewFromRuntime(rt) } // AuthService 获取认证服务 func (s *Server) AuthService() *auth.Service { return s.authSvc() } // BizdevService 获取业务交付服务 func (s *Server) BizdevService() *bizdev.Service { return s.svc() } // TaskRuns 获取长任务管理器（测试与运维钩子）。 func (s *Server) TaskRuns() *taskrun.Manager { return s.taskRuns() } // SetIntentPlansForTest swaps the intent-plan service (Wave 2 restart/refresh harness only). func (s *Server) SetIntentPlansForTest(plans intentflow.Service) { if s == nil { return } s.intentPlansOverride = plans } // ProjectWorkflowService 获取项目多工作流控制服务 func (s *Server) ProjectWorkflowService() *projectworkflow.Service { return s.pwSvc() } // Routes 构建 HTTP 路由分发器 func (s *Server) Routes() http.Handler { mux := http.NewServeMux() // 1. 健康检查与系统自检 mux.HandleFunc("GET /health", s.handleHealth) mux.HandleFunc("GET /api/v1/health", s.handleHealth) mux.HandleFunc("GET /ready", s.handleReadiness) mux.HandleFunc("GET /api/v1/ready", s.handleReadiness) // FAV status stub (publisher/degraded default off; no ingest) mux.HandleFunc("GET /api/v1/vault/status", s.handleVaultStatus) // 2. 项目与多工作流控制面 mux.HandleFunc("POST /api/v1/projects", s.handleCreateProject) mux.HandleFunc("POST /api/v1/projects/creation-preview", s.handleProjectCreationPreview) mux.HandleFunc("GET /api/v1/project-types/catalog", s.handleProjectTypesCatalog) mux.HandleFunc("GET /api/v1/compliance-profiles/catalog", s.handleComplianceProfilesCatalog) mux.HandleFunc("GET /api/v1/projects", s.handleListProjects) mux.HandleFunc("GET /api/v1/projects/{id}", s.handleGetProjectByID) mux.HandleFunc("GET /api/v1/projects/{id}/members", s.handleProjectMembers) mux.HandleFunc("POST /api/v1/projects/{id}/members", s.handleProjectMembers) mux.HandleFunc("PUT /api/v1/projects/{id}/members", s.handleProjectMembers) mux.HandleFunc("DELETE /api/v1/projects/{id}/members", s.handleProjectMembers) mux.HandleFunc("GET /api/v1/projects/{id}/workflows", s.handleListProjectWorkflows) mux.HandleFunc("PUT /api/v1/projects/{id}/active-workflow", s.handleActivateWorkflow) mux.HandleFunc("GET /api/v1/projects/{id}/settings", s.handleGetProjectSettings) mux.HandleFunc("GET /api/v1/projects/{id}/workflow-plan/default", s.handleGetDefaultWorkflowPlan) mux.HandleFunc("PUT /api/v1/projects/{id}/workflow-plan", s.handleSaveWorkflowPlan) mux.HandleFunc("PATCH /api/v1/projects/{id}/doc-tasks/{taskKey}/duties", s.handleReassignDocTaskDuties) mux.HandleFunc("POST /api/v1/projects/{id}/activation-preview", s.handleActivationPreview) mux.HandleFunc("POST /api/v1/projects/{id}/workflow-instances/activate", s.handleActivateWorkflowInstance) mux.HandleFunc("GET /api/v1/workflow-instances/{id}", s.handleGetWorkflowInstance) mux.HandleFunc("GET /api/v1/workflow-instances/{id}/stages/{stage}/readiness", s.handleGetStageReadiness) mux.HandleFunc("POST /api/v1/workflow-instances/{id}/stages/{stage}/advance", s.handleAdvanceStage) mux.HandleFunc("GET /api/v1/workflow-instances/{id}/termination-preview", s.handleTerminationPreview) mux.HandleFunc("POST /api/v1/workflow-instances/{id}/terminate", s.handleTerminateWorkflowInstance) mux.HandleFunc("GET /api/v1/workflows", s.handleListWorkflowDefinitions) mux.HandleFunc("GET /api/v1/workflows/catalog", s.handleWorkflowCatalog) mux.HandleFunc("POST /api/v1/projects/{id}/swboot/executions", s.handleSwbootExecutions) mux.HandleFunc("GET /api/v1/projects/{id}/swboot/executions", s.handleSwbootExecutions) mux.HandleFunc("POST /api/v1/projects/{id}/swboot/bindings", s.handleSwbootBinding) mux.HandleFunc("POST /api/v1/projects/{id}/workflows/{workflow_key}/run", s.handleRunProjectWorkflow) mux.HandleFunc("GET /api/v1/projects/{id}/workflows/{workflow_key}/runs", s.handleListProjectWorkflowRuns) mux.HandleFunc("GET /api/v1/projects/{id}/lineage", s.handleGetProjectLineage) mux.HandleFunc("GET /api/v1/projects/{id}/trace", s.handleProjectTrace) mux.HandleFunc("POST /api/v1/projects/{id}/trace/index", s.handleProjectTraceIndex) mux.HandleFunc("POST /api/v1/projects/{id}/trace/analyze", s.handleProjectTraceAnalyze) mux.HandleFunc("POST /api/v1/projects/{id}/trace/edges/{edgeID}/decision", s.handleProjectTraceDecision) mux.HandleFunc("POST /api/v1/projects/{id}/trace/reopen", s.handleProjectTraceReopen) mux.HandleFunc("POST /api/v1/projects/{id}/trace/reopen/{requestID}/approve", s.handleProjectTraceReopenApprove) mux.HandleFunc("POST /api/v1/projects/{id}/trace/recheck/{markID}/clear", s.handleProjectTraceRecheckClear) mux.HandleFunc("GET /api/v1/projects/{id}/documents", s.handleListProjectWorkflowDocuments) mux.HandleFunc("GET /api/v1/projects/{id}/notifications", s.handleListProjectNotifications) mux.HandleFunc("GET /api/v1/notifications", s.handleListUserNotifications) mux.HandleFunc("GET /api/v1/supplement-impacts/{batch_id}", s.handleGetSupplementImpact) mux.HandleFunc("POST /api/v1/supplement-impacts/{batch_id}/decision", s.handleDecideSupplementImpact) // 3. 对话协同服务 (标准 JSON 阻塞) mux.HandleFunc("POST /api/v1/chat", s.handleChat) // 4. 对话协同服务 (核心 SSE 流式长连接) mux.HandleFunc("POST /api/v1/chat/stream", s.handleChatStream) mux.HandleFunc("GET /api/v1/chat/stream", s.handleChatStream) mux.HandleFunc("GET /api/v1/chat/turns/{client_turn_id}", s.handleGetChatTurn) mux.HandleFunc("GET "+intentflow.PathActivePlan, s.handleGetActiveIntentPlan) mux.HandleFunc("GET "+intentflow.PathPlanByID, s.handleGetIntentPlan) mux.HandleFunc("POST "+intentflow.PathPlanDecisions, s.handleIntentPlanDecision) // 5. 独立任务 / 阶段执行 mux.HandleFunc("POST /api/v1/tasks/run", s.handleRunTask) mux.HandleFunc("POST /api/v1/tasks/run/stream", s.handleRunTaskStream) mux.HandleFunc("GET /api/v1/tasks/run/stream", s.handleRunTaskStream) mux.HandleFunc("POST /api/v1/tasks/runs", s.handleCreateTaskRun) mux.HandleFunc("GET /api/v1/tasks/runs/{run_id}", s.handleGetTaskRun) mux.HandleFunc("GET /api/v1/tasks/runs/{run_id}/events", s.handleTaskRunEvents) mux.HandleFunc("POST /api/v1/tasks/runs/{run_id}/cancel", s.handleCancelTaskRun) mux.HandleFunc("POST /api/v1/tasks/runs/{run_id}/resume", s.handleResumeTaskRun) mux.HandleFunc("POST /api/v1/admin/"+workspace.WorkspacesDirName+"/reconcile", s.handleReconcileWorkspace) mux.HandleFunc("GET /api/v1/admin/"+workspace.WorkspacesDirName, s.handleAdminWorkspaces) mux.HandleFunc("POST /api/v1/admin/"+workspace.WorkspacesDirName+"/{id}/cleanup", s.handleAdminWorkspaceCleanup) mux.HandleFunc("POST /api/v1/projects/{id}/assets", s.handleUploadProjectAsset) mux.HandleFunc("GET /api/v1/projects/{id}/assets", s.handleListProjectAssets) mux.HandleFunc("GET /api/v1/projects/{id}/assets/{asset_id}/raw", s.handleGetProjectAssetRaw) mux.HandleFunc("POST /api/v1/projects/{id}/assets/{asset_id}/reparse", s.handleReparseProjectAsset) // 6. 多级上下文与偏好记忆管理 mux.HandleFunc("GET /api/v1/context/user", s.handleGetUserContext) mux.HandleFunc("PUT /api/v1/context/user/preference", s.handleUpdateUserPreference) mux.HandleFunc("GET /api/v1/context/project", s.handleGetProjectContext) // 7. 历史多轮会话管理与标杆案例语义检索 mux.HandleFunc("GET /api/v1/chat/sessions", s.handleListChatSessions) mux.HandleFunc("POST /api/v1/chat/sessions", s.handleUpsertChatSession) mux.HandleFunc("GET /api/v1/chat/messages", s.handleGetChatMessages) mux.HandleFunc("GET /api/v1/chat/messages/window", s.handleGetChatWindow) mux.HandleFunc("GET /api/v1/chat/messages/day", s.handleGetChatDayMessages) mux.HandleFunc("GET /api/v1/chat/dates", s.handleListChatDates) mux.HandleFunc("GET /api/v1/chat/categories", s.handleListChatCategories) mux.HandleFunc("GET /api/v1/chat/search", s.handleChatSearch) mux.HandleFunc("GET /api/v1/chat/digests", s.handleListDailyDigests) mux.HandleFunc("POST /api/v1/chat/digests/generate", s.handleGenerateDailyDigest) mux.HandleFunc("POST /api/v1/chat/messages/import", s.handleImportChatMessages) // P3: 未读提醒 & 聊天导出 mux.HandleFunc("GET /api/v1/chat/unread", s.handleChatUnread) mux.HandleFunc("POST /api/v1/chat/read", s.handleChatRead) mux.HandleFunc("GET /api/v1/chat/export/messages", s.handleChatExportMessages) mux.HandleFunc("POST /api/v1/chat/recategorize", s.handleChatRecategorize) mux.HandleFunc("GET /api/v1/context/cases/recommend", s.handleRecommendCases) // 7.1 单次助手回复导出与快照 mux.HandleFunc("GET /api/v1/chat/messages/{message_id}/export-capability", s.handleGetExportCapability) mux.HandleFunc("GET /api/v1/chat/messages/{message_id}/export-source", s.handleGetExportSource) mux.HandleFunc("POST /api/v1/chat/messages/{message_id}/pin-export", s.handlePinExport) // 7.2 自由私密文档与项目共享参考 mux.HandleFunc("POST /api/v1/freeform-documents", s.handleCreateFreeformDoc) mux.HandleFunc("POST /api/v1/freeform-documents/from-message", s.handleCreateFreeformDocFromMessage) mux.HandleFunc("GET /api/v1/freeform-documents", s.handleListFreeformDocs) mux.HandleFunc("GET /api/v1/freeform-documents/{doc_id}", s.handleGetFreeformDoc) mux.HandleFunc("PUT /api/v1/freeform-documents/{doc_id}", s.handleUpdateFreeformDoc) mux.HandleFunc("POST /api/v1/freeform-documents/{doc_id}/share", s.handleShareFreeformDoc) mux.HandleFunc("POST /api/v1/freeform-documents/{doc_id}/unshare", s.handleUnshareFreeformDoc) mux.HandleFunc("DELETE /api/v1/freeform-documents/{doc_id}", s.handleDeleteFreeformDoc) // 8. 受控文档编制、审核、批准生命周期与认领治理 mux.HandleFunc("GET /api/v1/docgov/document", s.handleDocGovGetDocument) mux.HandleFunc("GET /api/v1/docgov/documents/{doc_id}/content", s.handleDocGovGetDocumentContent) mux.HandleFunc("POST /api/v1/docgov/claim", s.handleDocGovClaim) mux.HandleFunc("POST /api/v1/docgov/submit-review", s.handleDocGovSubmitReview) mux.HandleFunc("POST /api/v1/docgov/review-decision", s.handleDocGovReviewDecision) mux.HandleFunc("POST /api/v1/docgov/approver-decision", s.handleDocGovApproverDecision) mux.HandleFunc("POST /api/v1/docgov/proposal", s.handleDocGovProposeModification) mux.HandleFunc("POST /api/v1/docgov/proposal/decide", s.handleDocGovDecideProposal) mux.HandleFunc("GET /api/v1/docgov/comments", s.handleDocGovListComments) mux.HandleFunc("POST /api/v1/docgov/comments", s.handleDocGovAddComment) mux.HandleFunc("POST /api/v1/docgov/comments/{id}/resolve", s.handleDocGovResolveComment) mux.HandleFunc("GET /api/v1/docgov/audit-logs", s.handleDocGovListAuditLogs) mux.HandleFunc("POST /api/v1/docgov/content", s.handleDocGovSaveContent) mux.HandleFunc("POST /api/v1/docgov/lease/acquire", s.handleDocGovAcquireLease) mux.HandleFunc("POST /api/v1/docgov/lease/heartbeat", s.handleDocGovHeartbeatLease) mux.HandleFunc("POST /api/v1/docgov/lease/release", s.handleDocGovReleaseLease) // 8.1 文档版本 / 修订 ledger s.registerDocumentRevisionRoutes(mux) // 9. 用户认证、安全改密与超级管理员后台接口 mux.HandleFunc("POST /api/v1/auth/login", s.handleAuthLogin) mux.HandleFunc("POST /api/v1/auth/register", s.handleAuthRegister) mux.HandleFunc("GET /api/v1/auth/public-config", s.handleAuthPublicConfig) mux.HandleFunc("POST /api/v1/auth/logout", s.handleAuthLogout) mux.HandleFunc("POST /api/v1/auth/change-password", s.handleAuthChangePassword) mux.HandleFunc("GET /api/v1/auth/me", s.handleAuthMe) mux.HandleFunc("GET /api/v1/me/profile", s.handleMeProfile) mux.HandleFunc("PATCH /api/v1/me/profile", s.handleMeProfile) mux.HandleFunc("GET /api/v1/me/preferences", s.handleMePreferences) mux.HandleFunc("PUT /api/v1/me/preferences", s.handleMePreferences) mux.HandleFunc("GET /api/v1/users/directory", s.handleUserDirectory) mux.HandleFunc("GET /api/v1/auth/sessions", s.handleAuthSessions) mux.HandleFunc("GET /api/v1/admin/permissions", s.handleAdminPermissions) mux.HandleFunc("DELETE /api/v1/auth/sessions/{id}", s.handleAuthSessionByID) mux.HandleFunc("GET /api/v1/admin/roles", s.handleAdminRoles) mux.HandleFunc("POST /api/v1/admin/roles", s.handleAdminRoles) mux.HandleFunc("PATCH /api/v1/admin/roles/{id}", s.handleAdminRoleByID) mux.HandleFunc("DELETE /api/v1/admin/roles/{id}", s.handleAdminRoleByID) mux.HandleFunc("GET /api/v1/admin/workflows", s.handleAdminWorkflows) mux.HandleFunc("POST /api/v1/admin/workflows", s.handleAdminWorkflows) mux.HandleFunc("GET /api/v1/admin/workflows/{id}", s.handleAdminWorkflowByID) mux.HandleFunc("PATCH /api/v1/admin/workflows/{id}/versions/{version}", s.handleAdminWorkflowVersion) mux.HandleFunc("POST /api/v1/admin/workflows/{id}/versions/{version}/validate", s.handleAdminWorkflowValidate) mux.HandleFunc("POST /api/v1/admin/workflows/{id}/versions/{version}/publish", s.handleAdminWorkflowPublish) mux.HandleFunc("POST /api/v1/admin/workflows/{id}/versions/{version}/retire", s.handleAdminWorkflowRetire) mux.HandleFunc("GET /api/v1/admin/templates", s.handleAdminTemplates) mux.HandleFunc("POST /api/v1/admin/templates", s.handleAdminTemplates) mux.HandleFunc("GET /api/v1/admin/templates/{key}/versions", s.handleAdminTemplateVersions) mux.HandleFunc("GET /api/v1/admin/users", s.handleAdminUsers) mux.HandleFunc("POST /api/v1/admin/users", s.handleAdminUsers) mux.HandleFunc("PUT /api/v1/admin/users/{id}/role", s.handleAdminUpdateRole) mux.HandleFunc("PUT /api/v1/admin/users/{id}/status", s.handleAdminUpdateStatus) mux.HandleFunc("PATCH /api/v1/admin/users/{id}/profile", s.handleAdminUpdateProfile) mux.HandleFunc("POST /api/v1/admin/users/{id}/reset-password", s.handleAdminResetPassword) mux.HandleFunc("GET /api/v1/admin/config", s.handleAdminConfig) mux.HandleFunc("PUT /api/v1/admin/config", s.handleAdminConfig) // 9.1 系统管理台：细粒度 system:* + 白名单设置 + 策略审计 mux.HandleFunc("GET /api/v1/admin/system/overview", s.handleAdminSystemOverview) mux.HandleFunc("GET /api/v1/admin/system/kb", s.handleAdminSystemKB) mux.HandleFunc("GET /api/v1/admin/system/features", s.handleAdminSystemFeatures) mux.HandleFunc("GET /api/v1/admin/system/settings", s.handleAdminSystemSettings) mux.HandleFunc("GET /api/v1/admin/system/settings/{key}/audits", s.handleAdminSystemSettingAudits) mux.HandleFunc("GET /api/v1/admin/system/error-events", s.handleAdminErrorEvents) mux.HandleFunc("PUT /api/v1/admin/system/settings/{key}", s.handleAdminSystemSettingByKey) mux.HandleFunc("GET /api/v1/maintenance/status", s.handleMaintenanceStatus) mux.HandleFunc("GET /api/v1/maintenance/site-status", s.handleMaintenanceSiteStatus) mux.HandleFunc("GET /api/v1/admin/maintenance", s.handleAdminMaintenanceList) mux.HandleFunc("POST /api/v1/admin/maintenance", s.handleAdminMaintenanceSchedule) mux.HandleFunc("POST /api/v1/admin/maintenance/{id}/cancel", s.handleAdminMaintenanceCancel) // 10. 现代 SPA 前端页面与静态资产托管 mux.HandleFunc("GET /", s.serveSPA) return s.corsMiddleware(s.accessLogMiddleware(s.bodyLimitMiddleware(s.maintenanceMiddleware(s.authMiddleware(mux))))) } // Start 启动 HTTP 守护进程（支持 AGENT_INHERIT_FD 继承 listener） func (s *Server) Start() error { s.startWorkspaceGC() addr := fmt.Sprintf("%s:%d", s.cfg.HTTPHost, s.cfg.HTTPPort) ln, err := handoff.ListenOrAdopt(addr) if err != nil { return err } s.listener = ln _ = handoff.SignalReady() // 若从维护公告恢复，标记窗口 ended 并清理 token s.finishMaintenanceResume() if s.maintenance() != nil { s.maintenance().SetHandoff(func(ctx context.Context, w maintenance.Window) error { return s.performMaintenanceHandoff(w) }) s.maintenance().Start(context.Background()) } s.httpServer = &http.Server{ Addr: addr, Handler: s.Routes(), ReadHeaderTimeout: 10 * time.Second, ReadTimeout: 60 * time.Second, WriteTimeout: 120 * time.Second, IdleTimeout: 120 * time.Second, } return s.httpServer.Serve(ln) } func (s *Server) finishMaintenanceResume() { if os.Getenv("MAINT_RESUME") != "1" || s.maintenance() == nil { return } id, _ := strconv.ParseInt(os.Getenv("MAINT_WINDOW_ID"), 10, 64) if id > 0 { ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() _, _ = s.maintenance().MarkEnded(ctx, id) } dataDir := config.DefaultDataDir() if s.cfg != nil && s.cfg.DataDir != "" { dataDir = s.cfg.DataDir } _ = os.Remove(filepath.Join(dataDir, "maintenance-token")) _ = os.Unsetenv("MAINT_RESUME") _ = os.Unsetenv("MAINT_WINDOW_ID") } // Shutdown 优雅停机 func (s *Server) Shutdown(ctx context.Context) error { s.wsGCLoop.stop() s.reminderLoop.stop() if s.maintenance() != nil { s.maintenance().Stop() } s.vaultLoop.stop() s.chatTurnSweepLoop.stop() if s.taskRuns() != nil { _ = s.taskRuns().WaitIdle(ctx) } if s.rt != nil { _ = s.rt.Stop(ctx) } if s.httpServer != nil { return s.httpServer.Shutdown(ctx) } return nil } // CleanStaleSessions 清理超过空闲 TTL 的内存会话 (释放内存，后续请求可通过 RestoreSessionContext 自动从 DB 恢复) func (s *Server) CleanStaleSessions(ttl time.Duration) int { s.sessionsMu.Lock() defer s.sessionsMu.Unlock() return s.cleanStaleSessionsLocked(ttl) } func (s *Server) cleanStaleSessionsLocked(ttl time.Duration) int { if ttl <= 0 { ttl = 120 * time.Minute } now := time.Now() cleaned := 0 for id, sess := range s.sessions { if sess == nil || now.Sub(sess.LastAccessed()) > ttl { delete(s.sessions, id) cleaned++ } } return cleaned } const maxLiveSessions = 4096 func (s *Server) enforceSessionCapLocked() { for len(s.sessions) >= maxLiveSessions { var oldestID string var oldest time.Time found := false for id, sess := range s.sessions { if sess == nil { delete(s.sessions, id) continue } at := sess.LastAccessed() if !found || at.Before(oldest) { found = true oldest = at oldestID = id } } if !found || oldestID == "" { return } delete(s.sessions, oldestID) } } // getOrCreateSession 获取或动态构建会话。callerUserID>0 时校验持久层归属，不匹配返回 ErrSessionOwnerMismatch。 func (s *Server) getOrCreateSession( ctx context.Context, callerUserID int64, sessionID string, projectName string, customerName string, deviceType string, intendedUse string, targetMarket string, workDir string, userName string, userRole string, ) (*chat.Session, error) { if sessionID == "" { sessionID = fmt.Sprintf("sess_%d", time.Now().UnixNano()) } // 归属校验必须先于任何缓存命中/持久层灌入 if callerUserID > 0 && s.ctxMgr() != nil { dbCtx, cancel := context.WithTimeout(ctx, 5*time.Second) owner, err := s.ctxMgr().GetSessionOwnerID(dbCtx, sessionID) cancel() if err == nil && owner > 0 && owner != callerUserID { return nil, ctxmgr.ErrSessionOwnerMismatch } } s.sessionsMu.RLock() existing, exists := s.sessions[sessionID] s.sessionsMu.RUnlock() if exists && existing != nil { if callerUserID > 0 && existing.UserID > 0 && existing.UserID != callerUserID { return nil, ctxmgr.ErrSessionOwnerMismatch } existing.TouchLastAccessed() return existing, nil } s.sessionsMu.Lock() defer s.sessionsMu.Unlock() // 双重检查 if existing, exists := s.sessions[sessionID]; exists && existing != nil { if callerUserID > 0 && existing.UserID > 0 && existing.UserID != callerUserID { return nil, ctxmgr.ErrSessionOwnerMismatch } existing.TouchLastAccessed() return existing, nil } // 惰性清理，并在仍超过硬上限时按最近访问淘汰，避免随机 session id 撑爆内存。 // 会话 id 仍由客户端提供：历史恢复按该 id 查找，不能改成只接受服务端签发。 if len(s.sessions) > 200 { s.cleanStaleSessionsLocked(120 * time.Minute) } s.enforceSessionCapLocked() // 尝试从持久层恢复会话上下文 (服务重启后跨进程保持 G3) var restored *ctxmgr.RestoredSessionContext if s.ctxMgr() != nil { dbCtx, cancel := context.WithTimeout(ctx, 5*time.Second) r, err := s.ctxMgr().RestoreSessionContext(dbCtx, sessionID) cancel() if err == nil && r != nil && len(r.Messages) > 0 { restored = r } } if userName == "" { userName = "用户" } if userRole == "" { userRole = "engineer" } u := coordinator.UserInfo{ ID: "u_" + sessionID, Name: userName, Role: userRole, ProjectRole: "composer", } var userCtx *coordinator.UserProjectContext if workDir != "" { assembler := coordinator.NewContextAssembler() userCtx, _ = assembler.AssembleFromWorkspace(workDir, u) } if userCtx == nil { if projectName == "" { projectName = "医疗器械研发项目" } if customerName == "" { customerName = "水木东方客户" } userCtx = &coordinator.UserProjectContext{ SnapshotID: fmt.Sprintf("ctx_%d", time.Now().UnixNano()), AssembledAt: time.Now(), User: u, WorkDir: workDir, Project: coordinator.ProjectInfo{ Name: projectName, CustomerName: customerName, DeviceType: deviceType, IntendedUse: intendedUse, TargetMarket: targetMarket, CurrentWorkflow: "intake", CurrentStage: bizdev.StageV01_Triage, }, BlackboardFacts: make(map[string]string), AvailableActions: coordinator.ComputeAvailableActions(bizdev.StageV01_Triage, u.ProjectRole), } if targetMarket != "" || deviceType != "" { fill := map[string]string{} if targetMarket != "" { fill["target_market"] = targetMarket } if deviceType != "" { fill["device_type"] = deviceType } _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ ExpectedVersion: userCtx.Version, BlackboardFill: fill, }) } } sess := chat.NewSession(sessionID, userCtx, s.rec(), s.engine()) sess.CtxMgr = s.ctxMgr() env := "dev" if s.cfg != nil && s.cfg.Env != "" { env = s.cfg.Env } // Adapters live in internal/app; use server fields so SetIntentPlansForTest stays effective. sess.ChatLog = app.AdaptChatMemory(s.chatLog) sess.Actions = &chat.ActionServices{ DocGov: s.gov(), TaskRuns: app.AdaptChatTaskStarter(s.taskRuns()), IntentPlans: app.AdaptChatIntentPlans(s.intentPlans()), Workflow: adaptChatWorkflow(s.pwSvc()), Environment: env, } if s.rt != nil { sess.Actions.Reopens = s.rt.ChatReopens } if s.vaultIngest != nil { ingest := s.vaultIngest sess.OnFinalArtifact = func(ctx context.Context, projectID int64, logical, eventID string, msgID int64, policy string, payload []byte) { now := time.Now().UTC() _ = ingest.AssistantReply(ctx, projectID, logical, eventID, msgID, "", policy, &now, payload) } } // 若从持久层恢复了会话历史，回填消息列表、事实黑板与近期对话 (G3)。 // 正文注入走 ChatLog.Restore，避免把未授权消息写进 Prompt 窗口。 if restored != nil { if restored.UserID > 0 { sess.UserID = restored.UserID } if restored.ProjectID > 0 { sess.SetProjectID(restored.ProjectID) } if len(restored.Facts) > 0 { _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ ExpectedVersion: userCtx.Version, BlackboardFill: restored.Facts, }) } if s.chatLog == nil { for _, rm := range restored.Messages { sess.AppendMessage(chat.Message{ ID: fmt.Sprintf("msg_%d", rm.ID), Role: rm.Role, Content: rm.Content, Intent: intent.IntentType(rm.Intent), Branch: intent.ResolutionBranch(rm.Branch), Timestamp: rm.CreatedAt, }) } sess.SyncRecentDialogue() } if sess.UserID > 0 && sess.CtxMgr != nil { dbCtx, cancel := context.WithTimeout(ctx, 5*time.Second) guidance, err := sess.CtxMgr.CompilePromptGuidance(dbCtx, sess.UserID) cancel() if err == nil && guidance != "" { _ = coordinator.ApplyContextUpdate(userCtx, coordinator.ContextUpdate{ ExpectedVersion: userCtx.Version, SetPersonalizedGuidance: true, PersonalizedGuidance: guidance, }) } } } // 新建会话时立即绑定调用者。NewSession 历史占位 UserID=1/ProjectID=1 不能当作已绑定归属。 if restored == nil { sess.SetProjectID(0) if callerUserID > 0 { sess.UserID = callerUserID } } else if restored.UserID <= 0 && callerUserID > 0 { sess.UserID = callerUserID } sess.TouchLastAccessed() s.sessions[sessionID] = sess return sess, nil } func (s *Server) hydrateVisibleHistory(ctx context.Context, sess *chat.Session, viewerUserID int64) { if s.chatLog == nil || sess == nil || viewerUserID <= 0 { return } if len(sess.SnapshotMessages()) > 0 { return } restored, err := s.chatLog.Restore(ctx, chatflow.Principal{UserID: viewerUserID, Action: chatflow.ActionAssemble}, sess.ID) if err != nil || restored == nil { return } for _, rm := range restored.Messages { sess.AppendMessage(chat.Message{ ID: fmt.Sprintf("msg_%d", rm.ID), Role: rm.Role, Content: rm.Content, Intent: intent.IntentType(rm.Intent), Branch: intent.ResolutionBranch(rm.Branch), Timestamp: rm.CreatedAt, }) } sess.SyncRecentDialogue() } func (s *Server) taskUseCase() *stageflow.UseCase { if s == nil { return stageflow.New(nil) } if s.stageflow() != nil { return s.stageflow() } return stageflow.New(s.svc()) } func (s *Server) corsMiddleware(next http.Handler) http.Handler { return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { applySecurityHeaders(w) s.writeCORS(w, r) w.Header().Set("Access-Control-Allow-Methods", "GET, POST, OPTIONS, PUT, DELETE") w.Header().Set("Access-Control-Allow-Headers", "Content-Type, Authorization, Accept, X-Session-Token, X-Requested-With") if r.Method == http.MethodOptions { w.WriteHeader(http.StatusNoContent) return } next.ServeHTTP(w, r) }) }