package server import ( "context" "fmt" "path/filepath" "strings" "time" "unicode" "chat-agent-system/internal/auth" "chat-agent-system/internal/bizdev" "chat-agent-system/internal/chat" "chat-agent-system/internal/coordinator" "chat-agent-system/internal/docgov" "chat-agent-system/internal/intent" "chat-agent-system/internal/projectinstance" "chat-agent-system/internal/squad" "chat-agent-system/internal/storage" "chat-agent-system/internal/stream" "chat-agent-system/internal/workflowscope" ) // bindSessionDocumentSummaryReader 将正文访问绑定到已认证的用户与项目。 // 读取前先应用与文档列表/正文 API 相同的阶段权限过滤。 func (s *Server) bindSessionDocumentSummaryReader(sess *chat.Session, user *auth.UserAccount) { if sess == nil || sess.UserCtx == nil { return } sess.UserCtx.DocumentSummaryReader = nil projectID := sess.Project() userID := int64(0) if user != nil { userID = user.ID } if projectID <= 0 || userID <= 0 || s.gov == nil { return } sess.UserCtx.DocumentSummaryReader = func(ctx context.Context, target string) (*coordinator.ReadableDocument, error) { if s.authorizer == nil { return nil, fmt.Errorf("project authorization unavailable") } if err := s.authorizer.RequireProjectAccessOrAdmin(ctx, projectID, userID, s.isSystemAdminUser(ctx, user)); err != nil { return nil, err } if strings.TrimSpace(target) == intent.TargetDocLatestUpload { return s.resolveLatestUploadDocument(ctx, projectID) } listed, err := s.gov.ListWorkflowDocumentsByProject(ctx, projectID) if err != nil { return nil, err } docs := s.filterWorkflowDocsByStageAccess(ctx, projectID, userID, listed) // 工作流作用域:仅在激活工作流的文档池内解析正文;跨流点名返回可解释的引导错误。 // 激活键由 enrichSessionWorkflowStage 每轮刷新;任务集合按 live 实例快照解析。 workflowKey, scopeTaskKeys := s.activeWorkflowScopeForDocs(ctx, projectID) if sess.UserCtx != nil && sess.UserCtx.Project.ActiveWorkflowKey != "" { workflowKey = sess.UserCtx.Project.ActiveWorkflowKey } if workflowscope.CurrentMode() == workflowscope.ModeWeighted && len(scopeTaskKeys) > 0 { // 权重模式把被阶段过滤掉的他流台账加回候选,胜出后再做查看权限校验。 docs = append(append([]docgov.WorkflowDocumentSummary{}, docs...), filterDocsOutsideTaskKeys(listed, scopeTaskKeys)...) } doc, err := selectReadableWorkflowDocument(docs, target, workflowKey, scopeTaskKeys) if err != nil { return nil, err } owner := intent.TaskKeyWorkflowOwner(doc.TaskKey) cross := workflowKey != "" && owner != "" && owner != workflowKey && (len(scopeTaskKeys) == 0 || !projectinstance.TaskKeyInScope(scopeTaskKeys, doc.TaskKey)) if cross && !s.crossWorkflowViewAllowed(ctx, projectID, userID, doc.TaskKey) { title := doc.Title if title == "" { title = doc.TaskKey } return nil, &coordinator.ErrDocViewDenied{ Message: fmt.Sprintf("您无权查看【%s】的《%s》。", intent.WorkflowDisplayName(owner), title), } } content, err := s.loadControlledDocumentContent(ctx, doc.DocID) if err != nil { return nil, err } return &coordinator.ReadableDocument{ DocID: doc.DocID, Title: doc.Title, TaskKey: doc.TaskKey, LifecycleState: string(doc.LifecycleState), Version: doc.Version, Markdown: content.Markdown, IsTemplateOnly: doc.IsTemplateOnly || content.IsTemplateOnly, WorkflowKey: owner, ActiveWorkflow: workflowKey, CrossWorkflow: cross, }, nil } } // resolveLatestUploadDocument 读取项目内最新上传的 CAS 资产正文(供 @latest_upload 哨兵)。 func (s *Server) resolveLatestUploadDocument(ctx context.Context, projectID int64) (*coordinator.ReadableDocument, error) { if s == nil || s.cas == nil { return nil, &coordinator.ErrNoUploadAsset{ Message: "当前项目还没有上传过附件。请先在对话中上传文件后再试总结。", } } refs, err := s.cas.ListProjectRefs(ctx, projectID) if err != nil { return nil, err } latest := pickLatestFileReference(refs) if latest == nil { return nil, &coordinator.ErrNoUploadAsset{ Message: "当前项目还没有上传过附件。请先在对话中上传文件后再试总结。", } } matIdx := bizdev.Snapshot(s.svc).MaterialIndex state := "" if matIdx != nil { state = strings.TrimSpace(matIdx.GetParseState(ctx, latest.BlobSHA256)) } switch state { case "failed": return nil, &coordinator.ErrUploadParseState{ State: state, Filename: latest.FileName, } case "pending", "running": limit := 90 * time.Second if s.cfg != nil && s.cfg.MineruTimeout > 0 { limit = s.cfg.MineruTimeout } if matIdx != nil { if w := matIdx.ParseWaitLimit(); w > 0 { limit = w } } finalState := waitForUploadParse(ctx, func() string { if matIdx == nil { return "" } return strings.TrimSpace(matIdx.GetParseState(ctx, latest.BlobSHA256)) }, latest.FileName, limit) switch finalState { case "done": // continue to load content case "failed": return nil, &coordinator.ErrUploadParseState{ State: "failed", Filename: latest.FileName, } case "slow", "pending", "running": return nil, &coordinator.ErrUploadParseState{ State: "slow", Filename: latest.FileName, Message: fmt.Sprintf( "《%s》解析耗时较长(仍在后台进行),请稍后再让我总结。", defaultUploadTitle(latest.FileName), ), } default: if ctx.Err() != nil { return nil, ctx.Err() } // unknown / empty after wait: fall through and try EnsureParsed } } data, ref, err := s.cas.ReadRefBytes(ctx, latest.ID) if err != nil || ref == nil { return nil, fmt.Errorf("load latest upload asset: %w", err) } markdown := "" if matIdx != nil { md, _, parseErr := matIdx.EnsureParsed(ctx, ref.FileName, ref.BlobSHA256, data) if parseErr != nil { return nil, parseErr } markdown = strings.TrimSpace(md) } else { md, extractErr := squad.ExtractTextWithAnchors(ref.FileName, data) if extractErr != nil { return nil, extractErr } markdown = strings.TrimSpace(md) } if markdown == "" { return nil, &coordinator.ErrUploadParseState{ State: "empty", Filename: ref.FileName, Message: fmt.Sprintf("《%s》暂无可提取的正文,无法总结。", defaultUploadTitle(ref.FileName)), } } title := defaultUploadTitle(ref.FileName) return &coordinator.ReadableDocument{ DocID: fmt.Sprintf("asset:%d", ref.ID), Title: title, TaskKey: "uploaded_material", LifecycleState: "uploaded", Version: 1, Markdown: markdown, IsTemplateOnly: false, }, nil } var uploadParsePollInterval = 2 * time.Second // waitForUploadParse 在同轮对话内有界等待 MinerU 解析,并周期性发 document_parsing phase。 // stateFn 返回当前解析状态;终态 "done"/"failed" 立即返回;超时返回 "slow"。 func waitForUploadParse(ctx context.Context, stateFn func() string, fileName string, limit time.Duration) string { if limit <= 0 { limit = 90 * time.Second } title := defaultUploadTitle(fileName) started := time.Now() emitUploadParsePhase(ctx, title, 0, 5) ticker := time.NewTicker(uploadParsePollInterval) defer ticker.Stop() deadline := time.NewTimer(limit) defer deadline.Stop() for { state := "" if stateFn != nil { state = strings.TrimSpace(stateFn()) } switch state { case "done", "failed": return state } select { case <-ctx.Done(): return state case <-deadline.C: final := "" if stateFn != nil { final = strings.TrimSpace(stateFn()) } if final == "done" || final == "failed" { return final } return "slow" case <-ticker.C: elapsed := time.Since(started) progress := int(elapsed.Seconds() / limit.Seconds() * 100) if progress > 95 { progress = 95 } if progress < 5 { progress = 5 } emitUploadParsePhase(ctx, title, int(elapsed.Seconds()), progress) } } } func emitUploadParsePhase(ctx context.Context, title string, elapsedSec, progress int) { st := stream.FromContext(ctx) if st == nil { return } msg := fmt.Sprintf("《%s》已排队解析,正在等待 MinerU…", title) if elapsedSec > 0 { msg = fmt.Sprintf("《%s》正在解析中…已等待 %ds", title, elapsedSec) } st.EmitPhase("document_parsing", msg, "assistant", 0, 0, progress) } func pickLatestFileReference(refs []*storage.FileReference) *storage.FileReference { var latest *storage.FileReference for _, ref := range refs { if ref == nil { continue } if latest == nil { latest = ref continue } if ref.CreatedAt.After(latest.CreatedAt) || (ref.CreatedAt.Equal(latest.CreatedAt) && ref.ID > latest.ID) { latest = ref } } return latest } func defaultUploadTitle(fileName string) string { base := filepath.Base(strings.TrimSpace(fileName)) if base == "" || base == "." { return "上传附件" } stem := strings.TrimSuffix(base, filepath.Ext(base)) if strings.TrimSpace(stem) == "" { return base } return stem } // activeWorkflowScopeForDocs 读取项目激活工作流键与任务集合;失败时返回空(= 不过滤)。 func (s *Server) activeWorkflowScopeForDocs(ctx context.Context, projectID int64) (string, map[string]bool) { if s == nil || s.instSvc == nil || !s.instSvc.Available() || projectID <= 0 { return "", nil } workflowKey, taskKeys, err := s.instSvc.ActiveWorkflowScope(ctx, projectID) if err != nil { return "", nil } return workflowKey, taskKeys } // taskKeyOutsideActiveWorkflow 判断任务键是否落在激活工作流之外。 // 未激活 / 任务键未知归属时放行(存量兼容);跨流时返回引导话术。 func (s *Server) taskKeyOutsideActiveWorkflow(ctx context.Context, projectID int64, taskKey string) (bool, string) { taskKey = strings.TrimSpace(taskKey) if taskKey == "" || projectID <= 0 { return false, "" } workflowKey, scopeTaskKeys := s.activeWorkflowScopeForDocs(ctx, projectID) if workflowKey == "" { return false, "" } if len(scopeTaskKeys) > 0 { if projectinstance.TaskKeyInScope(scopeTaskKeys, taskKey) { return false, "" } owner := intent.TaskKeyWorkflowOwner(taskKey) if workflowscope.BlocksWrite(workflowKey, owner, taskKey) { return true, intent.WorkflowScopeGuidanceQuestion(taskKey, []string{owner}, workflowKey) } return false, "" } owner := intent.TaskKeyWorkflowOwner(taskKey) if workflowscope.BlocksWrite(workflowKey, owner, taskKey) { return true, intent.WorkflowScopeGuidanceQuestion(taskKey, []string{owner}, workflowKey) } return false, "" } // workflowKeyForDocument 为列表项标注归属工作流:优先激活流任务集合,其次任务键前缀推断。 func workflowKeyForDocument(taskKey, activeKey string, scopeTaskKeys map[string]bool) string { if activeKey != "" && len(scopeTaskKeys) > 0 && projectinstance.TaskKeyInScope(scopeTaskKeys, taskKey) { return activeKey } if owner := intent.TaskKeyWorkflowOwner(taskKey); owner != "" { return owner } return activeKey } // errDocWrongWorkflow 目标文档存在但不属于当前激活工作流。 type errDocWrongWorkflow struct { Title string TaskKey string OwnerWorkflow string ActiveWorkflow string } func (e *errDocWrongWorkflow) Error() string { if e == nil { return "document belongs to another workflow" } owners := []string{} if e.OwnerWorkflow != "" { owners = []string{e.OwnerWorkflow} } doc := e.Title if doc == "" { doc = e.TaskKey } return intent.WorkflowScopeGuidanceQuestion(doc, owners, e.ActiveWorkflow) } func (e *errDocWrongWorkflow) AsCoordinatorErr() error { return &coordinator.ErrWrongWorkflow{Message: e.Error()} } // selectReadableWorkflowDocument 在可读文档池中解析目标。 // scopeTaskKeys 非空时先在激活流池内匹配;未命中但精确/归一化匹配到被剔除的他流文档时返回 ErrWrongWorkflow。 // scopeTaskKeys 为空(未激活 / planning)时保持现状:在全量池内匹配。 func selectReadableWorkflowDocument( docs []docgov.WorkflowDocumentSummary, target string, activeWorkflow string, scopeTaskKeys map[string]bool, ) (*docgov.WorkflowDocumentSummary, error) { if workflowscope.CurrentMode() == workflowscope.ModeWeighted && activeWorkflow != "" && len(scopeTaskKeys) > 0 && strings.TrimSpace(target) != "" { return selectReadableWeighted(docs, target, activeWorkflow, scopeTaskKeys) } return selectReadableStrict(docs, target, activeWorkflow, scopeTaskKeys) } func selectReadableWeighted( docs []docgov.WorkflowDocumentSummary, target string, activeWorkflow string, scopeTaskKeys map[string]bool, ) (*docgov.WorkflowDocumentSummary, error) { candidates := make([]workflowscope.Candidate, 0, len(docs)) byID := make(map[string]docgov.WorkflowDocumentSummary, len(docs)) for _, doc := range docs { quality := workflowscope.BestQuality(target, doc.DocID, doc.TaskKey, doc.Title) if quality <= 0 { continue } owner := "" if projectinstance.TaskKeyInScope(scopeTaskKeys, doc.TaskKey) { owner = activeWorkflow } else if inferred := intent.TaskKeyWorkflowOwner(doc.TaskKey); inferred != "" { owner = inferred } id := doc.DocID if id == "" { id = doc.TaskKey } candidates = append(candidates, workflowscope.Candidate{ ID: id, Label: doc.Title, Owner: owner, Quality: quality, }) byID[id] = doc } decision := workflowscope.Decide(candidates, activeWorkflow, workflowscope.ActionRead) switch decision.Kind { case workflowscope.KindInScope, workflowscope.KindCrossAllow: doc := byID[decision.Winner.ID] return &doc, nil case workflowscope.KindCrossBlock: doc := byID[decision.Winner.ID] owner := decision.Winner.Owner if owner == "" { owner = intent.TaskKeyWorkflowOwner(doc.TaskKey) } return nil, (&errDocWrongWorkflow{ Title: doc.Title, TaskKey: doc.TaskKey, OwnerWorkflow: owner, ActiveWorkflow: activeWorkflow, }).AsCoordinatorErr() default: return nil, fmt.Errorf("document not found in readable scope") } } func selectReadableStrict( docs []docgov.WorkflowDocumentSummary, target string, activeWorkflow string, scopeTaskKeys map[string]bool, ) (*docgov.WorkflowDocumentSummary, error) { if len(docs) == 0 { return nil, fmt.Errorf("no readable workflow documents") } targetKey := normalizeDocumentLookupKey(target) if targetKey == "" { inScope := filterDocsByTaskKeys(docs, scopeTaskKeys) pool := inScope if len(scopeTaskKeys) == 0 { pool = docs } if len(pool) == 1 { return &pool[0], nil } return nil, fmt.Errorf("document target required") } inScope := filterDocsByTaskKeys(docs, scopeTaskKeys) pool := inScope if len(scopeTaskKeys) == 0 { pool = docs } if hit := matchDocumentInPool(pool, target, targetKey); hit != nil { return hit, nil } // 激活流池未命中:检查是否精确/归一化命中被剔除的他流文档。 if len(scopeTaskKeys) > 0 { outOfScope := filterDocsOutsideTaskKeys(docs, scopeTaskKeys) if hit := matchDocumentInPool(outOfScope, target, targetKey); hit != nil { owner := intent.TaskKeyWorkflowOwner(hit.TaskKey) return nil, (&errDocWrongWorkflow{ Title: hit.Title, TaskKey: hit.TaskKey, OwnerWorkflow: owner, ActiveWorkflow: activeWorkflow, }).AsCoordinatorErr() } } return nil, fmt.Errorf("document not found in readable scope") } func filterDocsByTaskKeys(docs []docgov.WorkflowDocumentSummary, taskKeys map[string]bool) []docgov.WorkflowDocumentSummary { if len(taskKeys) == 0 { return docs } out := make([]docgov.WorkflowDocumentSummary, 0, len(docs)) for _, d := range docs { if projectinstance.TaskKeyInScope(taskKeys, d.TaskKey) { out = append(out, d) } } return out } func filterDocsOutsideTaskKeys(docs []docgov.WorkflowDocumentSummary, taskKeys map[string]bool) []docgov.WorkflowDocumentSummary { if len(taskKeys) == 0 { return nil } out := make([]docgov.WorkflowDocumentSummary, 0) for _, d := range docs { if !projectinstance.TaskKeyInScope(taskKeys, d.TaskKey) { out = append(out, d) } } return out } func matchDocumentInPool(docs []docgov.WorkflowDocumentSummary, target, targetKey string) *docgov.WorkflowDocumentSummary { for i := range docs { if target == docs[i].DocID || target == docs[i].TaskKey || target == docs[i].Title { return &docs[i] } } for i := range docs { keys := []string{ normalizeDocumentLookupKey(docs[i].DocID), normalizeDocumentLookupKey(docs[i].TaskKey), normalizeDocumentLookupKey(docs[i].Title), } for _, key := range keys { if key != "" && (key == targetKey || strings.Contains(key, targetKey) || strings.Contains(targetKey, key)) { return &docs[i] } } } return nil } func normalizeDocumentLookupKey(value string) string { value = strings.ToLower(strings.TrimSpace(value)) value = strings.NewReplacer("收集", "", "采集", "", "正文", "", "文档", "", "《", "", "》", "").Replace(value) return strings.Map(func(r rune) rune { if unicode.IsSpace(r) || unicode.IsPunct(r) { return -1 } return r }, value) }