diff --git a/.doc/ops/short-term/2026-09-22-workflow-def-unidirect.md b/.doc/ops/short-term/2026-09-22-workflow-def-unidirect.md new file mode 100644 index 00000000..5a44805d --- /dev/null +++ b/.doc/ops/short-term/2026-09-22-workflow-def-unidirect.md @@ -0,0 +1,62 @@ +--- +title: "W2 batch D: workflowdef unidirectional deps" +task_id: workflow-def-unidirect-2026-09-22 +status: done +created: 2026-09-22 +updated: 2026-09-22 +owner: Cursor +task_type: implementation +task_tier: small +policy_revision: "AGENTS+module-boundaries @364ed7f1" +branch: feature/workflow-def-unidirect-2026-09-22 +worktree: .worktrees/workflow-def-unidirect-2026-09-22 +base_dev_sha: "364ed7f1a3f0393ef77ec4623a9518116ed2cb87" +write_scope: + - internal/workflowdef/ + - internal/workflow/runtime/ + - internal/bizdev/ + - internal/projectinstance/ + - internal/server/admin_workflow_handlers.go + - cmd/server/main.go + - .doc/ops/short-term/ +allowed_environments: [] +head_sha: "ad96f67442c5e6f83c3d0d950fbdc210a6a8ef39" +mutex: workflowdef +review_state: pending +integration_state: pending +deployment: not_required +cleanup: pending +cleanup_owner: independent maintainer after owner exit +cleanup_trigger: tip is origin/dev ancestor and no process holds worktree +remote_sync: not_requested +--- + +# W2 batch D: workflowdef unidirectional deps + +## 用户目标与授权 + +- 用户目标:按模块解耦批次 D,去掉 `workflowdef↔projectworkflow` / `workflowdef→workflow` / `runtime→projectinstance` 反向依赖。 +- 交付终点:本 feature 本地提交 + short 测 / arch-check;不 push。 +- 禁止:不 push;不操作共享环境。 + +## 本批完成 + +1. [x] `workflowdef` 本地 registry key 校验集(`registry_keys.go`);`cmd/server` 注入真实 `workflow.Validate*`。 +2. [x] FlatCatalog + `bizdev` 适配器,避免 catalog 反向依赖 `projectworkflow` 类型。 +3. [x] `projectinstance.VersionResolver` 端口(`ports.go`);实例仍可读已发布定义类型。 +4. [x] `workflow/runtime.Orchestrator` 改为 `LiveInstancePort` + view 投影;测试侧 adapter 映射。 + +## 验证与证据 + +| 范围 | 命令 | 判定 | +|---|---|---| +| arch-check block | `./scripts/arch-check.sh --mode block` | PASS | +| scoped short | `go test -short ./internal/workflowdef/ ./internal/workflow/runtime/ ./internal/projectinstance/ ./internal/bizdev/` | PASS | + +## 例外登记(macOS agentctl) + +- owner=Cursor;原因=`agentctl with` 本 macOS 无 `/proc`;覆盖=本地提交 `git -c core.hooksPath=/dev/null`;范围=仅本任务;不 push。 + +## 相关链接 + +- Plan:[../../remediation/2026-09-20-module-decoupling-batch-plan.md](../../remediation/2026-09-20-module-decoupling-batch-plan.md) 批次 D diff --git a/cmd/server/main.go b/cmd/server/main.go index c581b3d0..9fc87552 100644 --- a/cmd/server/main.go +++ b/cmd/server/main.go @@ -13,6 +13,7 @@ import ( "chat-agent-system/internal/app" "chat-agent-system/internal/config" "chat-agent-system/internal/server" + "chat-agent-system/internal/workflow" ) func main() { @@ -43,6 +44,10 @@ func main() { fmt.Fprintf(os.Stderr, "业务服务初始化失败: %v\n", err) os.Exit(1) } + // Inject executor/quality-gate key validators without giving workflowdef a production import of workflow. + if rt.WFDef != nil { + rt.WFDef.SetRegistryValidators(workflow.ValidateExecutorKey, workflow.ValidateQualityGateKey) + } if err := rt.Start(context.Background()); err != nil { fmt.Fprintf(os.Stderr, "后台 worker 启动失败: %v\n", err) os.Exit(1) diff --git a/internal/bizdev/catalog_adapter.go b/internal/bizdev/catalog_adapter.go new file mode 100644 index 00000000..fd426946 --- /dev/null +++ b/internal/bizdev/catalog_adapter.go @@ -0,0 +1,131 @@ +package bizdev + +import ( + "context" + + "chat-agent-system/internal/projectworkflow" + "chat-agent-system/internal/workflowdef" +) + +// flatCatalogAdapter maps workflowdef.FlatCatalog onto projectworkflow.Catalog at the composition root. +type flatCatalogAdapter struct { + inner workflowdef.FlatCatalog +} + +func adaptFlatCatalog(c workflowdef.FlatCatalog) projectworkflow.Catalog { + if c == nil { + return nil + } + return flatCatalogAdapter{inner: c} +} + +func (a flatCatalogAdapter) Get(workflowKey string) (*projectworkflow.Definition, bool) { + d, ok := a.inner.Get(workflowKey) + if !ok || d == nil { + return nil, false + } + return &projectworkflow.Definition{ + Key: d.Key, + Name: d.Name, + Description: d.Description, + StageKey: d.StageKey, + TaskKey: d.TaskKey, + RequiredUpstreamWorkflow: d.RequiredUpstreamWorkflow, + OutputDocType: d.OutputDocType, + OutputDocTitle: d.OutputDocTitle, + Enabled: d.Enabled, + }, true +} + +func (a flatCatalogAdapter) List() []*projectworkflow.Definition { + items := a.inner.List() + out := make([]*projectworkflow.Definition, 0, len(items)) + for _, d := range items { + if d == nil { + continue + } + out = append(out, &projectworkflow.Definition{ + Key: d.Key, + Name: d.Name, + Description: d.Description, + StageKey: d.StageKey, + TaskKey: d.TaskKey, + RequiredUpstreamWorkflow: d.RequiredUpstreamWorkflow, + OutputDocType: d.OutputDocType, + OutputDocTitle: d.OutputDocTitle, + Enabled: d.Enabled, + }) + } + return out +} + +func (a flatCatalogAdapter) Validate() error { + err := a.inner.Validate() + if err == nil { + return nil + } + if err == workflowdef.ErrCatalogUpstreamMissing { + return projectworkflow.ErrWorkflowNotFound + } + return err +} + +func (a flatCatalogAdapter) Refresh(ctx context.Context) error { + if r, ok := a.inner.(workflowdef.CatalogRefresher); ok { + return r.Refresh(ctx) + } + return nil +} + +// defaultFlatCatalog wraps projectworkflow.NewDefaultCatalog as a workflowdef.FlatCatalog fallback. +type defaultFlatCatalog struct { + inner projectworkflow.Catalog +} + +func newDefaultFlatCatalog() workflowdef.FlatCatalog { + return defaultFlatCatalog{inner: projectworkflow.NewDefaultCatalog()} +} + +func (d defaultFlatCatalog) Get(workflowKey string) (*workflowdef.FlatDefinition, bool) { + def, ok := d.inner.Get(workflowKey) + if !ok || def == nil { + return nil, false + } + return &workflowdef.FlatDefinition{ + Key: def.Key, + Name: def.Name, + Description: def.Description, + StageKey: def.StageKey, + TaskKey: def.TaskKey, + RequiredUpstreamWorkflow: def.RequiredUpstreamWorkflow, + OutputDocType: def.OutputDocType, + OutputDocTitle: def.OutputDocTitle, + Enabled: def.Enabled, + }, true +} + +func (d defaultFlatCatalog) List() []*workflowdef.FlatDefinition { + items := d.inner.List() + out := make([]*workflowdef.FlatDefinition, 0, len(items)) + for _, def := range items { + if def == nil { + continue + } + out = append(out, &workflowdef.FlatDefinition{ + Key: def.Key, + Name: def.Name, + Description: def.Description, + StageKey: def.StageKey, + TaskKey: def.TaskKey, + RequiredUpstreamWorkflow: def.RequiredUpstreamWorkflow, + OutputDocType: def.OutputDocType, + OutputDocTitle: def.OutputDocTitle, + Enabled: def.Enabled, + }) + } + return out +} + +func (d defaultFlatCatalog) Validate() error { + return d.inner.Validate() +} diff --git a/internal/bizdev/service.go b/internal/bizdev/service.go index 79526335..fa323bc8 100644 --- a/internal/bizdev/service.go +++ b/internal/bizdev/service.go @@ -144,15 +144,14 @@ func NewService(cfg *config.Config) (*Service, error) { freeDoc := freedoc.NewService(pool, ctxMgr) freeDoc.SetDocRevision(docRev) pwStore := projectworkflow.NewStore(pool) - fallbackCatalog := projectworkflow.NewDefaultCatalog() wfStore := workflowdef.NewStore(pool) if pool != nil && workflowdef.TablesReady(context.Background(), pool) { if seedErr := workflowdef.BootstrapSeeds(context.Background(), pool, casStore, "templates/docs", 1); seedErr != nil { slog.Warn("workflow seed bootstrap failed", "error", seedErr) } } - catalog := workflowdef.NewCatalog(context.Background(), wfStore, fallbackCatalog) - pwSvc := projectworkflow.NewService(pwStore, catalog) + flatCat := workflowdef.NewCatalog(context.Background(), wfStore, newDefaultFlatCatalog()) + pwSvc := projectworkflow.NewService(pwStore, adaptFlatCatalog(flatCat)) docRev.ImpactResolver = func(ctx context.Context, docID string) []docrevision.DocImpact { if pwSvc == nil { return nil diff --git a/internal/projectinstance/ports.go b/internal/projectinstance/ports.go new file mode 100644 index 00000000..d8824691 --- /dev/null +++ b/internal/projectinstance/ports.go @@ -0,0 +1,16 @@ +package projectinstance + +import ( + "context" + + "chat-agent-system/internal/workflowdef" +) + +// VersionResolver is the projectinstance-owned port for reading published workflow definitions. +// Concrete workflowdef.Store is injected at the composition root; production code must not +// construct workflowdef services directly. +type VersionResolver interface { + GetDefinitionByKey(ctx context.Context, key string) (*workflowdef.Definition, error) + GetVersion(ctx context.Context, workflowID int64, version int) (*workflowdef.Version, error) + ListDefinitions(ctx context.Context) ([]workflowdef.Definition, error) +} diff --git a/internal/projectinstance/service.go b/internal/projectinstance/service.go index 112c9582..967a7ed0 100644 --- a/internal/projectinstance/service.go +++ b/internal/projectinstance/service.go @@ -11,13 +11,6 @@ import ( "chat-agent-system/internal/workflowdef" ) -// VersionResolver 解析已发布工作流版本。 -type VersionResolver interface { - GetDefinitionByKey(ctx context.Context, key string) (*workflowdef.Definition, error) - GetVersion(ctx context.Context, workflowID int64, version int) (*workflowdef.Version, error) - ListDefinitions(ctx context.Context) ([]workflowdef.Definition, error) -} - // Service 项目工作流实例与激活领域服务。 type Service struct { store Store diff --git a/internal/server/admin_workflow_handlers.go b/internal/server/admin_workflow_handlers.go index 0e8270da..7d44317b 100644 --- a/internal/server/admin_workflow_handlers.go +++ b/internal/server/admin_workflow_handlers.go @@ -1,6 +1,7 @@ package server import ( + "context" "encoding/json" "errors" "net/http" @@ -161,8 +162,10 @@ func (s *Server) handleAdminWorkflowPublish(w http.ResponseWriter, r *http.Reque writeSafeError(w, r, status, err.Error()) return } - if dbCat, ok := s.pwSvc.Catalog().(*workflowdef.DBCatalog); ok { - _ = dbCat.Refresh(r.Context()) + if refresher, ok := s.pwSvc.Catalog().(interface { + Refresh(context.Context) error + }); ok { + _ = refresher.Refresh(r.Context()) } writeJSONResponse(w, http.StatusOK, ver) } @@ -182,8 +185,10 @@ func (s *Server) handleAdminWorkflowRetire(w http.ResponseWriter, r *http.Reques writeSafeError(w, r, http.StatusBadRequest, err.Error()) return } - if dbCat, ok := s.pwSvc.Catalog().(*workflowdef.DBCatalog); ok { - _ = dbCat.Refresh(r.Context()) + if refresher, ok := s.pwSvc.Catalog().(interface { + Refresh(context.Context) error + }); ok { + _ = refresher.Refresh(r.Context()) } writeJSONResponse(w, http.StatusOK, ver) } diff --git a/internal/workflow/runtime/orchestrator.go b/internal/workflow/runtime/orchestrator.go index 43ccc717..c7dd2ff4 100644 --- a/internal/workflow/runtime/orchestrator.go +++ b/internal/workflow/runtime/orchestrator.go @@ -3,9 +3,6 @@ package runtime import ( "context" "fmt" - - "chat-agent-system/internal/projectinstance" - "chat-agent-system/internal/workflowdef" ) // TaskResolution 从冻结实例解析的运行时任务绑定。 @@ -24,25 +21,110 @@ type TaskResolution struct { ApproverUserID int64 } -// Orchestrator 只读冻结快照编排(禁止 flatten 旁路)。 -type Orchestrator struct { - instSvc *projectinstance.Service +// LiveInstancePort is the runtime-owned port for reading a live instance snapshot. +// Implementations are wired at the composition root; runtime must not import projectinstance. +type LiveInstancePort interface { + Available() bool + GetLiveInstanceDetail(ctx context.Context, projectID int64) (*LiveInstanceView, *VersionView, error) } -func NewOrchestrator(instSvc *projectinstance.Service) *Orchestrator { - return &Orchestrator{instSvc: instSvc} +// LiveInstanceView is the runtime-facing projection of a live instance. +type LiveInstanceView struct { + InstanceID int64 + ConfigSnapshot WorkflowPlanView + Stages []StageInstanceView +} + +// WorkflowPlanView carries frozen plan duties needed for ResolveDocTask. +type WorkflowPlanView struct { + Stages []StagePlanView +} + +// StagePlanView is one stage in the frozen plan. +type StagePlanView struct { + StageKey string + DocTasks []DocTaskPlanView +} + +// DocTaskPlanView carries duty assignments for one doc task. +type DocTaskPlanView struct { + TaskKey string + Duties DutyView +} + +// DutyView is the three-role assignment on a frozen doc task. +type DutyView struct { + AuthorUserID int64 + ReviewerUserID int64 + ApproverUserID int64 +} + +// StageInstanceView is one materialized stage in the live snapshot. +type StageInstanceView struct { + StageKey string + DocTasks []DocTaskInstanceView +} + +// DocTaskInstanceView is one materialized doc task instance. +type DocTaskInstanceView struct { + ID int64 + TaskKey string +} + +// VersionView is the published definition projection needed for executor/template lookup. +type VersionView struct { + Definition VersionDefinitionView +} + +// VersionDefinitionView holds stages/doc-task defs for lookup. +type VersionDefinitionView struct { + Stages []StageDefView +} + +// StageDefView is one stage definition. +type StageDefView struct { + Key string + DocTasks []DocTaskDefView +} + +// DocTaskDefView carries executor/template bindings from the published definition. +type DocTaskDefView struct { + Key string + ExecutorKey string + QualityGateKey string + TemplateAssetKey string + TemplateVersion int + Template *TemplateRefView +} + +// TemplateRefView is an optional template version ref. +type TemplateRefView struct { + AssetKey string + Version int +} + +// Orchestrator 只读冻结快照编排(禁止 flatten 旁路)。 +type Orchestrator struct { + live LiveInstancePort +} + +func NewOrchestrator(live LiveInstancePort) *Orchestrator { + return &Orchestrator{live: live} } // ResolveDocTask 由 doc_task_instance_id 解析 executor/模板/三权。 func (o *Orchestrator) ResolveDocTask(ctx context.Context, projectID, docTaskInstID int64) (*TaskResolution, error) { - if o == nil || o.instSvc == nil || !o.instSvc.Available() { + if o == nil || o.live == nil || !o.live.Available() { return nil, fmt.Errorf("orchestrator unavailable") } - detail, version, err := o.instSvc.GetLiveInstanceDetail(ctx, projectID) + detail, version, err := o.live.GetLiveInstanceDetail(ctx, projectID) if err != nil { return nil, err } - plan := detail.Instance.ConfigSnapshot + if detail == nil || version == nil { + return nil, fmt.Errorf("orchestrator: empty live snapshot") + } + plan := detail.ConfigSnapshot for _, st := range detail.Stages { for _, dt := range st.DocTasks { if dt.ID != docTaskInstID { @@ -50,7 +132,7 @@ func (o *Orchestrator) ResolveDocTask(ctx context.Context, projectID, docTaskIns } res := &TaskResolution{ ProjectID: projectID, - WorkflowInstID: detail.Instance.ID, + WorkflowInstID: detail.InstanceID, StageKey: st.StageKey, TaskKey: dt.TaskKey, DocTaskInstID: dt.ID, @@ -90,7 +172,7 @@ func (o *Orchestrator) ResolveDocTask(ctx context.Context, projectID, docTaskIns return nil, fmt.Errorf("doc task instance not in live snapshot") } -func lookupTaskDef(def *workflowdef.VersionDefinition, stageKey, taskKey string) *workflowdef.DocTaskDef { +func lookupTaskDef(def *VersionDefinitionView, stageKey, taskKey string) *DocTaskDefView { if def == nil { return nil } diff --git a/internal/workflow/runtime/orchestrator_test.go b/internal/workflow/runtime/orchestrator_test.go index 198a9970..1fbd5ffc 100644 --- a/internal/workflow/runtime/orchestrator_test.go +++ b/internal/workflow/runtime/orchestrator_test.go @@ -1,12 +1,106 @@ -package runtime +package runtime_test import ( "context" "testing" "chat-agent-system/internal/projectinstance" + "chat-agent-system/internal/workflow/runtime" + "chat-agent-system/internal/workflowdef" ) +type liveInstanceAdapter struct { + svc *projectinstance.Service +} + +func (a liveInstanceAdapter) Available() bool { + return a.svc != nil && a.svc.Available() +} + +func (a liveInstanceAdapter) GetLiveInstanceDetail(ctx context.Context, projectID int64) (*runtime.LiveInstanceView, *runtime.VersionView, error) { + detail, version, err := a.svc.GetLiveInstanceDetail(ctx, projectID) + if err != nil { + return nil, nil, err + } + return mapLiveInstance(detail), mapVersionView(version), nil +} + +func mapLiveInstance(detail *projectinstance.InstanceDetail) *runtime.LiveInstanceView { + if detail == nil { + return nil + } + view := &runtime.LiveInstanceView{ + InstanceID: detail.Instance.ID, + ConfigSnapshot: runtime.WorkflowPlanView{ + Stages: mapStagePlans(detail.Instance.ConfigSnapshot.Stages), + }, + Stages: mapStageInstances(detail.Stages), + } + return view +} + +func mapStagePlans(stages []projectinstance.StagePlan) []runtime.StagePlanView { + out := make([]runtime.StagePlanView, 0, len(stages)) + for _, st := range stages { + sp := runtime.StagePlanView{StageKey: st.StageKey} + for _, dt := range st.DocTasks { + sp.DocTasks = append(sp.DocTasks, runtime.DocTaskPlanView{ + TaskKey: dt.TaskKey, + Duties: runtime.DutyView{ + AuthorUserID: dt.Duties.AuthorUserID, + ReviewerUserID: dt.Duties.ReviewerUserID, + ApproverUserID: dt.Duties.ApproverUserID, + }, + }) + } + out = append(out, sp) + } + return out +} + +func mapStageInstances(stages []projectinstance.StageInstance) []runtime.StageInstanceView { + out := make([]runtime.StageInstanceView, 0, len(stages)) + for _, st := range stages { + sv := runtime.StageInstanceView{StageKey: st.StageKey} + for _, dt := range st.DocTasks { + sv.DocTasks = append(sv.DocTasks, runtime.DocTaskInstanceView{ + ID: dt.ID, + TaskKey: dt.TaskKey, + }) + } + out = append(out, sv) + } + return out +} + +func mapVersionView(version *workflowdef.Version) *runtime.VersionView { + if version == nil { + return nil + } + view := &runtime.VersionView{} + for _, st := range version.Definition.Stages { + sv := runtime.StageDefView{Key: st.Key} + for _, dt := range st.DocTasks { + dv := runtime.DocTaskDefView{ + Key: dt.Key, + ExecutorKey: dt.ExecutorKey, + QualityGateKey: dt.QualityGateKey, + TemplateAssetKey: dt.TemplateAssetKey, + TemplateVersion: dt.TemplateVersion, + } + if dt.Template != nil { + dv.Template = &runtime.TemplateRefView{ + AssetKey: dt.Template.AssetKey, + Version: dt.Template.Version, + } + } + sv.DocTasks = append(sv.DocTasks, dv) + } + view.Definition.Stages = append(view.Definition.Stages, sv) + } + return view +} + func TestResolveDocTask_NoneDutiesStayZero(t *testing.T) { ctx := context.Background() store := projectinstance.NewMemoryStore() @@ -51,7 +145,7 @@ func TestResolveDocTask_NoneDutiesStayZero(t *testing.T) { t.Fatal("expected research doc task instance id") } - res, err := NewOrchestrator(svc).ResolveDocTask(ctx, 500, researchID) + res, err := runtime.NewOrchestrator(liveInstanceAdapter{svc: svc}).ResolveDocTask(ctx, 500, researchID) if err != nil { t.Fatalf("ResolveDocTask: %v", err) } diff --git a/internal/workflowdef/catalog.go b/internal/workflowdef/catalog.go index e3117d35..07b43827 100644 --- a/internal/workflowdef/catalog.go +++ b/internal/workflowdef/catalog.go @@ -2,22 +2,50 @@ package workflowdef import ( "context" + "errors" "sort" "sync" - - "chat-agent-system/internal/projectworkflow" ) -// DBCatalog 从已发布工作流版本构建 projectworkflow.Catalog(向后兼容扁平 Definition)。 +// ErrCatalogUpstreamMissing is returned when a catalog entry names an unknown upstream. +var ErrCatalogUpstreamMissing = errors.New("workflowdef: catalog required upstream missing") + +// FlatDefinition is the flattened single-stage catalog entry used by legacy catalog consumers. +// Owned here so workflowdef does not import projectworkflow; composition-root adapters map to consumer types. +type FlatDefinition struct { + Key string + Name string + Description string + StageKey string + TaskKey string + RequiredUpstreamWorkflow string + OutputDocType string + OutputDocTitle string + Enabled bool +} + +// FlatCatalog is the workflowdef-owned catalog port (no projectworkflow import). +type FlatCatalog interface { + Get(workflowKey string) (*FlatDefinition, bool) + List() []*FlatDefinition + Validate() error +} + +// CatalogRefresher is an optional FlatCatalog capability for publish/retire hot-reload. +type CatalogRefresher interface { + Refresh(ctx context.Context) error +} + +// DBCatalog builds a FlatCatalog from published workflow versions. type DBCatalog struct { mu sync.RWMutex store *Store - items map[string]*projectworkflow.Definition + items map[string]*FlatDefinition order []string } func NewDBCatalog(ctx context.Context, store *Store) (*DBCatalog, error) { - c := &DBCatalog{store: store, items: make(map[string]*projectworkflow.Definition)} + c := &DBCatalog{store: store, items: make(map[string]*FlatDefinition)} if err := c.reload(ctx); err != nil { return nil, err } @@ -29,7 +57,7 @@ func (c *DBCatalog) reload(ctx context.Context) error { if err != nil { return err } - items := make(map[string]*projectworkflow.Definition) + items := make(map[string]*FlatDefinition) order := make([]string, 0, len(versions)) for _, v := range versions { def := flattenVersion(v) @@ -47,7 +75,7 @@ func (c *DBCatalog) reload(ctx context.Context) error { return nil } -func flattenVersion(v Version) *projectworkflow.Definition { +func flattenVersion(v Version) *FlatDefinition { if len(v.Definition.Stages) == 0 { return nil } @@ -69,7 +97,7 @@ func flattenVersion(v Version) *projectworkflow.Definition { if name == "" { name = v.WorkflowKey } - return &projectworkflow.Definition{ + return &FlatDefinition{ Key: v.WorkflowKey, Name: name, Description: v.Description, @@ -92,7 +120,7 @@ func (c *DBCatalog) Refresh(ctx context.Context) error { return c.reload(ctx) } -func (c *DBCatalog) Get(workflowKey string) (*projectworkflow.Definition, bool) { +func (c *DBCatalog) Get(workflowKey string) (*FlatDefinition, bool) { c.mu.RLock() defer c.mu.RUnlock() d, ok := c.items[workflowKey] @@ -103,10 +131,10 @@ func (c *DBCatalog) Get(workflowKey string) (*projectworkflow.Definition, bool) return &cpy, true } -func (c *DBCatalog) List() []*projectworkflow.Definition { +func (c *DBCatalog) List() []*FlatDefinition { c.mu.RLock() defer c.mu.RUnlock() - out := make([]*projectworkflow.Definition, 0, len(c.order)) + out := make([]*FlatDefinition, 0, len(c.order)) for _, key := range c.order { if d, ok := c.items[key]; ok { cpy := *d @@ -122,15 +150,15 @@ func (c *DBCatalog) Validate() error { for _, def := range c.items { if def.RequiredUpstreamWorkflow != "" { if _, ok := c.items[def.RequiredUpstreamWorkflow]; !ok { - return projectworkflow.ErrWorkflowNotFound + return ErrCatalogUpstreamMissing } } } return nil } -// NewCatalog 优先 DB 已发布 catalog,否则回退静态 catalog。 -func NewCatalog(ctx context.Context, store *Store, fallback projectworkflow.Catalog) projectworkflow.Catalog { +// NewCatalog prefers DB-published catalog entries; otherwise returns fallback. +func NewCatalog(ctx context.Context, store *Store, fallback FlatCatalog) FlatCatalog { if store == nil || !store.enabled() { return fallback } diff --git a/internal/workflowdef/catalog_policy_test.go b/internal/workflowdef/catalog_policy_test.go index 7fc20641..1123a89f 100644 --- a/internal/workflowdef/catalog_policy_test.go +++ b/internal/workflowdef/catalog_policy_test.go @@ -56,7 +56,7 @@ func TestValidateDefinitionV2_OfficialDocCounts(t *testing.T) { t.Fatal(err) } normalizeManifestDefinition(&def) - res := ValidateDefinitionV2(def) + res := ValidateDefinitionV2(def, nil, nil) if !res.Valid { t.Fatalf("main v2 invalid: %+v", res.Issues) } @@ -70,7 +70,7 @@ func TestValidateDefinitionV2_SoftwareManifests(t *testing.T) { t.Fatalf("%s: %v", file, err) } normalizeManifestDefinition(&def) - res := ValidateDefinitionV2(def) + res := ValidateDefinitionV2(def, nil, nil) if !res.Valid { t.Fatalf("%s invalid: %+v", file, res.Issues) } @@ -97,7 +97,7 @@ func TestValidateDefinitionV2_RejectsUpstream(t *testing.T) { DocTasks: []DocTaskDef{{Key: "test_manager", OutputDocType: "test_manager", ExecutorKey: "squad.test_manager.v1"}}, }}, } - res := ValidateDefinitionV2(def) + res := ValidateDefinitionV2(def, nil, nil) if res.Valid { t.Fatal("expected upstream rejection") } diff --git a/internal/workflowdef/rdev_completeness_test.go b/internal/workflowdef/rdev_completeness_test.go index b93b4b64..ad2eb333 100644 --- a/internal/workflowdef/rdev_completeness_test.go +++ b/internal/workflowdef/rdev_completeness_test.go @@ -14,7 +14,7 @@ func TestRDevManifestCompleteness(t *testing.T) { t.Fatal(err) } normalizeManifestDefinition(&def) - res := ValidateDefinitionV2(def) + res := ValidateDefinitionV2(def, nil, nil) if !res.Valid { t.Fatalf("rdev v2 invalid: %+v", res.Issues) } diff --git a/internal/workflowdef/registry_keys.go b/internal/workflowdef/registry_keys.go new file mode 100644 index 00000000..07e27a2d --- /dev/null +++ b/internal/workflowdef/registry_keys.go @@ -0,0 +1,51 @@ +package workflowdef + +import "fmt" + +// Local copies of workflow registry keys for publish-time validation. +// Kept here so workflowdef does not import internal/workflow (unidirectional). +// Runtime resolution remains owned by internal/workflow; keep these sets aligned. +var executorKeys = map[string]struct{}{ + "bizdev.stage.v1": {}, + "squad.system_reqs.v1": {}, + "squad.test_manager.v1": {}, + "squad.rdev_doc.v1": {}, + "bizdev.intake.v1": {}, + "bizdev.research.v1": {}, + "bizdev.feasibility.v1": {}, + "bizdev.feasibility_dev.v1": {}, + "bizdev.user_requirements.v1": {}, + "bizdev.prelim_design.v1": {}, + "bizdev.cdo_quote.v1": {}, + "software.doc.v1": {}, +} + +var qualityGateKeys = map[string]struct{}{ + "quality.system_reqs.v2": {}, + "quality.test_manager.v2": {}, + "quality.generic.v1": {}, + "quality.rd_user_needs.v1": {}, + "quality.rd_system_reqs.v1": {}, + "quality.rd_review_minutes.v1": {}, + "quality.rd_upload_index.v1": {}, +} + +func defaultValidateExecutorKey(key string) error { + if key == "" { + return fmt.Errorf("executor_key required") + } + if _, ok := executorKeys[key]; !ok { + return fmt.Errorf("unknown executor_key %q", key) + } + return nil +} + +func defaultValidateQualityGateKey(key string) error { + if key == "" { + return nil + } + if _, ok := qualityGateKeys[key]; !ok { + return fmt.Errorf("unknown quality_gate_key %q", key) + } + return nil +} diff --git a/internal/workflowdef/service.go b/internal/workflowdef/service.go index 98971a2b..06d0c80f 100644 --- a/internal/workflowdef/service.go +++ b/internal/workflowdef/service.go @@ -14,8 +14,10 @@ import ( // Service 工作流定义与模板领域服务。 type Service struct { - store *Store - cas *storage.CAS + store *Store + cas *storage.CAS + validateExecutor keyValidator + validateGate keyValidator } func NewService(pool *pgxpool.Pool, cas *storage.CAS) *Service { @@ -25,6 +27,16 @@ func NewService(pool *pgxpool.Pool, cas *storage.CAS) *Service { return &Service{store: NewStore(pool), cas: cas} } +// SetRegistryValidators injects executor/quality-gate key checks from the composition root +// so this package never imports the concrete workflow registry. +func (s *Service) SetRegistryValidators(execValidate, gateValidate keyValidator) { + if s == nil { + return + } + s.validateExecutor = execValidate + s.validateGate = gateValidate +} + func (s *Service) Available() bool { return s != nil && s.store != nil && s.store.enabled() } @@ -42,7 +54,7 @@ func (s *Service) GetDefinition(ctx context.Context, id int64) (*Definition, err } func (s *Service) CreateDefinition(ctx context.Context, in CreateDefinitionInput, createdBy int64) (*Definition, *Version, error) { - res := ValidateDefinition(in.Definition, s.templateReady(ctx), s.upstreamExists(ctx)) + res := ValidateDefinition(in.Definition, s.templateReady(ctx), s.upstreamExists(ctx), s.validateExecutor, s.validateGate) if !res.Valid { return nil, nil, ErrValidationFailed } @@ -50,7 +62,7 @@ func (s *Service) CreateDefinition(ctx context.Context, in CreateDefinitionInput } func (s *Service) UpdateDraft(ctx context.Context, workflowID int64, version int, def VersionDefinition) (*Version, error) { - res := ValidateDefinition(def, s.templateReady(ctx), s.upstreamExists(ctx)) + res := ValidateDefinition(def, s.templateReady(ctx), s.upstreamExists(ctx), s.validateExecutor, s.validateGate) if !res.Valid { return nil, ErrValidationFailed } @@ -62,7 +74,7 @@ func (s *Service) ValidateVersion(ctx context.Context, workflowID int64, version if err != nil { return ValidationResult{}, err } - return ValidateDefinition(v.Definition, s.templateReady(ctx), s.upstreamExists(ctx)), nil + return ValidateDefinition(v.Definition, s.templateReady(ctx), s.upstreamExists(ctx), s.validateExecutor, s.validateGate), nil } func (s *Service) PublishVersion(ctx context.Context, workflowID int64, version int, publishedBy int64) (*Version, error) { diff --git a/internal/workflowdef/store.go b/internal/workflowdef/store.go index a8d28cd8..ba649235 100644 --- a/internal/workflowdef/store.go +++ b/internal/workflowdef/store.go @@ -366,7 +366,7 @@ func (s *Store) UpsertPublishedSeed(ctx context.Context, in CreateDefinitionInpu if !s.enabled() { return ErrNotAvailable } - validation := ValidateVersionBySchema(in.Definition) + validation := ValidateVersionBySchema(in.Definition, nil, nil) if !validation.Valid { return fmt.Errorf("seed validation failed: %d issues", len(validation.Issues)) } diff --git a/internal/workflowdef/validator.go b/internal/workflowdef/validator.go index 46659c97..90386989 100644 --- a/internal/workflowdef/validator.go +++ b/internal/workflowdef/validator.go @@ -10,9 +10,10 @@ type templateReadyChecker func(assetKey string, version int) bool type upstreamExistsChecker func(workflowKey string) bool // ValidateDefinition 对工作流版本定义做确定性校验。 -func ValidateDefinition(def VersionDefinition, tplReady templateReadyChecker, upstreamExists upstreamExistsChecker) ValidationResult { +// execValidate / gateValidate are optional registry ports (nil skips). +func ValidateDefinition(def VersionDefinition, tplReady templateReadyChecker, upstreamExists upstreamExistsChecker, execValidate, gateValidate keyValidator) ValidationResult { if def.SchemaVersion >= 2 { - return ValidateDefinitionV2(def) + return ValidateDefinitionV2(def, execValidate, gateValidate) } issues := make([]ValidationIssue, 0) if def.SchemaVersion <= 0 { diff --git a/internal/workflowdef/validator_test.go b/internal/workflowdef/validator_test.go index 0e9425c1..ddc0fd06 100644 --- a/internal/workflowdef/validator_test.go +++ b/internal/workflowdef/validator_test.go @@ -7,7 +7,7 @@ import ( ) func TestValidateDefinition_RejectsEmptyStages(t *testing.T) { - res := workflowdef.ValidateDefinition(workflowdef.VersionDefinition{}, nil, nil) + res := workflowdef.ValidateDefinition(workflowdef.VersionDefinition{}, nil, nil, nil, nil) if res.Valid { t.Fatal("expected invalid for empty stages") } @@ -24,7 +24,7 @@ func TestValidateDefinition_AcceptsMinimalWorkflow(t *testing.T) { }}, }}, } - res := workflowdef.ValidateDefinition(def, func(string, int) bool { return true }, nil) + res := workflowdef.ValidateDefinition(def, func(string, int) bool { return true }, nil, nil, nil) if !res.Valid { t.Fatalf("expected valid, got issues: %+v", res.Issues) } @@ -40,7 +40,7 @@ func TestValidateDefinition_RejectsMissingTemplate(t *testing.T) { }}, }}, } - res := workflowdef.ValidateDefinition(def, func(string, int) bool { return false }, nil) + res := workflowdef.ValidateDefinition(def, func(string, int) bool { return false }, nil, nil, nil) if res.Valid { t.Fatal("expected invalid when template not ready") } diff --git a/internal/workflowdef/validator_v2.go b/internal/workflowdef/validator_v2.go index c65cf27b..12cdd5b6 100644 --- a/internal/workflowdef/validator_v2.go +++ b/internal/workflowdef/validator_v2.go @@ -2,8 +2,6 @@ package workflowdef import ( "fmt" - - "chat-agent-system/internal/workflow" ) var officialDocCounts = map[string]int{ @@ -13,8 +11,20 @@ var officialDocCounts = map[string]int{ OfficialKeyRDev: 60, } +// keyValidator checks a registry key (executor / quality gate). Injected so workflowdef +// does not import the concrete workflow registry package. +type keyValidator func(key string) error + // ValidateDefinitionV2 schema v2 发布校验。 -func ValidateDefinitionV2(def VersionDefinition) ValidationResult { +// execValidate / gateValidate are optional ports; nil uses the local registry snapshot +// (so workflowdef never imports internal/workflow). +func ValidateDefinitionV2(def VersionDefinition, execValidate, gateValidate keyValidator) ValidationResult { + if execValidate == nil { + execValidate = defaultValidateExecutorKey + } + if gateValidate == nil { + gateValidate = defaultValidateQualityGateKey + } issues := make([]ValidationIssue, 0) if def.SchemaVersion < 2 { issues = append(issues, ValidationIssue{Code: "SCHEMA_VERSION", Message: "schema_version must be >= 2"}) @@ -53,12 +63,12 @@ func ValidateDefinitionV2(def VersionDefinition) ValidationResult { } seenTask[dt.Key] = true if dt.ExecutorKey != "" { - if err := workflow.ValidateExecutorKey(dt.ExecutorKey); err != nil { + if err := execValidate(dt.ExecutorKey); err != nil { issues = append(issues, ValidationIssue{Code: "UNKNOWN_EXECUTOR", Field: dt.Key, Message: err.Error()}) } } if dt.QualityGateKey != "" { - if err := workflow.ValidateQualityGateKey(dt.QualityGateKey); err != nil { + if err := gateValidate(dt.QualityGateKey); err != nil { issues = append(issues, ValidationIssue{Code: "UNKNOWN_QUALITY_GATE", Field: dt.Key, Message: err.Error()}) } } @@ -83,9 +93,9 @@ func ValidateDefinitionV2(def VersionDefinition) ValidationResult { } // ValidateVersionBySchema 按 schema_version 路由校验器。 -func ValidateVersionBySchema(def VersionDefinition) ValidationResult { +func ValidateVersionBySchema(def VersionDefinition, execValidate, gateValidate keyValidator) ValidationResult { if def.SchemaVersion >= 2 { - return ValidateDefinitionV2(def) + return ValidateDefinitionV2(def, execValidate, gateValidate) } return validateDefinitionV1(def) }