weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

merge: feature/workflow-def-unidirect-2026-09-22

Browse Source
This commit is contained in:
WeiCHEN
2026-09-22 21:53:52 +08:00
parent 4662e39541 993b3dcbae
commit 6984f96c19
18 changed files with 554 additions and 65 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.doc/ops/short-term/2026-09-22-workflow-def-unidirect.md
+62
View File
@@ -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
cmd/server/main.go
+5
View File
@@ -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)
internal/bizdev/catalog_adapter.go
+131
View File
@@ -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()
}
internal/bizdev/service.go
+2 -3
View File
@@ -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
internal/projectinstance/ports.go
+16
View File
@@ -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)
}
internal/projectinstance/service.go
-7
View File
@@ -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
internal/server/admin_workflow_handlers.go
+9 -4
View File
@@ -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)
}
internal/workflow/runtime/orchestrator.go
+95 -13
View File
@@ -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
}
internal/workflow/runtime/orchestrator_test.go
+96 -2
View File
@@ -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)
}
internal/workflowdef/catalog.go
+42 -14
View File
@@ -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
}
internal/workflowdef/catalog_policy_test.go
+3 -3
View File
@@ -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")
}
internal/workflowdef/rdev_completeness_test.go
+1 -1
View File
@@ -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)
}
internal/workflowdef/registry_keys.go
+51
View File
@@ -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
}
internal/workflowdef/service.go
+17 -5
View File
@@ -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) {
internal/workflowdef/store.go
+1 -1
View File
@@ -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))
}
internal/workflowdef/validator.go
+3 -2
View File
@@ -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 {
internal/workflowdef/validator_test.go
+3 -3
View File
@@ -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")
}
internal/workflowdef/validator_v2.go
+17 -7
View File
@@ -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)
}