diff --git a/.doc/system-design/module-boundaries.yaml b/.doc/system-design/module-boundaries.yaml index 02352fd5..4de7eccf 100644 --- a/.doc/system-design/module-boundaries.yaml +++ b/.doc/system-design/module-boundaries.yaml @@ -88,6 +88,7 @@ imports: - projectworkflow - storage - taskrun + - traceimpact - workflowdef - workspace - sysadmin @@ -200,6 +201,7 @@ imports: - stream/transport - swboot - taskrun + - traceimpact - upload - workflowdef - workspace @@ -251,10 +253,27 @@ imports: - app/uow sysadmin: - config + traceimpact: [] table_owners: document_records: - docgov + project_trace_items: + - traceimpact + project_trace_item_revisions: + - traceimpact + project_trace_edges: + - traceimpact + project_trace_edge_decisions: + - traceimpact + project_trace_index_state: + - traceimpact + stage_reopen_requests: + - traceimpact + stage_recheck_marks: + - traceimpact + trace_revision_proposals: + - traceimpact chat_messages: - ctxmgr chat_turns: diff --git a/internal/app/app.go b/internal/app/app.go index 9eb16531..e7a18a86 100644 --- a/internal/app/app.go +++ b/internal/app/app.go @@ -36,6 +36,7 @@ import ( "chat-agent-system/internal/storage" "chat-agent-system/internal/sysadmin" "chat-agent-system/internal/taskrun" + "chat-agent-system/internal/traceimpact" "chat-agent-system/internal/workflowdef" "chat-agent-system/internal/workspace" @@ -71,6 +72,7 @@ type Runtime struct { Pool *pgxpool.Pool Gov *docgov.GovernanceEngine Rev *docrevision.Service + Trace *traceimpact.Service Ctx *ctxmgr.ContextManager CAS *storage.CAS FreeDoc *freedoc.Service @@ -191,6 +193,28 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { if instSvc != nil && instSvc.Available() { coordEngine.SetInstanceService(instSvc) } + traceSvc := traceimpact.Open(pool) + traceSvc.SetStageReopener(traceStage{inst: instSvc}) + traceSvc.SetDocUnlocker(traceUnlock{gov: parts.Gov}) + if instSvc != nil { + instSvc.SetRecheckGate(traceRecheck{svc: traceSvc}) + } + coordEngine.SetImpactActions(traceChat{svc: traceSvc}) + if pool != nil { + access := authz.NewAuthorizer(pool) + coordEngine.SetProjectAccess(func(ctx context.Context, projectID, userID int64) error { + return access.RequireProjectAccess(ctx, projectID, userID) + }) + } + if parts.Rev != nil { + parts.Rev.CoverageResolver = func(ctx context.Context, docID string) string { + projectID := projectIDFromControlledDoc(docID) + if projectID == 0 { + return traceimpact.ImpactIndexPending + } + return traceSvc.CoverageForDoc(ctx, projectID, docID) + } + } rt := &Runtime{ Cfg: cfg, @@ -212,6 +236,7 @@ func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) { Pool: pool, Gov: parts.Gov, Rev: parts.Rev, + Trace: traceSvc, Ctx: parts.Ctx, CAS: parts.CAS, FreeDoc: parts.FreeDoc, diff --git a/internal/app/trace_bridge.go b/internal/app/trace_bridge.go new file mode 100644 index 00000000..0b682b5c --- /dev/null +++ b/internal/app/trace_bridge.go @@ -0,0 +1,98 @@ +package app + +import ( + "context" + "errors" + "fmt" + "strconv" + "strings" + + "chat-agent-system/internal/coordinator" + "chat-agent-system/internal/docgov" + "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/traceimpact" +) + +type traceRecheck struct{ svc *traceimpact.Service } + +func (t traceRecheck) StageNeedsAck(ctx context.Context, projectID int64, stageKey string) (bool, error) { + if t.svc == nil { + return false, nil + } + return t.svc.NeedsRecheckAck(ctx, projectID, stageKey) +} + +func (t traceRecheck) Acknowledge(ctx context.Context, projectID int64, stageKey string, userID int64) error { + if t.svc == nil { + return nil + } + return t.svc.AcknowledgeRecheck(ctx, projectID, stageKey, userID) +} + +type traceStage struct{ inst *projectinstance.Service } + +func (t traceStage) ReopenCompletedStage(ctx context.Context, instanceID int64, stageKey string, docTaskKeys []string, operatorID int64) error { + if t.inst == nil { + return fmt.Errorf("trace: instance service unavailable") + } + return t.inst.ReopenStage(ctx, instanceID, stageKey, docTaskKeys, operatorID) +} + +type traceUnlock struct{ gov *docgov.GovernanceEngine } + +func (t traceUnlock) ReopenApprovedDocs(ctx context.Context, projectID int64, docTaskKeys []string, actorID int64, reason string) error { + if t.gov == nil { + return nil + } + for _, key := range docTaskKeys { + docID := fmt.Sprintf("doc_proj%d_%s", projectID, key) + _, err := t.gov.ReopenForRevision(ctx, docID, actorID, "creator", reason) + if err == nil || errors.Is(err, docgov.ErrDocumentNotFound) { + continue + } + if errors.Is(err, docgov.ErrInvalidStateTransition) { + continue + } + return err + } + return nil +} + +type traceChat struct{ svc *traceimpact.Service } + +func (t traceChat) AnalyzeImpact(ctx context.Context, projectID, _ int64, text string) (string, error) { + if t.svc == nil { + return "", fmt.Errorf("trace: unavailable") + } + rep, err := t.svc.Analyze(ctx, projectID, traceimpact.DocRef{DocID: "chat", Markdown: text, Visible: true}, nil) + if err != nil { + return "", err + } + return rep.Summary, nil +} + +func projectIDFromControlledDoc(docID string) int64 { + rest := strings.TrimPrefix(docID, "doc_proj") + if rest == docID { + return 0 + } + i := strings.Index(rest, "_") + if i <= 0 { + return 0 + } + n, err := strconv.ParseInt(rest[:i], 10, 64) + if err != nil { + return 0 + } + return n +} +func (t traceChat) ProposeRevision(ctx context.Context, projectID, userID int64, text string, confirmed bool) (string, error) { + if t.svc == nil { + return "", fmt.Errorf("trace: unavailable") + } + id, err := t.svc.ProposeRevision(ctx, projectID, userID, text, confirmed) + if errors.Is(err, traceimpact.ErrNeedsConfirm) { + return "", coordinator.ErrImpactNeedsConfirm + } + return id, err +} diff --git a/internal/server/project_settings_handlers.go b/internal/server/project_settings_handlers.go index dbd26293..cf7621d7 100644 --- a/internal/server/project_settings_handlers.go +++ b/internal/server/project_settings_handlers.go @@ -194,6 +194,10 @@ func writeProjectInstanceError(w http.ResponseWriter, err error) { writeSafeError(w, nil, http.StatusConflict, `{"error":"STAGE_GATE_NOT_READY","message":"阶段门禁未满足，必需文档尚未全部批准"}`) case errors.Is(err, projectinstance.ErrStageNotActive): writeSafeError(w, nil, http.StatusConflict, `{"error":"STAGE_NOT_ACTIVE","message":"目标阶段不是当前活动阶段"}`) + case errors.Is(err, projectinstance.ErrStageNotCompleted): + writeSafeError(w, nil, http.StatusConflict, `{"error":"STAGE_NOT_COMPLETED","message":"只能再打开已完成的阶段"}`) + case errors.Is(err, projectinstance.ErrRecheckAckRequired): + writeSafeError(w, nil, http.StatusConflict, `{"error":"RECHECK_ACK_REQUIRED","message":"该阶段有待复核标记，推进前须确认已知悉"}`) case errors.Is(err, projectinstance.ErrInvalidStage): writeSafeError(w, nil, http.StatusBadRequest, `{"error":"ERR_INVALID_STAGE","message":"无效的阶段"}`) case errors.Is(err, projectinstance.ErrInstanceRevisionConflict): diff --git a/internal/server/server.go b/internal/server/server.go index dfe01d57..4df7a335 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -211,6 +211,13 @@ func (s *Server) Routes() http.Handler { 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) diff --git a/internal/server/trace_handlers.go b/internal/server/trace_handlers.go new file mode 100644 index 00000000..43fdb2e1 --- /dev/null +++ b/internal/server/trace_handlers.go @@ -0,0 +1,263 @@ +package server + +import ( + "encoding/json" + "io" + "net/http" + "strconv" + + "chat-agent-system/internal/auth" + "chat-agent-system/internal/traceimpact" +) + +func (s *Server) handleProjectTrace(w http.ResponseWriter, r *http.Request) { + projectID, user, ok := s.traceProject(w, r) + if !ok { + return + } + items, err := s.rt.Trace.ListItems(r.Context(), projectID) + if err != nil { + writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`) + return + } + visible, hidden := traceimpact.RedactItems(items, func(docID string) bool { + return s.traceDocVisible(r, projectID, user.ID, docID) + }) + index, err := s.rt.Trace.ListIndex(r.Context(), projectID) + if err != nil { + writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`) + return + } + marks, err := s.rt.Trace.ListOpenMarks(r.Context(), projectID, "") + if err != nil { + writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`) + return + } + writeJSONResponse(w, http.StatusOK, map[string]any{ + "items": visible, + "hidden_count": hidden, + "index": index, + "recheck_marks": marks, + }) +} + +func (s *Server) handleProjectTraceIndex(w http.ResponseWriter, r *http.Request) { + projectID, _, ok := s.traceProject(w, r) + if !ok { + return + } + var body traceimpact.DocRef + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`) + return + } + st, err := s.rt.Trace.IndexDocument(r.Context(), projectID, body) + if err != nil { + writeJSONResponse(w, http.StatusBadRequest, map[string]any{"error": "TRACE_INDEX_FAILED", "message": err.Error(), "index": st}) + return + } + writeJSONResponse(w, http.StatusOK, st) +} + +func (s *Server) handleProjectTraceAnalyze(w http.ResponseWriter, r *http.Request) { + projectID, _, ok := s.traceProject(w, r) + if !ok { + return + } + var body struct { + traceimpact.DocRef + Others []traceimpact.DocRef `json:"others"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`) + return + } + rep, err := s.rt.Trace.Analyze(r.Context(), projectID, body.DocRef, body.Others) + if err != nil { + writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`) + return + } + writeJSONResponse(w, http.StatusOK, rep) +} + +func (s *Server) handleProjectTraceDecision(w http.ResponseWriter, r *http.Request) { + projectID, user, ok := s.traceProject(w, r) + if !ok { + return + } + edgeID, err := strconv.ParseInt(r.PathValue("edgeID"), 10, 64) + if err != nil || edgeID <= 0 { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid edge id"}`) + return + } + var body struct { + DownstreamDoc string `json:"downstream_doc"` + DownstreamAuthor int64 `json:"downstream_author"` + Decision string `json:"decision"` + Reason string `json:"reason"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`) + return + } + if !s.traceDocVisible(r, projectID, user.ID, body.DownstreamDoc) { + writeSafeError(w, r, http.StatusForbidden, `{"error":"ERR_FORBIDDEN","message":"看不见下游文档，不能裁决"}`) + return + } + creator, _ := s.pwSvc.IsProjectCreator(r.Context(), projectID, user.ID) + if err := s.rt.Trace.DecideCandidate(r.Context(), edgeID, user.ID, creator, body.DownstreamDoc, body.DownstreamAuthor, body.Decision, body.Reason); err != nil { + writeJSONResponse(w, http.StatusForbidden, map[string]string{"error": "TRACE_DECISION_DENIED", "message": err.Error()}) + return + } + writeJSONResponse(w, http.StatusOK, map[string]string{"status": "recorded"}) +} + +func (s *Server) handleProjectTraceReopen(w http.ResponseWriter, r *http.Request) { + projectID, user, ok := s.traceProject(w, r) + if !ok { + return + } + var req traceimpact.ReopenRequest + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`) + return + } + req.ProjectID = projectID + req.RequestedBy = user.ID + creator, _ := s.pwSvc.IsProjectCreator(r.Context(), projectID, user.ID) + if !s.callerMayRequestReopen(r, projectID, user.ID, creator, req) { + writeSafeError(w, r, http.StatusForbidden, `{"error":"ERR_FORBIDDEN","message":"只有该阶段文档职责人或创建人可以申请再打开"}`) + return + } + stored, err := s.rt.Trace.RequestReopen(r.Context(), req, creator) + if err != nil { + writeJSONResponse(w, http.StatusBadRequest, map[string]string{"error": "TRACE_REOPEN_FAILED", "message": err.Error()}) + return + } + writeJSONResponse(w, http.StatusOK, stored) +} + +func (s *Server) handleProjectTraceReopenApprove(w http.ResponseWriter, r *http.Request) { + projectID, user, ok := s.traceProject(w, r) + if !ok { + return + } + creator, _ := s.pwSvc.IsProjectCreator(r.Context(), projectID, user.ID) + if !creator { + writeSafeError(w, r, http.StatusForbidden, `{"error":"ERR_NOT_PROJECT_CREATOR","message":"仅项目创建人可批准再打开"}`) + return + } + requestID, err := strconv.ParseInt(r.PathValue("requestID"), 10, 64) + if err != nil || requestID <= 0 { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid request id"}`) + return + } + existing, err := s.rt.Trace.GetReopen(r.Context(), requestID) + if err != nil || existing.ProjectID != projectID { + writeSafeError(w, r, http.StatusNotFound, `{"error":"ERR_NOT_FOUND","message":"申请不属于该项目"}`) + return + } + stored, err := s.rt.Trace.ApproveReopen(r.Context(), requestID, user.ID) + if err != nil { + writeJSONResponse(w, http.StatusConflict, map[string]string{"error": "TRACE_REOPEN_FAILED", "message": err.Error()}) + return + } + writeJSONResponse(w, http.StatusOK, stored) +} + +func (s *Server) handleProjectTraceRecheckClear(w http.ResponseWriter, r *http.Request) { + projectID, _, ok := s.traceProject(w, r) + if !ok { + return + } + markID, err := strconv.ParseInt(r.PathValue("markID"), 10, 64) + if err != nil || markID <= 0 { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid mark id"}`) + return + } + var body struct { + Reason string `json:"reason"` + } + if err := json.NewDecoder(r.Body).Decode(&body); err != nil && err != io.EOF { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`) + return + } + marks, err := s.rt.Trace.ListOpenMarks(r.Context(), projectID, "") + if err != nil { + writeSafeError(w, r, http.StatusInternalServerError, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`) + return + } + found := false + for _, m := range marks { + if m.ID == markID { + found = true + break + } + } + if !found { + writeSafeError(w, r, http.StatusNotFound, `{"error":"ERR_NOT_FOUND","message":"未找到该项目的打开标记"}`) + return + } + if err := s.rt.Trace.ClearRecheck(r.Context(), markID, body.Reason); err != nil { + writeJSONResponse(w, http.StatusBadRequest, map[string]string{"error": "TRACE_RECHECK_CLEAR_FAILED", "message": err.Error()}) + return + } + writeJSONResponse(w, http.StatusOK, map[string]string{"status": "cleared"}) +} + +func (s *Server) traceProject(w http.ResponseWriter, r *http.Request) (int64, *auth.UserAccount, bool) { + projectID, err := strconv.ParseInt(r.PathValue("id"), 10, 64) + if err != nil || projectID <= 0 { + writeSafeError(w, r, http.StatusBadRequest, `{"error":"ERR_INVALID_PARAM","message":"invalid project id"}`) + return 0, nil, false + } + if s.rt == nil || s.rt.Trace == nil { + writeSafeError(w, r, http.StatusServiceUnavailable, `{"error":"ERR_NOT_AVAILABLE","message":"追踪索引不可用"}`) + return 0, nil, false + } + user, ok := s.requireProjectAccess(w, r, projectID) + if !ok { + return 0, nil, false + } + return projectID, user, true +} + +func (s *Server) callerMayRequestReopen(r *http.Request, projectID, userID int64, creator bool, req traceimpact.ReopenRequest) bool { + if creator { + return true + } + if s.instSvc == nil { + return false + } + settings, err := s.instSvc.GetSettings(r.Context(), projectID, userID, false) + if err != nil || settings == nil || settings.Plan == nil { + return false + } + var seats []traceimpact.DutySeat + for _, stage := range settings.Plan.Stages { + if stage.StageKey != req.StageKey { + continue + } + for _, task := range stage.DocTasks { + seats = append(seats, traceimpact.DutySeat{ + TaskKey: task.TaskKey, AuthorID: task.Duties.AuthorUserID, + ReviewerID: task.Duties.ReviewerUserID, ApproverID: task.Duties.ApproverUserID, + }) + } + } + return traceimpact.MayRequestReopen(userID, false, req.DocTasks, seats) +} + +func (s *Server) traceDocVisible(r *http.Request, projectID, userID int64, docID string) bool { + if docID == "" || docID == "chat" { + return true + } + parsedProject, taskKey, ok := parseCanonicalWorkflowDocID(docID) + if !ok { + return false + } + if parsedProject != projectID { + return false + } + return s.allowWorkflowDocTaskAccess(r.Context(), projectID, userID, taskKey) +}