diff --git a/deploy/postgres/migrations/2026-09-11-project-workflows-and-lineage.sql b/deploy/postgres/migrations/2026-09-11-project-workflows-and-lineage.sql new file mode 100644 index 00000000..0d6e7b4d --- /dev/null +++ b/deploy/postgres/migrations/2026-09-11-project-workflows-and-lineage.sql @@ -0,0 +1,100 @@ +-- 2026-09-11: 项目多工作流体系、创建人权威化与文档血缘追溯 +-- 支持：项目创建人权限、三层工作流模型、工作流原子激活与审计、智能产品经理与测试经理文档血缘追踪 + +-- 1. 为 projects 表添加创建人与工作流激活控制字段 +ALTER TABLE projects ADD COLUMN IF NOT EXISTS creator_user_id BIGINT REFERENCES users(id); +ALTER TABLE projects ADD COLUMN IF NOT EXISTS active_workflow_key TEXT NOT NULL DEFAULT 'main'; +ALTER TABLE projects ADD COLUMN IF NOT EXISTS workflow_selection_version INT NOT NULL DEFAULT 1; +ALTER TABLE projects ADD COLUMN IF NOT EXISTS active_workflow_changed_at TIMESTAMPTZ; +ALTER TABLE projects ADD COLUMN IF NOT EXISTS active_workflow_changed_by BIGINT REFERENCES users(id); + +CREATE INDEX IF NOT EXISTS idx_projects_creator ON projects(creator_user_id); +CREATE INDEX IF NOT EXISTS idx_projects_active_workflow ON projects(active_workflow_key); + +-- 2. 项目工作流实例表 (挂载到项目的可用工作流实例与局部状态) +CREATE TABLE IF NOT EXISTS project_workflows ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + workflow_key TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'ready', -- ready | active | disabled + enabled BOOLEAN NOT NULL DEFAULT true, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uq_project_workflow UNIQUE(project_id, workflow_key) +); + +CREATE INDEX IF NOT EXISTS idx_project_workflows_pid ON project_workflows(project_id); + +-- 3. 工作流执行运行实例表 (追踪执行历史、步骤与产出文档) +CREATE TABLE IF NOT EXISTS workflow_runs ( + id BIGSERIAL PRIMARY KEY, + run_id TEXT NOT NULL UNIQUE, + project_id BIGINT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + workflow_key TEXT NOT NULL, + status TEXT NOT NULL DEFAULT 'pending', -- pending | running | completed | failed | cancelled + started_by BIGINT REFERENCES users(id), + stage_key TEXT NOT NULL DEFAULT '', + task_key TEXT NOT NULL DEFAULT '', + input_manifest JSONB NOT NULL DEFAULT '{}'::jsonb, + output_doc_id TEXT NOT NULL DEFAULT '', + output_summary TEXT NOT NULL DEFAULT '', + error_message TEXT NOT NULL DEFAULT '', + started_at TIMESTAMPTZ, + finished_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_workflow_runs_pid ON workflow_runs(project_id); +CREATE INDEX IF NOT EXISTS idx_workflow_runs_status ON workflow_runs(status); + +-- 4. 工作流激活切换审计日志 (只追加写入，记录切换人、版本变化与原因) +CREATE TABLE IF NOT EXISTS workflow_activation_audits ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + from_workflow_key TEXT NOT NULL DEFAULT '', + to_workflow_key TEXT NOT NULL, + changed_by BIGINT REFERENCES users(id), + version_before INT NOT NULL, + version_after INT NOT NULL, + reason TEXT NOT NULL DEFAULT '', + created_at TIMESTAMPTZ NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS idx_workflow_activation_audits_pid ON workflow_activation_audits(project_id); + +-- 5. 文档血缘与 Stale 追溯表 (上下游依赖、版本快照哈希与过期标记) +CREATE TABLE IF NOT EXISTS document_lineage ( + id BIGSERIAL PRIMARY KEY, + project_id BIGINT NOT NULL REFERENCES projects(id) ON DELETE CASCADE, + downstream_doc_id TEXT NOT NULL, + downstream_version INT NOT NULL DEFAULT 1, + upstream_doc_id TEXT NOT NULL, + upstream_version INT NOT NULL DEFAULT 1, + upstream_sha256 TEXT NOT NULL DEFAULT '', + is_stale BOOLEAN NOT NULL DEFAULT false, + stale_reason TEXT NOT NULL DEFAULT '', + stale_detected_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT uq_doc_lineage UNIQUE(downstream_doc_id, downstream_version, upstream_doc_id, upstream_version) +); + +CREATE INDEX IF NOT EXISTS idx_doc_lineage_downstream ON document_lineage(downstream_doc_id, downstream_version); +CREATE INDEX IF NOT EXISTS idx_doc_lineage_upstream ON document_lineage(upstream_doc_id, upstream_version); +CREATE INDEX IF NOT EXISTS idx_doc_lineage_stale ON document_lineage(is_stale); + +-- 6. 历史数据回填：为缺失创建人的历史项目回填默认创建人 (关联首个会话或设为管理员 1) +UPDATE projects p +SET creator_user_id = COALESCE( + (SELECT cs.user_id FROM chat_sessions cs WHERE cs.project_id = p.id AND cs.user_id IS NOT NULL ORDER BY cs.id ASC LIMIT 1), + 1 +) +WHERE p.creator_user_id IS NULL; + +-- 7. 历史数据回填：为已有项目自动挂载三个标准工作流实例 (main, product_manager, test_manager) +INSERT INTO project_workflows (project_id, workflow_key, status, enabled) +SELECT p.id, w.key, + CASE WHEN w.key = COALESCE(p.active_workflow_key, 'main') THEN 'active' ELSE 'ready' END, + true +FROM projects p +CROSS JOIN (VALUES ('main'), ('product_manager'), ('test_manager')) AS w(key) +ON CONFLICT (project_id, workflow_key) DO NOTHING; diff --git a/internal/auth/service.go b/internal/auth/service.go index 421c99a3..d6b5e12c 100644 --- a/internal/auth/service.go +++ b/internal/auth/service.go @@ -207,6 +207,17 @@ func (s *Service) GetUserByToken(ctx context.Context, token string) (*UserAccoun return user, nil } +// GetUserByID 根据用户 ID 直接查询用户信息 (安全过滤密码哈希) +func (s *Service) GetUserByID(ctx context.Context, userID int64) (*UserAccount, error) { + user, err := s.store.GetUserByID(ctx, userID) + if err != nil { + return nil, err + } + cpy := *user + cpy.PasswordHash = "" + return &cpy, nil +} + // Logout 注销会话 func (s *Service) Logout(token string) { s.tokensMu.Lock() diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index 0e854430..16ac90d3 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -17,6 +17,7 @@ import ( "chat-agent-system/internal/docgov" "chat-agent-system/internal/freedoc" "chat-agent-system/internal/kb" + "chat-agent-system/internal/projectworkflow" "chat-agent-system/internal/squad" "chat-agent-system/internal/storage" "chat-agent-system/internal/stream" @@ -25,13 +26,15 @@ import ( ) type Service struct { - cfg *config.Config - kb *kb.Client - squad *squad.Service - ctxMgr *ctxmgr.ContextManager - cas *storage.CAS - docGov *docgov.GovernanceEngine - freeDoc *freedoc.Service + cfg *config.Config + kb *kb.Client + squad *squad.Service + ctxMgr *ctxmgr.ContextManager + cas *storage.CAS + docGov *docgov.GovernanceEngine + freeDoc *freedoc.Service + projectWorkflow *projectworkflow.Service + pool *pgxpool.Pool } func NewService(cfg *config.Config) *Service { @@ -57,15 +60,19 @@ func NewService(cfg *config.Config) *Service { } docGov := docgov.NewGovernanceEngine(pool) freeDoc := freedoc.NewService(pool, ctxMgr) + pwStore := projectworkflow.NewStore(pool) + pwSvc := projectworkflow.NewService(pwStore, projectworkflow.NewDefaultCatalog()) svc := &Service{ - cfg: cfg, - kb: kb.NewClient(cfg), - squad: squad.NewService(cfg), - ctxMgr: ctxMgr, - cas: casStore, - docGov: docGov, - freeDoc: freeDoc, + cfg: cfg, + kb: kb.NewClient(cfg), + squad: squad.NewService(cfg), + ctxMgr: ctxMgr, + cas: casStore, + docGov: docGov, + freeDoc: freeDoc, + projectWorkflow: pwSvc, + pool: pool, } // 注册终审批准回调：自动切片向量化沉淀为往期标杆案例 @@ -123,6 +130,14 @@ func (s *Service) DocGov() *docgov.GovernanceEngine { return s.docGov } +func (s *Service) ProjectWorkflow() *projectworkflow.Service { + return s.projectWorkflow +} + +func (s *Service) Pool() *pgxpool.Pool { + return s.pool +} + func (s *Service) FreeformDoc() *freedoc.Service { return s.freeDoc } diff --git a/internal/projectworkflow/catalog.go b/internal/projectworkflow/catalog.go new file mode 100644 index 00000000..ec42b490 --- /dev/null +++ b/internal/projectworkflow/catalog.go @@ -0,0 +1,110 @@ +package projectworkflow + +import "sync" + +// Definition 工作流静态规范与契约定义 +type Definition struct { + Key string `json:"key"` + Name string `json:"name"` + Description string `json:"description"` + StageKey string `json:"stage_key"` + TaskKey string `json:"task_key"` + RequiredUpstreamWorkflow string `json:"required_upstream_workflow,omitempty"` + OutputDocType string `json:"output_doc_type"` + OutputDocTitle string `json:"output_doc_title"` + Enabled bool `json:"enabled"` +} + +// Catalog 工作流目录接口 +type Catalog interface { + Get(workflowKey string) (*Definition, bool) + List() []*Definition + Validate() error +} + +type memoryCatalog struct { + mu sync.RWMutex + definitions map[string]*Definition + order []string +} + +// NewDefaultCatalog 创建默认工作流目录（内建 main、product_manager、test_manager） +func NewDefaultCatalog() Catalog { + c := &memoryCatalog{ + definitions: make(map[string]*Definition), + order: []string{"main", "product_manager", "test_manager"}, + } + + c.definitions["main"] = &Definition{ + Key: "main", + Name: "商务与研发综合主线", + Description: "覆盖从 V0.1 分诊到 V0.6 阶段评审的全流程智能体编排与商务研发主干协同", + StageKey: "main", + TaskKey: "main", + OutputDocType: "main_proposal", + OutputDocTitle: "商务研发综合方案", + Enabled: true, + } + + c.definitions["product_manager"] = &Definition{ + Key: "product_manager", + Name: "智能产品经理", + Description: "基于用户需求或项目上下文深度提炼，生成高可靠、可追溯的系统需求规格书", + StageKey: "system_reqs", + TaskKey: "system_reqs", + OutputDocType: "system_reqs", + OutputDocTitle: "系统需求规格书", + Enabled: true, + } + + c.definitions["test_manager"] = &Definition{ + Key: "test_manager", + Name: "智能测试经理", + Description: "依赖产品经理已批准的系统需求规格书，生成全覆盖、满足法规要求的系统验证方案及计划", + StageKey: "system_test_plan", + TaskKey: "test_manager", + RequiredUpstreamWorkflow: "product_manager", + OutputDocType: "system_test_plan", + OutputDocTitle: "系统验证方案及计划", + Enabled: true, + } + + return c +} + +func (c *memoryCatalog) Get(workflowKey string) (*Definition, bool) { + c.mu.RLock() + defer c.mu.RUnlock() + d, ok := c.definitions[workflowKey] + if !ok { + return nil, false + } + cpy := *d + return &cpy, true +} + +func (c *memoryCatalog) List() []*Definition { + c.mu.RLock() + defer c.mu.RUnlock() + list := make([]*Definition, 0, len(c.order)) + for _, key := range c.order { + if d, ok := c.definitions[key]; ok { + cpy := *d + list = append(list, &cpy) + } + } + return list +} + +func (c *memoryCatalog) Validate() error { + c.mu.RLock() + defer c.mu.RUnlock() + for _, def := range c.definitions { + if def.RequiredUpstreamWorkflow != "" { + if _, ok := c.definitions[def.RequiredUpstreamWorkflow]; !ok { + return ErrWorkflowNotFound + } + } + } + return nil +} diff --git a/internal/projectworkflow/errors.go b/internal/projectworkflow/errors.go new file mode 100644 index 00000000..e0ed3ca8 --- /dev/null +++ b/internal/projectworkflow/errors.go @@ -0,0 +1,26 @@ +package projectworkflow + +import "errors" + +var ( + // ErrUnauthenticated 未提供认证 Token 或会话已失效 + ErrUnauthenticated = errors.New("unauthenticated: valid authentication token required") + + // ErrForbiddenNotCreator 仅项目创建人有权切换工作流或配置 + ErrForbiddenNotCreator = errors.New("forbidden: only project creator can switch active workflow") + + // ErrProjectNotFound 指定项目不存在或无权访问 + ErrProjectNotFound = errors.New("project not found") + + // ErrWorkflowNotFound 工作流未定义或未在当前项目启用 + ErrWorkflowNotFound = errors.New("workflow definition not found or disabled for project") + + // ErrConcurrentConflict 乐观锁版本冲突 (workflow_selection_version 不匹配) + ErrConcurrentConflict = errors.New("conflict: workflow selection version mismatch") + + // ErrPrerequisiteNotMet 前置条件未满足 (例如缺少已批准的上游需求规格书) + ErrPrerequisiteNotMet = errors.New("precondition failed: prerequisite upstream document not satisfied") + + // ErrInvalidInput 输入参数不合法 + ErrInvalidInput = errors.New("invalid parameter: required field missing or malformed") +) diff --git a/internal/projectworkflow/service.go b/internal/projectworkflow/service.go new file mode 100644 index 00000000..1e49a9ee --- /dev/null +++ b/internal/projectworkflow/service.go @@ -0,0 +1,280 @@ +package projectworkflow + +import ( + "context" + "crypto/rand" + "encoding/hex" + "fmt" + "strings" + "time" + + "chat-agent-system/internal/auth" +) + +// Service 项目多工作流与血缘领域服务 +type Service struct { + store Store + catalog Catalog +} + +// NewService 构造项目工作流服务 +func NewService(store Store, catalog Catalog) *Service { + if catalog == nil { + catalog = NewDefaultCatalog() + } + return &Service{ + store: store, + catalog: catalog, + } +} + +// Catalog 获取静态工作流规范目录 +func (s *Service) Catalog() Catalog { + return s.catalog +} + +// Store 获取持久化存储接口 +func (s *Service) Store() Store { + return s.store +} + +// CreateProject 创建新项目，严格从认证令牌注入权威创建人，并自动初始化挂载工作流 +func (s *Service) CreateProject(ctx context.Context, user *auth.UserAccount, in CreateProjectInput) (*Project, error) { + if user == nil || user.ID <= 0 { + return nil, ErrUnauthenticated + } + + name := strings.TrimSpace(in.Name) + if name == "" { + return nil, fmt.Errorf("%w: 项目名称为必填项", ErrInvalidInput) + } + + code := strings.TrimSpace(in.Code) + if code == "" { + code = generateProjectCode(name) + } + + initialWorkflow := in.InitialWorkflowKey + if initialWorkflow == "" { + initialWorkflow = "main" + } + if _, ok := s.catalog.Get(initialWorkflow); !ok { + return nil, fmt.Errorf("%w: 初始工作流 %s 不存在", ErrWorkflowNotFound, initialWorkflow) + } + + p := &Project{ + Code: code, + Name: name, + CustomerName: in.CustomerName, + DeviceType: in.DeviceType, + IntendedUse: in.IntendedUse, + TargetMarket: in.TargetMarket, + CurrentStage: "V0.1_triage", + Environment: "dev", + Status: "active", + CreatorUserID: user.ID, + ActiveWorkflowKey: initialWorkflow, + WorkflowSelectionVersion: 1, + } + + created, err := s.store.CreateProject(ctx, p) + if err != nil { + return nil, err + } + + // 自动挂载目录中的工作流实例 + var allKeys []string + for _, def := range s.catalog.List() { + allKeys = append(allKeys, def.Key) + } + if err := s.store.EnsureProjectWorkflows(ctx, created.ID, allKeys); err != nil { + return nil, fmt.Errorf("initialize project workflows failed: %w", err) + } + + return created, nil +} + +// GetProject 根据项目 ID 获取项目详情 +func (s *Service) GetProject(ctx context.Context, projectID int64, userID int64) (*Project, error) { + if projectID <= 0 { + return nil, ErrProjectNotFound + } + p, err := s.store.GetProjectByID(ctx, projectID) + if err != nil { + return nil, err + } + return p, nil +} + +// GetProjectByCode 根据项目唯一编码获取项目详情 +func (s *Service) GetProjectByCode(ctx context.Context, code string) (*Project, error) { + return s.store.GetProjectByCode(ctx, code) +} + +// ListProjects 查询指定用户可见的项目列表 +func (s *Service) ListProjects(ctx context.Context, userID int64) ([]*Project, error) { + return s.store.ListProjectsForUser(ctx, userID) +} + +// IsProjectCreator 判断用户是否为指定项目的权威创建人 +func (s *Service) IsProjectCreator(ctx context.Context, projectID int64, userID int64) (bool, error) { + p, err := s.store.GetProjectByID(ctx, projectID) + if err != nil { + return false, err + } + return p.CreatorUserID == userID, nil +} + +// ListProjectWorkflows 查询项目已挂载的工作流清单 +func (s *Service) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) { + // 确保默认工作流已初始化 + var allKeys []string + for _, def := range s.catalog.List() { + allKeys = append(allKeys, def.Key) + } + _ = s.store.EnsureProjectWorkflows(ctx, projectID, allKeys) + + return s.store.ListProjectWorkflows(ctx, projectID) +} + +// ActivateWorkflow 原子切换项目当前激活的工作流焦点（仅限创建人） +func (s *Service) ActivateWorkflow( + ctx context.Context, + projectID int64, + userID int64, + in ActivateWorkflowInput, +) (*ActivateWorkflowResult, error) { + if userID <= 0 { + return nil, ErrUnauthenticated + } + + targetKey := strings.TrimSpace(in.WorkflowKey) + if targetKey == "" { + return nil, fmt.Errorf("%w: target workflow_key cannot be empty", ErrInvalidInput) + } + + def, ok := s.catalog.Get(targetKey) + if !ok || !def.Enabled { + return nil, fmt.Errorf("%w: workflow '%s' is not supported or disabled", ErrWorkflowNotFound, targetKey) + } + + updatedProject, audit, err := s.store.UpdateActiveWorkflow( + ctx, + projectID, + targetKey, + in.ExpectedVersion, + userID, + in.Reason, + ) + if err != nil { + return nil, err + } + + now := time.Now() + if audit != nil { + now = audit.CreatedAt + } else if updatedProject.ActiveWorkflowChangedAt != nil { + now = *updatedProject.ActiveWorkflowChangedAt + } + + return &ActivateWorkflowResult{ + ProjectID: updatedProject.ID, + ActiveWorkflowKey: updatedProject.ActiveWorkflowKey, + Version: updatedProject.WorkflowSelectionVersion, + ChangedAt: now, + ChangedBy: userID, + }, nil +} + +// CreateWorkflowRun 创建工作流运行实例 +func (s *Service) CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error) { + if run.RunID == "" { + run.RunID = generateRunID(run.WorkflowKey) + } + if run.Status == "" { + run.Status = "running" + } + return s.store.CreateWorkflowRun(ctx, run) +} + +// UpdateWorkflowRun 结束/更新工作流运行实例 +func (s *Service) UpdateWorkflowRun( + ctx context.Context, + runID string, + status string, + outputDocID string, + summary string, + errMsg string, +) error { + return s.store.UpdateWorkflowRun(ctx, runID, status, outputDocID, summary, errMsg) +} + +// GetWorkflowRun 获取运行记录详情 +func (s *Service) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error) { + return s.store.GetWorkflowRun(ctx, runID) +} + +// ListWorkflowRuns 列出项目的运行历史 +func (s *Service) ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error) { + return s.store.ListWorkflowRuns(ctx, projectID, workflowKey, limit) +} + +// RecordLineage 记录文档版本血缘 +func (s *Service) RecordLineage(ctx context.Context, lineage *DocumentLineage) error { + return s.store.RecordLineage(ctx, lineage) +} + +// GetLineageForDownstream 查询下游文档所依赖的上游版本信息 +func (s *Service) GetLineageForDownstream(ctx context.Context, docID string, version int) ([]*DocumentLineage, error) { + return s.store.GetLineageForDownstream(ctx, docID, version) +} + +// CheckAndPropagateStale 当上游文档更新且内容哈希变化时，通知并将关联下游文档标记为 stale +func (s *Service) CheckAndPropagateStale( + ctx context.Context, + upstreamDocID string, + newSHA256 string, + reason string, +) ([]*DocumentLineage, error) { + lineages, err := s.store.GetLineagesForUpstream(ctx, upstreamDocID) + if err != nil { + return nil, err + } + + needsStale := false + for _, l := range lineages { + if l.UpstreamSHA256 != "" && l.UpstreamSHA256 != newSHA256 && !l.IsStale { + needsStale = true + break + } + } + + if !needsStale { + return nil, nil + } + + if reason == "" { + reason = fmt.Sprintf("上游文档 %s 生成了新版本，内容已发生实质变化", upstreamDocID) + } + + return s.store.MarkDownstreamStale(ctx, upstreamDocID, reason) +} + +// 辅助方法：生成项目短代码 +func generateProjectCode(name string) string { + b := make([]byte, 4) + _, _ = rand.Read(b) + hash := hex.EncodeToString(b) + cleanName := strings.ToLower(strings.ReplaceAll(name, " ", "-")) + if len(cleanName) > 12 { + cleanName = cleanName[:12] + } + return fmt.Sprintf("%s-%s", cleanName, hash) +} + +// 辅助方法：生成运行实例 ID +func generateRunID(workflowKey string) string { + b := make([]byte, 6) + _, _ = rand.Read(b) + return fmt.Sprintf("run_%s_%s", workflowKey, hex.EncodeToString(b)) +} diff --git a/internal/projectworkflow/service_test.go b/internal/projectworkflow/service_test.go new file mode 100644 index 00000000..6b9ea821 --- /dev/null +++ b/internal/projectworkflow/service_test.go @@ -0,0 +1,166 @@ +package projectworkflow + +import ( + "context" + "errors" + "testing" + + "chat-agent-system/internal/auth" +) + +func TestProjectWorkflowService_CreateAndCreatorACL(t *testing.T) { + ctx := context.Background() + store := NewStore(nil) + catalog := NewDefaultCatalog() + svc := NewService(store, catalog) + + creator := &auth.UserAccount{ID: 101, Username: "zhangsan", Name: "张工"} + otherUser := &auth.UserAccount{ID: 102, Username: "lisi", Name: "李主管"} + + // 1. 未认证创建应被拦截 + _, err := svc.CreateProject(ctx, nil, CreateProjectInput{Name: "测试项目A"}) + if err != ErrUnauthenticated { + t.Fatalf("expected ErrUnauthenticated, got: %v", err) + } + + // 2. 正常创建 + proj, err := svc.CreateProject(ctx, creator, CreateProjectInput{ + Name: "神经脉冲射频仪 (NeuroPulse)", + DeviceType: "高频神经射频治疗设备", + IntendedUse: "用于中枢神经阻滞与射频毁损治疗", + TargetMarket: "NMPA Class III", + }) + if err != nil { + t.Fatalf("CreateProject failed: %v", err) + } + if proj.CreatorUserID != creator.ID { + t.Errorf("expected CreatorUserID %d, got %d", creator.ID, proj.CreatorUserID) + } + if proj.ActiveWorkflowKey != "main" { + t.Errorf("expected default active workflow 'main', got '%s'", proj.ActiveWorkflowKey) + } + if proj.WorkflowSelectionVersion != 1 { + t.Errorf("expected initial version 1, got %d", proj.WorkflowSelectionVersion) + } + + // 3. 验证自动挂载的工作流清单 + wfs, err := svc.ListProjectWorkflows(ctx, proj.ID) + if err != nil { + t.Fatalf("ListProjectWorkflows failed: %v", err) + } + if len(wfs) != 3 { + t.Fatalf("expected 3 default workflows, got %d", len(wfs)) + } + + // 4. 非创建人尝试切换工作流 -> 必须被拒绝 + _, err = svc.ActivateWorkflow(ctx, proj.ID, otherUser.ID, ActivateWorkflowInput{ + WorkflowKey: "product_manager", + ExpectedVersion: 1, + Reason: "尝试越权切换", + }) + if err != ErrForbiddenNotCreator { + t.Fatalf("expected ErrForbiddenNotCreator, got: %v", err) + } + + // 5. 创建人正常切换工作流 -> 成功升级版本 + res, err := svc.ActivateWorkflow(ctx, proj.ID, creator.ID, ActivateWorkflowInput{ + WorkflowKey: "product_manager", + ExpectedVersion: 1, + Reason: "进入产品需求阶段", + }) + if err != nil { + t.Fatalf("ActivateWorkflow failed: %v", err) + } + if res.ActiveWorkflowKey != "product_manager" { + t.Errorf("expected active workflow 'product_manager', got '%s'", res.ActiveWorkflowKey) + } + if res.Version != 2 { + t.Errorf("expected version 2 after switch, got %d", res.Version) + } + + // 6. 并发冲突：使用旧版本 1 再次切换 -> 必须返回 ErrConcurrentConflict + _, err = svc.ActivateWorkflow(ctx, proj.ID, creator.ID, ActivateWorkflowInput{ + WorkflowKey: "test_manager", + ExpectedVersion: 1, // 旧版本 + Reason: "并发落后切换", + }) + if err != ErrConcurrentConflict { + t.Fatalf("expected ErrConcurrentConflict, got: %v", err) + } + + // 7. 幂等测试：重复激活当前工作流 'product_manager' + resIdempotent, err := svc.ActivateWorkflow(ctx, proj.ID, creator.ID, ActivateWorkflowInput{ + WorkflowKey: "product_manager", + ExpectedVersion: 2, + Reason: "重复激活", + }) + if err != nil { + t.Fatalf("idempotent switch failed: %v", err) + } + if resIdempotent.ActiveWorkflowKey != "product_manager" { + t.Errorf("expected active 'product_manager', got '%s'", resIdempotent.ActiveWorkflowKey) + } + + // 8. 切换到不存在的工作流 -> 返回 ErrWorkflowNotFound + _, err = svc.ActivateWorkflow(ctx, proj.ID, creator.ID, ActivateWorkflowInput{ + WorkflowKey: "non_existent_wf", + }) + if err == nil || !errors.Is(err, ErrWorkflowNotFound) { + t.Fatalf("expected ErrWorkflowNotFound, got: %v", err) + } +} + +func TestProjectWorkflowService_LineageAndStale(t *testing.T) { + ctx := context.Background() + store := NewStore(nil) + catalog := NewDefaultCatalog() + svc := NewService(store, catalog) + + creator := &auth.UserAccount{ID: 101, Username: "zhangsan", Name: "张工"} + proj, _ := svc.CreateProject(ctx, creator, CreateProjectInput{Name: "超声手术刀"}) + + // 记录测试经理方案依赖产品经理需求规格书 v1 + err := svc.RecordLineage(ctx, &DocumentLineage{ + ProjectID: proj.ID, + DownstreamDocID: "doc_test_plan_001", + DownstreamVersion: 1, + UpstreamDocID: "doc_system_reqs_001", + UpstreamVersion: 1, + UpstreamSHA256: "hash_v1_aaa", + }) + if err != nil { + t.Fatalf("RecordLineage failed: %v", err) + } + + lineages, err := svc.GetLineageForDownstream(ctx, "doc_test_plan_001", 1) + if err != nil || len(lineages) == 0 { + t.Fatalf("expected lineage record, got err: %v, len: %d", err, len(lineages)) + } + if lineages[0].IsStale { + t.Errorf("initial lineage should not be stale") + } + + // 上游生成相同 SHA256 -> 不触发 stale + staleUpdates, err := svc.CheckAndPropagateStale(ctx, "doc_system_reqs_001", "hash_v1_aaa", "") + if err != nil { + t.Fatalf("CheckAndPropagateStale failed: %v", err) + } + if len(staleUpdates) > 0 { + t.Errorf("expected 0 stale updates for identical hash, got %d", len(staleUpdates)) + } + + // 上游生成不同 SHA256 (内容更新) -> 触发下游 stale 传播 + staleUpdates, err = svc.CheckAndPropagateStale(ctx, "doc_system_reqs_001", "hash_v2_bbb", "系统需求新增 RF-009 射频功率限制") + if err != nil { + t.Fatalf("CheckAndPropagateStale failed: %v", err) + } + if len(staleUpdates) != 1 { + t.Fatalf("expected 1 stale update, got %d", len(staleUpdates)) + } + if !staleUpdates[0].IsStale { + t.Errorf("lineage should be marked as stale") + } + if staleUpdates[0].DownstreamDocID != "doc_test_plan_001" { + t.Errorf("expected downstream 'doc_test_plan_001', got '%s'", staleUpdates[0].DownstreamDocID) + } +} diff --git a/internal/projectworkflow/store.go b/internal/projectworkflow/store.go new file mode 100644 index 00000000..ae79cfc5 --- /dev/null +++ b/internal/projectworkflow/store.go @@ -0,0 +1,846 @@ +package projectworkflow + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "sync" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// Store 数据存储层接口 +type Store interface { + CreateProject(ctx context.Context, p *Project) (*Project, error) + GetProjectByID(ctx context.Context, id int64) (*Project, error) + GetProjectByCode(ctx context.Context, code string) (*Project, error) + ListProjectsForUser(ctx context.Context, userID int64) ([]*Project, error) + ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) + EnsureProjectWorkflows(ctx context.Context, projectID int64, defaultKeys []string) error + UpdateActiveWorkflow(ctx context.Context, projectID int64, toWorkflowKey string, expectedVersion int, changedBy int64, reason string) (*Project, *WorkflowActivationAudit, error) + + CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error) + UpdateWorkflowRun(ctx context.Context, runID string, status string, outputDocID string, summary string, errMsg string) error + GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error) + ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error) + + RecordLineage(ctx context.Context, lineage *DocumentLineage) error + GetLineageForDownstream(ctx context.Context, docID string, version int) ([]*DocumentLineage, error) + GetLineagesForUpstream(ctx context.Context, upstreamDocID string) ([]*DocumentLineage, error) + MarkDownstreamStale(ctx context.Context, upstreamDocID string, staleReason string) ([]*DocumentLineage, error) +} + +// NewStore 根据 pgxpool 是否可用构建 PGStore 或内存 Fallback Store +func NewStore(pool *pgxpool.Pool) Store { + if pool != nil { + return &pgStore{pool: pool} + } + return newMemStore() +} + +// ============================================================================= +// PostgreSQL 权威持久化实现 +// ============================================================================= + +type pgStore struct { + pool *pgxpool.Pool +} + +func (s *pgStore) CreateProject(ctx context.Context, p *Project) (*Project, error) { + now := time.Now() + p.CreatedAt = now + p.UpdatedAt = now + if p.Status == "" { + p.Status = "active" + } + if p.Environment == "" { + p.Environment = "dev" + } + if p.CurrentStage == "" { + p.CurrentStage = "V0.1_triage" + } + if p.ActiveWorkflowKey == "" { + p.ActiveWorkflowKey = "main" + } + if p.WorkflowSelectionVersion <= 0 { + p.WorkflowSelectionVersion = 1 + } + + query := ` + INSERT INTO projects ( + code, name, customer_name, device_type, intended_use, target_market, + current_stage, environment, status, creator_user_id, active_workflow_key, + workflow_selection_version, created_at, updated_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14) + RETURNING id + ` + err := s.pool.QueryRow( + ctx, query, + p.Code, p.Name, p.CustomerName, p.DeviceType, p.IntendedUse, p.TargetMarket, + p.CurrentStage, p.Environment, p.Status, p.CreatorUserID, p.ActiveWorkflowKey, + p.WorkflowSelectionVersion, p.CreatedAt, p.UpdatedAt, + ).Scan(&p.ID) + if err != nil { + return nil, fmt.Errorf("insert project failed: %w", err) + } + + return p, nil +} + +func (s *pgStore) GetProjectByID(ctx context.Context, id int64) (*Project, error) { + query := ` + SELECT id, code, name, customer_name, device_type, intended_use, target_market, + current_stage, environment, status, COALESCE(creator_user_id, 1), + COALESCE(active_workflow_key, 'main'), COALESCE(workflow_selection_version, 1), + active_workflow_changed_at, active_workflow_changed_by, created_at, updated_at + FROM projects + WHERE id = $1 + ` + p := &Project{} + err := s.pool.QueryRow(ctx, query, id).Scan( + &p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket, + &p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID, + &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, + &p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrProjectNotFound + } + if err != nil { + return nil, fmt.Errorf("query project failed: %w", err) + } + return p, nil +} + +func (s *pgStore) GetProjectByCode(ctx context.Context, code string) (*Project, error) { + query := ` + SELECT id, code, name, customer_name, device_type, intended_use, target_market, + current_stage, environment, status, COALESCE(creator_user_id, 1), + COALESCE(active_workflow_key, 'main'), COALESCE(workflow_selection_version, 1), + active_workflow_changed_at, active_workflow_changed_by, created_at, updated_at + FROM projects + WHERE code = $1 + ` + p := &Project{} + err := s.pool.QueryRow(ctx, query, code).Scan( + &p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket, + &p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID, + &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, + &p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, ErrProjectNotFound + } + if err != nil { + return nil, fmt.Errorf("query project by code failed: %w", err) + } + return p, nil +} + +func (s *pgStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Project, error) { + // 允许查看项目（自己创建，或公开项目，或已有会话参与） + query := ` + SELECT DISTINCT p.id, p.code, p.name, p.customer_name, p.device_type, p.intended_use, p.target_market, + p.current_stage, p.environment, p.status, COALESCE(p.creator_user_id, 1), + COALESCE(p.active_workflow_key, 'main'), COALESCE(p.workflow_selection_version, 1), + p.active_workflow_changed_at, p.active_workflow_changed_by, p.created_at, p.updated_at + FROM projects p + LEFT JOIN chat_sessions cs ON cs.project_id = p.id AND cs.user_id = $1 + WHERE p.creator_user_id = $1 OR cs.id IS NOT NULL OR p.creator_user_id IS NULL OR p.creator_user_id = 1 + ORDER BY p.id DESC + ` + rows, err := s.pool.Query(ctx, query, userID) + if err != nil { + return nil, fmt.Errorf("list projects failed: %w", err) + } + defer rows.Close() + + var list []*Project + for rows.Next() { + p := &Project{} + if err := rows.Scan( + &p.ID, &p.Code, &p.Name, &p.CustomerName, &p.DeviceType, &p.IntendedUse, &p.TargetMarket, + &p.CurrentStage, &p.Environment, &p.Status, &p.CreatorUserID, + &p.ActiveWorkflowKey, &p.WorkflowSelectionVersion, + &p.ActiveWorkflowChangedAt, &p.ActiveWorkflowChangedBy, &p.CreatedAt, &p.UpdatedAt, + ); err != nil { + return nil, err + } + list = append(list, p) + } + return list, nil +} + +func (s *pgStore) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) { + query := ` + SELECT id, project_id, workflow_key, status, enabled, created_at, updated_at + FROM project_workflows + WHERE project_id = $1 + ORDER BY id ASC + ` + rows, err := s.pool.Query(ctx, query, projectID) + if err != nil { + return nil, fmt.Errorf("list project workflows failed: %w", err) + } + defer rows.Close() + + var list []*ProjectWorkflow + for rows.Next() { + pw := &ProjectWorkflow{} + if err := rows.Scan(&pw.ID, &pw.ProjectID, &pw.WorkflowKey, &pw.Status, &pw.Enabled, &pw.CreatedAt, &pw.UpdatedAt); err != nil { + return nil, err + } + list = append(list, pw) + } + return list, nil +} + +func (s *pgStore) EnsureProjectWorkflows(ctx context.Context, projectID int64, defaultKeys []string) error { + for _, key := range defaultKeys { + status := "ready" + query := ` + INSERT INTO project_workflows (project_id, workflow_key, status, enabled, created_at, updated_at) + VALUES ($1, $2, $3, true, now(), now()) + ON CONFLICT (project_id, workflow_key) DO NOTHING + ` + if _, err := s.pool.Exec(ctx, query, projectID, key, status); err != nil { + return fmt.Errorf("ensure workflow %s failed: %w", key, err) + } + } + return nil +} + +func (s *pgStore) UpdateActiveWorkflow( + ctx context.Context, + projectID int64, + toWorkflowKey string, + expectedVersion int, + changedBy int64, + reason string, +) (*Project, *WorkflowActivationAudit, error) { + tx, err := s.pool.Begin(ctx) + if err != nil { + return nil, nil, fmt.Errorf("begin transaction failed: %w", err) + } + defer tx.Rollback(ctx) //nolint:errcheck + + // 1. 行级排他锁项目记录，读取当前版本与激活状态 + var curVersion int + var curKey string + var creatorID int64 + lockQuery := ` + SELECT workflow_selection_version, active_workflow_key, COALESCE(creator_user_id, 1) + FROM projects + WHERE id = $1 + FOR UPDATE + ` + err = tx.QueryRow(ctx, lockQuery, projectID).Scan(&curVersion, &curKey, &creatorID) + if errors.Is(err, pgx.ErrNoRows) { + return nil, nil, ErrProjectNotFound + } + if err != nil { + return nil, nil, fmt.Errorf("lock project failed: %w", err) + } + + // 2. 校验权限 + if creatorID > 0 && changedBy > 0 && creatorID != changedBy { + return nil, nil, ErrForbiddenNotCreator + } + + // 3. 校验乐观版本 + if expectedVersion > 0 && expectedVersion != curVersion { + return nil, nil, ErrConcurrentConflict + } + + // 若已经是目标工作流，幂等返回 + if curKey == toWorkflowKey { + p, err := s.GetProjectByID(ctx, projectID) + if err != nil { + return nil, nil, err + } + return p, nil, nil + } + + nextVersion := curVersion + 1 + now := time.Now() + + // 4. 更新 projects 表激活信息 + updateProjectQuery := ` + UPDATE projects + SET active_workflow_key = $1, + workflow_selection_version = $2, + active_workflow_changed_at = $3, + active_workflow_changed_by = $4, + updated_at = $3 + WHERE id = $5 + ` + if _, err := tx.Exec(ctx, updateProjectQuery, toWorkflowKey, nextVersion, now, changedBy, projectID); err != nil { + return nil, nil, fmt.Errorf("update project active workflow failed: %w", err) + } + + // 5. 更新 project_workflows 表状态 (旧 active -> ready, 新 -> active) + _, _ = tx.Exec(ctx, `UPDATE project_workflows SET status = 'ready', updated_at = now() WHERE project_id = $1 AND status = 'active'`, projectID) + _, _ = tx.Exec(ctx, `UPDATE project_workflows SET status = 'active', updated_at = now() WHERE project_id = $1 AND workflow_key = $2`, projectID, toWorkflowKey) + + // 6. 追加审计日志 + auditQuery := ` + INSERT INTO workflow_activation_audits ( + project_id, from_workflow_key, to_workflow_key, changed_by, + version_before, version_after, reason, created_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8) + RETURNING id + ` + audit := &WorkflowActivationAudit{ + ProjectID: projectID, + FromWorkflowKey: curKey, + ToWorkflowKey: toWorkflowKey, + ChangedBy: changedBy, + VersionBefore: curVersion, + VersionAfter: nextVersion, + Reason: reason, + CreatedAt: now, + } + if err := tx.QueryRow(ctx, auditQuery, + audit.ProjectID, audit.FromWorkflowKey, audit.ToWorkflowKey, audit.ChangedBy, + audit.VersionBefore, audit.VersionAfter, audit.Reason, audit.CreatedAt, + ).Scan(&audit.ID); err != nil { + return nil, nil, fmt.Errorf("insert audit log failed: %w", err) + } + + if err := tx.Commit(ctx); err != nil { + return nil, nil, fmt.Errorf("commit active workflow transaction failed: %w", err) + } + + updatedProject, err := s.GetProjectByID(ctx, projectID) + if err != nil { + return nil, nil, err + } + return updatedProject, audit, nil +} + +func (s *pgStore) CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error) { + manifestJSON, _ := json.Marshal(run.InputManifest) + query := ` + INSERT INTO workflow_runs ( + run_id, project_id, workflow_key, status, started_by, stage_key, task_key, + input_manifest, output_doc_id, output_summary, error_message, started_at, created_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, now()) + RETURNING id, created_at + ` + now := time.Now() + run.StartedAt = &now + err := s.pool.QueryRow( + ctx, query, + run.RunID, run.ProjectID, run.WorkflowKey, run.Status, run.StartedBy, + run.StageKey, run.TaskKey, manifestJSON, run.OutputDocID, run.OutputSummary, + run.ErrorMessage, run.StartedAt, + ).Scan(&run.ID, &run.CreatedAt) + if err != nil { + return nil, fmt.Errorf("create workflow run failed: %w", err) + } + return run, nil +} + +func (s *pgStore) UpdateWorkflowRun(ctx context.Context, runID string, status string, outputDocID string, summary string, errMsg string) error { + now := time.Now() + query := ` + UPDATE workflow_runs + SET status = $1, + output_doc_id = CASE WHEN $2 <> '' THEN $2 ELSE output_doc_id END, + output_summary = CASE WHEN $3 <> '' THEN $3 ELSE output_summary END, + error_message = $4, + finished_at = $5 + WHERE run_id = $6 + ` + _, err := s.pool.Exec(ctx, query, status, outputDocID, summary, errMsg, now, runID) + return err +} + +func (s *pgStore) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error) { + query := ` + SELECT id, run_id, project_id, workflow_key, status, started_by, stage_key, task_key, + input_manifest, output_doc_id, output_summary, error_message, started_at, finished_at, created_at + FROM workflow_runs + WHERE run_id = $1 + ` + run := &WorkflowRun{} + var manifestRaw []byte + err := s.pool.QueryRow(ctx, query, runID).Scan( + &run.ID, &run.RunID, &run.ProjectID, &run.WorkflowKey, &run.Status, &run.StartedBy, + &run.StageKey, &run.TaskKey, &manifestRaw, &run.OutputDocID, &run.OutputSummary, + &run.ErrorMessage, &run.StartedAt, &run.FinishedAt, &run.CreatedAt, + ) + if errors.Is(err, pgx.ErrNoRows) { + return nil, errors.New("workflow run not found") + } + if err != nil { + return nil, err + } + if len(manifestRaw) > 0 { + _ = json.Unmarshal(manifestRaw, &run.InputManifest) + } + return run, nil +} + +func (s *pgStore) ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error) { + if limit <= 0 { + limit = 20 + } + query := ` + SELECT id, run_id, project_id, workflow_key, status, started_by, stage_key, task_key, + output_doc_id, output_summary, error_message, started_at, finished_at, created_at + FROM workflow_runs + WHERE project_id = $1 AND ($2 = '' OR workflow_key = $2) + ORDER BY id DESC + LIMIT $3 + ` + rows, err := s.pool.Query(ctx, query, projectID, workflowKey, limit) + if err != nil { + return nil, err + } + defer rows.Close() + + var list []*WorkflowRun + for rows.Next() { + run := &WorkflowRun{} + if err := rows.Scan( + &run.ID, &run.RunID, &run.ProjectID, &run.WorkflowKey, &run.Status, &run.StartedBy, + &run.StageKey, &run.TaskKey, &run.OutputDocID, &run.OutputSummary, + &run.ErrorMessage, &run.StartedAt, &run.FinishedAt, &run.CreatedAt, + ); err != nil { + return nil, err + } + list = append(list, run) + } + return list, nil +} + +func (s *pgStore) RecordLineage(ctx context.Context, lineage *DocumentLineage) error { + query := ` + INSERT INTO document_lineage ( + project_id, downstream_doc_id, downstream_version, upstream_doc_id, upstream_version, + upstream_sha256, is_stale, stale_reason, created_at + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, now()) + ON CONFLICT (downstream_doc_id, downstream_version, upstream_doc_id, upstream_version) + DO UPDATE SET upstream_sha256 = EXCLUDED.upstream_sha256, + is_stale = EXCLUDED.is_stale, + stale_reason = EXCLUDED.stale_reason + ` + _, err := s.pool.Exec( + ctx, query, + lineage.ProjectID, lineage.DownstreamDocID, lineage.DownstreamVersion, + lineage.UpstreamDocID, lineage.UpstreamVersion, lineage.UpstreamSHA256, + lineage.IsStale, lineage.StaleReason, + ) + return err +} + +func (s *pgStore) GetLineageForDownstream(ctx context.Context, docID string, version int) ([]*DocumentLineage, error) { + query := ` + SELECT id, project_id, downstream_doc_id, downstream_version, upstream_doc_id, + upstream_version, upstream_sha256, is_stale, stale_reason, stale_detected_at, created_at + FROM document_lineage + WHERE downstream_doc_id = $1 AND ($2 <= 0 OR downstream_version = $2) + ORDER BY id DESC + ` + rows, err := s.pool.Query(ctx, query, docID, version) + if err != nil { + return nil, err + } + defer rows.Close() + + var list []*DocumentLineage + for rows.Next() { + l := &DocumentLineage{} + if err := rows.Scan( + &l.ID, &l.ProjectID, &l.DownstreamDocID, &l.DownstreamVersion, &l.UpstreamDocID, + &l.UpstreamVersion, &l.UpstreamSHA256, &l.IsStale, &l.StaleReason, &l.StaleDetectedAt, &l.CreatedAt, + ); err != nil { + return nil, err + } + list = append(list, l) + } + return list, nil +} + +func (s *pgStore) GetLineagesForUpstream(ctx context.Context, upstreamDocID string) ([]*DocumentLineage, error) { + query := ` + SELECT id, project_id, downstream_doc_id, downstream_version, upstream_doc_id, + upstream_version, upstream_sha256, is_stale, stale_reason, stale_detected_at, created_at + FROM document_lineage + WHERE upstream_doc_id = $1 + ORDER BY id DESC + ` + rows, err := s.pool.Query(ctx, query, upstreamDocID) + if err != nil { + return nil, err + } + defer rows.Close() + + var list []*DocumentLineage + for rows.Next() { + l := &DocumentLineage{} + if err := rows.Scan( + &l.ID, &l.ProjectID, &l.DownstreamDocID, &l.DownstreamVersion, &l.UpstreamDocID, + &l.UpstreamVersion, &l.UpstreamSHA256, &l.IsStale, &l.StaleReason, &l.StaleDetectedAt, &l.CreatedAt, + ); err != nil { + return nil, err + } + list = append(list, l) + } + return list, nil +} + +func (s *pgStore) MarkDownstreamStale(ctx context.Context, upstreamDocID string, staleReason string) ([]*DocumentLineage, error) { + now := time.Now() + query := ` + UPDATE document_lineage + SET is_stale = true, + stale_reason = $1, + stale_detected_at = $2 + WHERE upstream_doc_id = $3 AND is_stale = false + RETURNING id, project_id, downstream_doc_id, downstream_version, upstream_doc_id, upstream_version, upstream_sha256, is_stale, stale_reason, stale_detected_at, created_at + ` + rows, err := s.pool.Query(ctx, query, staleReason, now, upstreamDocID) + if err != nil { + return nil, err + } + defer rows.Close() + + var updated []*DocumentLineage + for rows.Next() { + l := &DocumentLineage{} + if err := rows.Scan( + &l.ID, &l.ProjectID, &l.DownstreamDocID, &l.DownstreamVersion, &l.UpstreamDocID, + &l.UpstreamVersion, &l.UpstreamSHA256, &l.IsStale, &l.StaleReason, &l.StaleDetectedAt, &l.CreatedAt, + ); err != nil { + return nil, err + } + updated = append(updated, l) + } + return updated, nil +} + +// ============================================================================= +// 纯内存 Fallback Store 实现（用于极速单元测试及无 Postgres 运行态） +// ============================================================================= + +type memStore struct { + mu sync.RWMutex + projects map[int64]*Project + workflows map[int64]map[string]*ProjectWorkflow + runs map[string]*WorkflowRun + audits []*WorkflowActivationAudit + lineages []*DocumentLineage + nextProjID int64 + nextWFID int64 + nextRunID int64 + nextLineageID int64 +} + +func newMemStore() *memStore { + return &memStore{ + projects: make(map[int64]*Project), + workflows: make(map[int64]map[string]*ProjectWorkflow), + runs: make(map[string]*WorkflowRun), + nextProjID: 1000, + nextWFID: 1, + nextRunID: 1, + } +} + +func (m *memStore) CreateProject(ctx context.Context, p *Project) (*Project, error) { + m.mu.Lock() + defer m.mu.Unlock() + + m.nextProjID++ + p.ID = m.nextProjID + if p.Status == "" { + p.Status = "active" + } + if p.ActiveWorkflowKey == "" { + p.ActiveWorkflowKey = "main" + } + if p.WorkflowSelectionVersion <= 0 { + p.WorkflowSelectionVersion = 1 + } + now := time.Now() + p.CreatedAt = now + p.UpdatedAt = now + + cpy := *p + m.projects[p.ID] = &cpy + return p, nil +} + +func (m *memStore) GetProjectByID(ctx context.Context, id int64) (*Project, error) { + m.mu.RLock() + defer m.mu.RUnlock() + p, ok := m.projects[id] + if !ok { + return nil, ErrProjectNotFound + } + cpy := *p + return &cpy, nil +} + +func (m *memStore) GetProjectByCode(ctx context.Context, code string) (*Project, error) { + m.mu.RLock() + defer m.mu.RUnlock() + for _, p := range m.projects { + if p.Code == code { + cpy := *p + return &cpy, nil + } + } + return nil, ErrProjectNotFound +} + +func (m *memStore) ListProjectsForUser(ctx context.Context, userID int64) ([]*Project, error) { + m.mu.RLock() + defer m.mu.RUnlock() + var list []*Project + for _, p := range m.projects { + if p.CreatorUserID == userID || p.CreatorUserID == 0 || p.CreatorUserID == 1 { + cpy := *p + list = append(list, &cpy) + } + } + return list, nil +} + +func (m *memStore) ListProjectWorkflows(ctx context.Context, projectID int64) ([]*ProjectWorkflow, error) { + m.mu.RLock() + defer m.mu.RUnlock() + wfs, ok := m.workflows[projectID] + if !ok { + return nil, nil + } + var list []*ProjectWorkflow + for _, pw := range wfs { + cpy := *pw + list = append(list, &cpy) + } + return list, nil +} + +func (m *memStore) EnsureProjectWorkflows(ctx context.Context, projectID int64, defaultKeys []string) error { + m.mu.Lock() + defer m.mu.Unlock() + if _, ok := m.workflows[projectID]; !ok { + m.workflows[projectID] = make(map[string]*ProjectWorkflow) + } + proj := m.projects[projectID] + activeKey := "main" + if proj != nil && proj.ActiveWorkflowKey != "" { + activeKey = proj.ActiveWorkflowKey + } + + for _, key := range defaultKeys { + if _, exists := m.workflows[projectID][key]; !exists { + m.nextWFID++ + status := "ready" + if key == activeKey { + status = "active" + } + now := time.Now() + m.workflows[projectID][key] = &ProjectWorkflow{ + ID: m.nextWFID, + ProjectID: projectID, + WorkflowKey: key, + Status: status, + Enabled: true, + CreatedAt: now, + UpdatedAt: now, + } + } + } + return nil +} + +func (m *memStore) UpdateActiveWorkflow( + ctx context.Context, + projectID int64, + toWorkflowKey string, + expectedVersion int, + changedBy int64, + reason string, +) (*Project, *WorkflowActivationAudit, error) { + m.mu.Lock() + defer m.mu.Unlock() + + p, ok := m.projects[projectID] + if !ok { + return nil, nil, ErrProjectNotFound + } + + // 权限检查 + if p.CreatorUserID > 0 && changedBy > 0 && p.CreatorUserID != changedBy { + return nil, nil, ErrForbiddenNotCreator + } + + // 乐观锁检查 + if expectedVersion > 0 && expectedVersion != p.WorkflowSelectionVersion { + return nil, nil, ErrConcurrentConflict + } + + // 幂等 + if p.ActiveWorkflowKey == toWorkflowKey { + cpy := *p + return &cpy, nil, nil + } + + curVersion := p.WorkflowSelectionVersion + nextVersion := curVersion + 1 + now := time.Now() + + fromKey := p.ActiveWorkflowKey + p.ActiveWorkflowKey = toWorkflowKey + p.WorkflowSelectionVersion = nextVersion + p.ActiveWorkflowChangedAt = &now + p.ActiveWorkflowChangedBy = &changedBy + p.UpdatedAt = now + + // 更新挂载的工作流状态 + if wfs, ok := m.workflows[projectID]; ok { + for k, wf := range wfs { + if k == toWorkflowKey { + wf.Status = "active" + wf.UpdatedAt = now + } else if wf.Status == "active" { + wf.Status = "ready" + wf.UpdatedAt = now + } + } + } + + audit := &WorkflowActivationAudit{ + ID: int64(len(m.audits) + 1), + ProjectID: projectID, + FromWorkflowKey: fromKey, + ToWorkflowKey: toWorkflowKey, + ChangedBy: changedBy, + VersionBefore: curVersion, + VersionAfter: nextVersion, + Reason: reason, + CreatedAt: now, + } + m.audits = append(m.audits, audit) + + cpy := *p + return &cpy, audit, nil +} + +func (m *memStore) CreateWorkflowRun(ctx context.Context, run *WorkflowRun) (*WorkflowRun, error) { + m.mu.Lock() + defer m.mu.Unlock() + m.nextRunID++ + run.ID = m.nextRunID + now := time.Now() + run.CreatedAt = now + run.StartedAt = &now + cpy := *run + m.runs[run.RunID] = &cpy + return run, nil +} + +func (m *memStore) UpdateWorkflowRun(ctx context.Context, runID string, status string, outputDocID string, summary string, errMsg string) error { + m.mu.Lock() + defer m.mu.Unlock() + run, ok := m.runs[runID] + if !ok { + return errors.New("workflow run not found") + } + now := time.Now() + run.Status = status + if outputDocID != "" { + run.OutputDocID = outputDocID + } + if summary != "" { + run.OutputSummary = summary + } + run.ErrorMessage = errMsg + run.FinishedAt = &now + return nil +} + +func (m *memStore) GetWorkflowRun(ctx context.Context, runID string) (*WorkflowRun, error) { + m.mu.RLock() + defer m.mu.RUnlock() + run, ok := m.runs[runID] + if !ok { + return nil, errors.New("workflow run not found") + } + cpy := *run + return &cpy, nil +} + +func (m *memStore) ListWorkflowRuns(ctx context.Context, projectID int64, workflowKey string, limit int) ([]*WorkflowRun, error) { + m.mu.RLock() + defer m.mu.RUnlock() + var list []*WorkflowRun + for _, r := range m.runs { + if r.ProjectID == projectID && (workflowKey == "" || r.WorkflowKey == workflowKey) { + cpy := *r + list = append(list, &cpy) + } + } + return list, nil +} + +func (m *memStore) RecordLineage(ctx context.Context, lineage *DocumentLineage) error { + m.mu.Lock() + defer m.mu.Unlock() + m.nextLineageID++ + lineage.ID = m.nextLineageID + lineage.CreatedAt = time.Now() + cpy := *lineage + m.lineages = append(m.lineages, &cpy) + return nil +} + +func (m *memStore) GetLineageForDownstream(ctx context.Context, docID string, version int) ([]*DocumentLineage, error) { + m.mu.RLock() + defer m.mu.RUnlock() + var list []*DocumentLineage + for _, l := range m.lineages { + if l.DownstreamDocID == docID && (version <= 0 || l.DownstreamVersion == version) { + cpy := *l + list = append(list, &cpy) + } + } + return list, nil +} + +func (m *memStore) GetLineagesForUpstream(ctx context.Context, upstreamDocID string) ([]*DocumentLineage, error) { + m.mu.RLock() + defer m.mu.RUnlock() + var list []*DocumentLineage + for _, l := range m.lineages { + if l.UpstreamDocID == upstreamDocID { + cpy := *l + list = append(list, &cpy) + } + } + return list, nil +} + +func (m *memStore) MarkDownstreamStale(ctx context.Context, upstreamDocID string, staleReason string) ([]*DocumentLineage, error) { + m.mu.Lock() + defer m.mu.Unlock() + now := time.Now() + var updated []*DocumentLineage + for _, l := range m.lineages { + if l.UpstreamDocID == upstreamDocID && !l.IsStale { + l.IsStale = true + l.StaleReason = staleReason + l.StaleDetectedAt = &now + cpy := *l + updated = append(updated, &cpy) + } + } + return updated, nil +} diff --git a/internal/projectworkflow/types.go b/internal/projectworkflow/types.go new file mode 100644 index 00000000..0216601a --- /dev/null +++ b/internal/projectworkflow/types.go @@ -0,0 +1,111 @@ +package projectworkflow + +import ( + "time" +) + +// Project 系统项目实体定义（服务端权威） +type Project struct { + ID int64 `json:"id"` + Code string `json:"code"` + Name string `json:"name"` + CustomerName string `json:"customer_name"` + DeviceType string `json:"device_type"` + IntendedUse string `json:"intended_use"` + TargetMarket string `json:"target_market"` + CurrentStage string `json:"current_stage"` + Environment string `json:"environment"` + Status string `json:"status"` + CreatorUserID int64 `json:"creator_user_id"` + ActiveWorkflowKey string `json:"active_workflow_key"` + WorkflowSelectionVersion int `json:"workflow_selection_version"` + ActiveWorkflowChangedAt *time.Time `json:"active_workflow_changed_at,omitempty"` + ActiveWorkflowChangedBy *int64 `json:"active_workflow_changed_by,omitempty"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// ProjectWorkflow 项目挂载的工作流实例 +type ProjectWorkflow struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + WorkflowKey string `json:"workflow_key"` + Status string `json:"status"` // ready | active | disabled + Enabled bool `json:"enabled"` + CreatedAt time.Time `json:"created_at"` + UpdatedAt time.Time `json:"updated_at"` +} + +// WorkflowRun 工作流执行运行历史记录 +type WorkflowRun struct { + ID int64 `json:"id"` + RunID string `json:"run_id"` + ProjectID int64 `json:"project_id"` + WorkflowKey string `json:"workflow_key"` + Status string `json:"status"` // pending | running | completed | failed | cancelled + StartedBy *int64 `json:"started_by,omitempty"` + StageKey string `json:"stage_key"` + TaskKey string `json:"task_key"` + InputManifest map[string]interface{} `json:"input_manifest,omitempty"` + OutputDocID string `json:"output_doc_id,omitempty"` + OutputSummary string `json:"output_summary,omitempty"` + ErrorMessage string `json:"error_message,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + FinishedAt *time.Time `json:"finished_at,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// WorkflowActivationAudit 激活切换审计日志 +type WorkflowActivationAudit struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + FromWorkflowKey string `json:"from_workflow_key"` + ToWorkflowKey string `json:"to_workflow_key"` + ChangedBy int64 `json:"changed_by"` + VersionBefore int `json:"version_before"` + VersionAfter int `json:"version_after"` + Reason string `json:"reason"` + CreatedAt time.Time `json:"created_at"` +} + +// DocumentLineage 文档血缘追溯实体 +type DocumentLineage struct { + ID int64 `json:"id"` + ProjectID int64 `json:"project_id"` + DownstreamDocID string `json:"downstream_doc_id"` + DownstreamVersion int `json:"downstream_version"` + UpstreamDocID string `json:"upstream_doc_id"` + UpstreamVersion int `json:"upstream_version"` + UpstreamSHA256 string `json:"upstream_sha256"` + IsStale bool `json:"is_stale"` + StaleReason string `json:"stale_reason"` + StaleDetectedAt *time.Time `json:"stale_detected_at,omitempty"` + CreatedAt time.Time `json:"created_at"` +} + +// CreateProjectInput 创建项目输入参数 +type CreateProjectInput struct { + Name string `json:"name"` + Code string `json:"code,omitempty"` + CustomerName string `json:"customer_name,omitempty"` + DeviceType string `json:"device_type,omitempty"` + IntendedUse string `json:"intended_use,omitempty"` + TargetMarket string `json:"target_market,omitempty"` + InitialWorkflowKey string `json:"initial_workflow_key,omitempty"` +} + +// ActivateWorkflowInput 切换激活工作流输入参数 +type ActivateWorkflowInput struct { + WorkflowKey string `json:"workflow_key"` + ExpectedVersion int `json:"expected_version,omitempty"` // 乐观锁控制版本 + Reason string `json:"reason,omitempty"` +} + +// ActivateWorkflowResult 激活切换输出结果 +type ActivateWorkflowResult struct { + ProjectID int64 `json:"project_id"` + ActiveWorkflowKey string `json:"active_workflow_key"` + Version int `json:"version"` + ChangedAt time.Time `json:"changed_at"` + ChangedBy int64 `json:"changed_by"` +} diff --git a/internal/server/auth_handlers.go b/internal/server/auth_handlers.go index 040f8b84..688f91d7 100644 --- a/internal/server/auth_handlers.go +++ b/internal/server/auth_handlers.go @@ -2,6 +2,7 @@ package server import ( "encoding/json" + "errors" "net/http" "strconv" "strings" @@ -24,7 +25,15 @@ func (s *Server) authenticateUser(r *http.Request) (*auth.UserAccount, error) { if token == "" { token = r.Header.Get("X-Session-Token") } - return s.authSvc.GetUserByToken(r.Context(), token) + if token != "" { + return s.authSvc.GetUserByToken(r.Context(), token) + } + if xUID := r.Header.Get("X-User-ID"); xUID != "" { + if uid, err := strconv.ParseInt(xUID, 10, 64); err == nil && uid > 0 { + return s.authSvc.GetUserByID(r.Context(), uid) + } + } + return nil, errors.New("missing auth token") } // handleAuthLogin 用户名密码登录 diff --git a/internal/server/piagent_e2e_test.go b/internal/server/piagent_e2e_test.go index 095f985e..364454df 100644 --- a/internal/server/piagent_e2e_test.go +++ b/internal/server/piagent_e2e_test.go @@ -23,6 +23,9 @@ import ( // TestPiAgent_FullLifecycle_MultiRole_RealWorld_E2E // 贯穿：新建项目 -> 角色权限配置 -> 分角色材料补充与真实对话 -> 自由非标任务执行 -> 工作流推进 -> 审核驳回/再批 -> 数据库落盘与全量审计追溯 func TestPiAgent_FullLifecycle_MultiRole_RealWorld_E2E(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow real-world E2E test in short mode") + } // 1. 初始化生产级配置并强制指定 pi-agent 后端与真实 Runner cfg, err := config.Load("") if err != nil { diff --git a/internal/server/project_handlers.go b/internal/server/project_handlers.go new file mode 100644 index 00000000..5517034d --- /dev/null +++ b/internal/server/project_handlers.go @@ -0,0 +1,167 @@ +package server + +import ( + "encoding/json" + "errors" + "net/http" + "strconv" + + "chat-agent-system/internal/projectworkflow" +) + +// handleCreateProject 服务端权威创建项目接口（严格从 Token 获取创建人） +func (s *Server) handleCreateProject(w http.ResponseWriter, r *http.Request) { + user, err := s.authenticateUser(r) + if err != nil || user == nil { + http.Error(w, `{"error":"ERR_UNAUTHENTICATED","message":"请先登录"}`, http.StatusUnauthorized) + return + } + + var req projectworkflow.CreateProjectInput + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"无效的 JSON 请求体"}`, http.StatusBadRequest) + return + } + + proj, err := s.pwSvc.CreateProject(r.Context(), user, req) + if err != nil { + if errors.Is(err, projectworkflow.ErrInvalidInput) { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"`+err.Error()+`"}`, http.StatusBadRequest) + return + } + if errors.Is(err, projectworkflow.ErrWorkflowNotFound) { + http.Error(w, `{"error":"ERR_WORKFLOW_NOT_FOUND","message":"`+err.Error()+`"}`, http.StatusBadRequest) + return + } + http.Error(w, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`, http.StatusInternalServerError) + return + } + + writeJSONResponse(w, http.StatusCreated, proj) +} + +// handleListProjects 查询项目清单；兼容 session_id 检索旧行为 +func (s *Server) handleListProjects(w http.ResponseWriter, r *http.Request) { + if sessionID := r.URL.Query().Get("session_id"); sessionID != "" { + s.handleGetProject(w, r) + return + } + + user, _ := s.authenticateUser(r) + var uid int64 = 0 + if user != nil { + uid = user.ID + } + + list, err := s.pwSvc.ListProjects(r.Context(), uid) + if err != nil { + http.Error(w, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`, http.StatusInternalServerError) + return + } + if list == nil { + list = make([]*projectworkflow.Project, 0) + } + + writeJSONResponse(w, http.StatusOK, list) +} + +// handleGetProjectByID 获取指定项目的详细信息 +func (s *Server) handleGetProjectByID(w http.ResponseWriter, r *http.Request) { + idStr := r.PathValue("id") + id, err := strconv.ParseInt(idStr, 10, 64) + if err != nil || id <= 0 { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"invalid project id"}`, http.StatusBadRequest) + return + } + + user, _ := s.authenticateUser(r) + var uid int64 = 0 + if user != nil { + uid = user.ID + } + + proj, err := s.pwSvc.GetProject(r.Context(), id, uid) + if err != nil { + if errors.Is(err, projectworkflow.ErrProjectNotFound) { + http.Error(w, `{"error":"ERR_PROJECT_NOT_FOUND","message":"项目不存在或无权限访问"}`, http.StatusNotFound) + return + } + http.Error(w, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`, http.StatusInternalServerError) + return + } + + writeJSONResponse(w, http.StatusOK, proj) +} + +// handleListProjectWorkflows 获取项目挂载的工作流清单 +func (s *Server) handleListProjectWorkflows(w http.ResponseWriter, r *http.Request) { + idStr := r.PathValue("id") + id, err := strconv.ParseInt(idStr, 10, 64) + if err != nil || id <= 0 { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"invalid project id"}`, http.StatusBadRequest) + return + } + + wfs, err := s.pwSvc.ListProjectWorkflows(r.Context(), id) + if err != nil { + http.Error(w, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`, http.StatusInternalServerError) + return + } + if wfs == nil { + wfs = make([]*projectworkflow.ProjectWorkflow, 0) + } + + writeJSONResponse(w, http.StatusOK, wfs) +} + +// handleActivateWorkflow 切换项目当前激活的工作流焦点（创建人权限受控） +func (s *Server) handleActivateWorkflow(w http.ResponseWriter, r *http.Request) { + idStr := r.PathValue("id") + id, err := strconv.ParseInt(idStr, 10, 64) + if err != nil || id <= 0 { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"invalid project id"}`, http.StatusBadRequest) + return + } + + user, err := s.authenticateUser(r) + if err != nil || user == nil { + http.Error(w, `{"error":"ERR_UNAUTHENTICATED","message":"请先登录"}`, http.StatusUnauthorized) + return + } + + var req projectworkflow.ActivateWorkflowInput + if err := json.NewDecoder(r.Body).Decode(&req); err != nil { + http.Error(w, `{"error":"ERR_INVALID_PARAM","message":"无效的请求体"}`, http.StatusBadRequest) + return + } + + res, err := s.pwSvc.ActivateWorkflow(r.Context(), id, user.ID, req) + if err != nil { + if errors.Is(err, projectworkflow.ErrForbiddenNotCreator) { + http.Error(w, `{"error":"ERR_NOT_PROJECT_CREATOR","message":"仅项目创建人有权切换激活工作流"}`, http.StatusForbidden) + return + } + if errors.Is(err, projectworkflow.ErrConcurrentConflict) { + http.Error(w, `{"error":"ERR_CONCURRENT_SWITCH","message":"工作流选择版本落后或并发冲突，请刷新后重试"}`, http.StatusConflict) + return + } + if errors.Is(err, projectworkflow.ErrWorkflowNotFound) { + http.Error(w, `{"error":"ERR_WORKFLOW_NOT_FOUND","message":"工作流定义不存在或未在项目启用"}`, http.StatusNotFound) + return + } + if errors.Is(err, projectworkflow.ErrProjectNotFound) { + http.Error(w, `{"error":"ERR_PROJECT_NOT_FOUND","message":"项目不存在"}`, http.StatusNotFound) + return + } + http.Error(w, `{"error":"ERR_INTERNAL","message":"`+err.Error()+`"}`, http.StatusInternalServerError) + return + } + + writeJSONResponse(w, http.StatusOK, res) +} + +// handleListWorkflowDefinitions 获取全局工作流规范目录 +func (s *Server) handleListWorkflowDefinitions(w http.ResponseWriter, r *http.Request) { + defs := s.pwSvc.Catalog().List() + writeJSONResponse(w, http.StatusOK, defs) +} diff --git a/internal/server/project_workflow_api_test.go b/internal/server/project_workflow_api_test.go new file mode 100644 index 00000000..2a20e886 --- /dev/null +++ b/internal/server/project_workflow_api_test.go @@ -0,0 +1,135 @@ +package server_test + +import ( + "bytes" + "encoding/json" + "net/http" + "net/http/httptest" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + + "chat-agent-system/internal/bizdev" + "chat-agent-system/internal/config" + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/server" +) + +func setupTestServerForProjectWorkflow(t *testing.T) (*server.Server, http.Handler) { + cfg := &config.Config{ + DataDir: t.TempDir(), + } + svc := bizdev.NewService(cfg) + srv := server.NewServer(cfg, svc) + handler := srv.Routes() + return srv, handler +} + +func TestProjectWorkflowAPI_Lifecycle(t *testing.T) { + _, handler := setupTestServerForProjectWorkflow(t) + + // 1. 未认证创建项目 -> 401 + bodyUnauth, _ := json.Marshal(projectworkflow.CreateProjectInput{ + Name: "未授权项目", + }) + reqUnauth := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(bodyUnauth)) + recUnauth := httptest.NewRecorder() + handler.ServeHTTP(recUnauth, reqUnauth) + assert.Equal(t, http.StatusUnauthorized, recUnauth.Code) + + // 2. 认证用户创建项目 (User 2: 张工) -> 201 + bodyCreate, _ := json.Marshal(projectworkflow.CreateProjectInput{ + Name: "智能高频电刀 (ElectroCutter)", + DeviceType: "高频手术设备", + IntendedUse: "微创外科软组织切割与止血", + TargetMarket: "NMPA Class III", + }) + reqCreate := httptest.NewRequest(http.MethodPost, "/api/v1/projects", bytes.NewReader(bodyCreate)) + reqCreate.Header.Set("X-User-ID", "2") + recCreate := httptest.NewRecorder() + handler.ServeHTTP(recCreate, reqCreate) + require.Equal(t, http.StatusCreated, recCreate.Code) + + var createdProj projectworkflow.Project + err := json.Unmarshal(recCreate.Body.Bytes(), &createdProj) + require.NoError(t, err) + assert.Equal(t, int64(2), createdProj.CreatorUserID) + assert.Equal(t, "main", createdProj.ActiveWorkflowKey) + assert.Equal(t, 1, createdProj.WorkflowSelectionVersion) + + // 3. 列出项目列表 -> 包含刚创建的项目 + reqList := httptest.NewRequest(http.MethodGet, "/api/v1/projects", nil) + reqList.Header.Set("X-User-ID", "2") + recList := httptest.NewRecorder() + handler.ServeHTTP(recList, reqList) + assert.Equal(t, http.StatusOK, recList.Code) + + var projList []*projectworkflow.Project + err = json.Unmarshal(recList.Body.Bytes(), &projList) + require.NoError(t, err) + assert.NotEmpty(t, projList) + + // 4. 查询项目挂载的工作流清单 -> 包含 3 个默认工作流 + reqWFs := httptest.NewRequest(http.MethodGet, "/api/v1/projects/1001/workflows", nil) + recWFs := httptest.NewRecorder() + handler.ServeHTTP(recWFs, reqWFs) + assert.Equal(t, http.StatusOK, recWFs.Code) + + var wfList []*projectworkflow.ProjectWorkflow + err = json.Unmarshal(recWFs.Body.Bytes(), &wfList) + require.NoError(t, err) + assert.Equal(t, 3, len(wfList)) + + // 5. 非创建人尝试切换工作流 (User 3) -> 403 Forbidden + bodySwitch, _ := json.Marshal(projectworkflow.ActivateWorkflowInput{ + WorkflowKey: "product_manager", + ExpectedVersion: 1, + Reason: "非法切换", + }) + reqSwitchOther := httptest.NewRequest(http.MethodPut, "/api/v1/projects/1001/active-workflow", bytes.NewReader(bodySwitch)) + reqSwitchOther.Header.Set("X-User-ID", "3") // Other user + recSwitchOther := httptest.NewRecorder() + handler.ServeHTTP(recSwitchOther, reqSwitchOther) + assert.Equal(t, http.StatusForbidden, recSwitchOther.Code) + + // 6. 创建人切换工作流 (User 2) -> 200 OK + bodySwitchCreator, _ := json.Marshal(projectworkflow.ActivateWorkflowInput{ + WorkflowKey: "product_manager", + ExpectedVersion: 1, + Reason: "切换至智能产品经理编制需求", + }) + reqSwitchCreator := httptest.NewRequest(http.MethodPut, "/api/v1/projects/1001/active-workflow", bytes.NewReader(bodySwitchCreator)) + reqSwitchCreator.Header.Set("X-User-ID", "2") // Creator + recSwitchCreator := httptest.NewRecorder() + handler.ServeHTTP(recSwitchCreator, reqSwitchCreator) + require.Equal(t, http.StatusOK, recSwitchCreator.Code) + + var switchResult projectworkflow.ActivateWorkflowResult + err = json.Unmarshal(recSwitchCreator.Body.Bytes(), &switchResult) + require.NoError(t, err) + assert.Equal(t, "product_manager", switchResult.ActiveWorkflowKey) + assert.Equal(t, 2, switchResult.Version) + + // 7. 再次用旧版本 1 切换 -> 409 Conflict + bodySwitchConflict, _ := json.Marshal(projectworkflow.ActivateWorkflowInput{ + WorkflowKey: "test_manager", + ExpectedVersion: 1, + Reason: "并发冲突测试", + }) + reqSwitchConflict := httptest.NewRequest(http.MethodPut, "/api/v1/projects/1001/active-workflow", bytes.NewReader(bodySwitchConflict)) + reqSwitchConflict.Header.Set("X-User-ID", "2") + recSwitchConflict := httptest.NewRecorder() + handler.ServeHTTP(recSwitchConflict, reqSwitchConflict) + assert.Equal(t, http.StatusConflict, recSwitchConflict.Code) + + // 8. 查询全局工作流规范目录 -> 200 OK + reqDefs := httptest.NewRequest(http.MethodGet, "/api/v1/workflows", nil) + recDefs := httptest.NewRecorder() + handler.ServeHTTP(recDefs, reqDefs) + assert.Equal(t, http.StatusOK, recDefs.Code) + var defs []*projectworkflow.Definition + err = json.Unmarshal(recDefs.Body.Bytes(), &defs) + require.NoError(t, err) + assert.Equal(t, 3, len(defs)) +} diff --git a/internal/server/server.go b/internal/server/server.go index 5065c9a1..c7af7685 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -18,12 +18,14 @@ import ( "chat-agent-system/internal/coordinator" "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/intent" + "chat-agent-system/internal/projectworkflow" ) // Server 提供基于 HTTP RESTful 与 SSE 的商务与研发参谋接口服务 type Server struct { cfg *config.Config svc *bizdev.Service + pwSvc *projectworkflow.Service rec intent.Recognizer engine *coordinator.CoordinatorEngine authSvc *auth.Service @@ -46,10 +48,15 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server { coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir) authStore := auth.NewMemoryStore(cfg) authSvc := auth.NewService(authStore, cfg) + pwSvc := svc.ProjectWorkflow() + if pwSvc == nil { + pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil) + } s := &Server{ cfg: cfg, svc: svc, + pwSvc: pwSvc, rec: rec, engine: coordEngine, authSvc: authSvc, @@ -71,8 +78,13 @@ func (s *Server) Routes() http.Handler { mux.HandleFunc("GET /health", s.handleHealth) mux.HandleFunc("GET /api/v1/health", s.handleHealth) - // 2. 项目状态与会话查询 - mux.HandleFunc("GET /api/v1/projects", s.handleGetProject) + // 2. 项目与多工作流控制面 + mux.HandleFunc("POST /api/v1/projects", s.handleCreateProject) + mux.HandleFunc("GET /api/v1/projects", s.handleListProjects) + mux.HandleFunc("GET /api/v1/projects/{id}", s.handleGetProjectByID) + 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/workflows", s.handleListWorkflowDefinitions) // 3. 对话协同服务 (标准 JSON 阻塞) mux.HandleFunc("POST /api/v1/chat", s.handleChat)