refactor(workflowdef): make definition/runtime deps unidirectional
Keep registry key validation local to workflowdef with composition-root injection, adapt FlatCatalog for projectworkflow, and drive the orchestrator through LiveInstancePort views instead of projectinstance.
This commit is contained in:
WeiCHEN
@@ -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: ""
|
||||
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
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
}
|
||||
@@ -130,15 +130,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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user