agent-bot
2c75dcaf69
KTD1 fifth step (behavior preserved):
- squad.AssetProvider{DocTemplate,RenderDoc,AgentInstruction,
AgentInstructionFile} injected via Options.Assets into Service,
WorkspaceEngine and AdaptivePlanner
- bizdev templateAssetProvider maps port to internal/template (assembly side)
- nil-provider paths degrade to the same fallbacks the old error branches
used (placeholder template, role header) — fail-visible, no guessing
- squad -> template import eliminated; arch edge moved bizdev->template
- verified: build/vet/arch green, squad+bizdev short tests pass
342 lines
12 KiB
Go
342 lines
12 KiB
Go
package bizdev
|
||
|
||
import (
|
||
"context"
|
||
"errors"
|
||
"fmt"
|
||
"log/slog"
|
||
"os"
|
||
"path/filepath"
|
||
"strings"
|
||
"time"
|
||
|
||
"chat-agent-system/internal/agent"
|
||
"chat-agent-system/internal/app/uow"
|
||
"chat-agent-system/internal/authz"
|
||
"chat-agent-system/internal/config"
|
||
"chat-agent-system/internal/ctxmgr"
|
||
"chat-agent-system/internal/db"
|
||
"chat-agent-system/internal/docgov"
|
||
"chat-agent-system/internal/docrevision"
|
||
"chat-agent-system/internal/freedoc"
|
||
"chat-agent-system/internal/kb"
|
||
"chat-agent-system/internal/materialindex"
|
||
"chat-agent-system/internal/notification"
|
||
"chat-agent-system/internal/projectworkflow"
|
||
"chat-agent-system/internal/squad"
|
||
"chat-agent-system/internal/storage"
|
||
"chat-agent-system/internal/workflowdef"
|
||
"chat-agent-system/internal/workspace"
|
||
|
||
"github.com/jackc/pgx/v5/pgxpool"
|
||
)
|
||
|
||
type Service struct {
|
||
cfg *config.Config
|
||
kb *kb.Client
|
||
squad *squad.Service
|
||
ctxMgr *ctxmgr.ContextManager
|
||
cas *storage.CAS
|
||
materialIndex *materialindex.Pipeline
|
||
docGov *docgov.GovernanceEngine
|
||
docRev *docrevision.Service
|
||
freeDoc *freedoc.Service
|
||
projectWorkflow *projectworkflow.Service
|
||
pool *pgxpool.Pool
|
||
schemaVersion int64
|
||
stageGate StageAccessGate
|
||
materializer AgentOutputMaterializer
|
||
parts Parts
|
||
// stageTaskHook 仅测试注入：覆盖 ExecuteStage 子任务执行，验证链中止边界。
|
||
stageTaskHook func(context.Context, TaskRunRequest) (*TaskRunResult, error)
|
||
// createDocumentHook 仅测试注入：覆盖账本创建失败路径。
|
||
createDocumentHook func(ctx context.Context, docID string, projectID int64, taskKey, title, stage, workDir string, authorID int64, authorName string, reviewerID int64, reviewerName string, approverID int64, approverName string) (*docgov.DocumentRecord, error)
|
||
// completedMainHook 仅测试注入：覆盖「main 实例已 completed」查询。
|
||
completedMainHook func(ctx context.Context, projectID int64) (bool, error)
|
||
// instanceDuty 可选：活跃工作流实例对 taskKey 的治理职责（由 app 装配，避免导入 projectinstance）。
|
||
instanceDuty InstanceDutySource
|
||
// notify 可选：提案拦截时写入 document digest（composition 注入）。
|
||
notify DigestNotifier
|
||
}
|
||
|
||
// DigestNotifier is the subset of notification.Service used on proposal_required.
|
||
type DigestNotifier interface {
|
||
UpsertDocumentDigest(ctx context.Context, p notification.UpsertDigestParams) error
|
||
}
|
||
|
||
// InstanceDutySource 返回项目当前活跃实例对 taskKey 的治理结论。
|
||
// governed=false 表示未受实例治理，调用方应回退项目默认职责逻辑。
|
||
type InstanceDutySource interface {
|
||
ResolveGovernedDuty(ctx context.Context, projectID int64, taskKey string) (author, reviewer, approver int64, dutyPolicy string, governed bool, err error)
|
||
}
|
||
|
||
// SetInstanceDutySource 注入实例职责解析源（组合根可选装配）。
|
||
func (s *Service) SetInstanceDutySource(src InstanceDutySource) {
|
||
if s == nil {
|
||
return
|
||
}
|
||
s.instanceDuty = src
|
||
}
|
||
|
||
// SetDigestNotifier wires optional claim-aware digest notifications.
|
||
func (s *Service) SetDigestNotifier(n DigestNotifier) {
|
||
if s == nil {
|
||
return
|
||
}
|
||
s.notify = n
|
||
}
|
||
|
||
// NewService 构造业务服务；生产/test 环境数据库不可用且未显式 ALLOW_IN_MEMORY 时返回 error。
|
||
func NewService(cfg *config.Config) (*Service, error) {
|
||
if cfg == nil {
|
||
return nil, errors.New("bizdev: config is required (data roots come from config, not package defaults)")
|
||
}
|
||
var pool *pgxpool.Pool
|
||
var schemaVersion int64
|
||
dbConfigured := cfg.ContextDbHost != ""
|
||
if dbConfigured {
|
||
poolCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
|
||
defer cancel()
|
||
// T20-B4：统一池 profile（default 30s statement_timeout；迁移链由 goose
|
||
// 单独会话跑，不受池默认影响）。
|
||
p, err := db.OpenAppPool(poolCtx, cfg.ContextDSN(), db.DefaultProfile())
|
||
if err != nil {
|
||
if !cfg.AllowInMemory {
|
||
return nil, fmt.Errorf("database connect failed: %w", err)
|
||
}
|
||
} else if pingErr := p.Ping(poolCtx); pingErr != nil {
|
||
p.Close()
|
||
if !cfg.AllowInMemory {
|
||
return nil, fmt.Errorf("database ping failed: %w", pingErr)
|
||
}
|
||
} else {
|
||
pool = p
|
||
migrationsDir := db.ResolveMigrationsDir()
|
||
if migErr := db.RunMigrations(poolCtx, pool, migrationsDir); migErr != nil {
|
||
pool.Close()
|
||
if !cfg.AllowInMemory {
|
||
return nil, fmt.Errorf("database migration failed: %w", migErr)
|
||
}
|
||
} else {
|
||
authzStore := authz.NewAuthorizer(pool)
|
||
authzStore.SetDutySlotWriter(docgov.DutySlotWriter{})
|
||
_ = authzStore.BackfillCreatorMembers(poolCtx)
|
||
if ver, verErr := db.CurrentVersion(poolCtx, pool); verErr == nil {
|
||
schemaVersion = ver
|
||
}
|
||
}
|
||
}
|
||
}
|
||
var casEmbedder storage.Embedder
|
||
if cfg.EmbeddingEndpointURL != "" {
|
||
casEmbedder = kb.NewEmbedder(cfg.EmbeddingEndpointURL, cfg.EmbeddingAPIKey, cfg.EmbeddingModel)
|
||
}
|
||
casStore, _ := storage.NewCAS(filepath.Join(cfg.DataDir, "storage"), pool, casEmbedder)
|
||
ctxMgr := ctxmgr.NewContextManager(pool, casStore, ctxmgr.DefaultTimeDecayConfig())
|
||
if cfg.EmbeddingEndpointURL != "" {
|
||
ctxMgr.SetEmbedder(kb.NewEmbedder(cfg.EmbeddingEndpointURL, cfg.EmbeddingAPIKey, cfg.EmbeddingModel))
|
||
}
|
||
matIndex := materialindex.NewPipeline(cfg, pool)
|
||
docGov := docgov.NewGovernanceEngine(pool)
|
||
docRev := docrevision.NewService(pool)
|
||
docRev.DataDir = cfg.DataDir
|
||
docRev.SetRootPointerProjector(docrevision.CompositeRootPointerProjector{
|
||
Controlled: func(ctx context.Context, scope uow.TxScope, root *docrevision.DocRootPointers) error {
|
||
return (docgov.Repository{}).ProjectHeadFull(ctx, scope, root.DocID,
|
||
root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer, root.ContentSHA256)
|
||
},
|
||
Freeform: func(ctx context.Context, scope uow.TxScope, root *docrevision.DocRootPointers) error {
|
||
return (freedoc.Repository{}).ProjectHead(ctx, scope, root.DocID,
|
||
root.LedgerVersionID, root.HeadSnapshotID, root.HeadRevisionNo, root.RevisionSchemaVer)
|
||
},
|
||
})
|
||
docRev.SetDeliverableStore(workspace.DirDeliverable{})
|
||
freeDoc := freedoc.NewService(pool, ctxMgr)
|
||
freeDoc.SetDocRevision(docRev)
|
||
pwStore := projectworkflow.NewStore(pool)
|
||
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)
|
||
}
|
||
}
|
||
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
|
||
}
|
||
lineages, err := pwSvc.Store().GetLineagesForUpstream(ctx, docID)
|
||
if err != nil || len(lineages) == 0 {
|
||
return nil
|
||
}
|
||
down := make([]struct {
|
||
DocID string
|
||
Reason string
|
||
}, 0, len(lineages))
|
||
for _, ln := range lineages {
|
||
if ln == nil {
|
||
continue
|
||
}
|
||
reason := ln.StaleReason
|
||
if reason == "" {
|
||
reason = "上游文档内容变更，下游需复核"
|
||
}
|
||
down = append(down, struct {
|
||
DocID string
|
||
Reason string
|
||
}{DocID: ln.DownstreamDocID, Reason: reason})
|
||
}
|
||
return docrevision.ImpactFromLineage(down)
|
||
}
|
||
|
||
svc := &Service{
|
||
cfg: cfg,
|
||
kb: kb.NewClient(cfg),
|
||
ctxMgr: ctxMgr,
|
||
cas: casStore,
|
||
materialIndex: matIndex,
|
||
docGov: docGov,
|
||
docRev: docRev,
|
||
freeDoc: freeDoc,
|
||
projectWorkflow: pwSvc,
|
||
pool: pool,
|
||
schemaVersion: schemaVersion,
|
||
}
|
||
svc.squad = squad.NewService(squad.Options{
|
||
BackendName: cfg.AgentBackend,
|
||
BackendTimeout: cfg.AgentTimeout,
|
||
DataDir: cfg.DataDir,
|
||
NewBackend: func(name string) agent.Backend {
|
||
return agent.NewBackend(name, cfg)
|
||
},
|
||
TaskPreparer: svc.squadTaskPreparer,
|
||
Assets: templateAssetProvider{},
|
||
})
|
||
svc.parts = Parts{
|
||
Pool: pool,
|
||
Gov: docGov,
|
||
Rev: docRev,
|
||
Ctx: ctxMgr,
|
||
CAS: casStore,
|
||
MaterialIndex: matIndex,
|
||
PW: pwSvc,
|
||
FreeDoc: freeDoc,
|
||
KB: svc.kb,
|
||
}
|
||
|
||
// 注册终审批准回调：只读冻结 ledger snapshot，禁止扫描 workspace / output.md
|
||
docGov.SetOnApprovedHook(func(ctx context.Context, doc *docgov.DocumentRecord) {
|
||
if ctxMgr == nil || doc == nil {
|
||
return
|
||
}
|
||
indexed := false
|
||
if doc.ReviewSnapshotID > 0 {
|
||
if md, err := docGov.LedgerMarkdown(ctx, doc.ReviewSnapshotID); err == nil && strings.TrimSpace(md) != "" {
|
||
_, _ = ctxMgr.IndexApprovedDocument(ctx, doc, md)
|
||
indexed = true
|
||
}
|
||
}
|
||
if !indexed {
|
||
var contentPath string
|
||
if doc.WorkDir != "" {
|
||
candidates := []string{
|
||
doc.WorkDir,
|
||
filepath.Join(doc.WorkDir, workspace.DeliverableKey),
|
||
filepath.Join(cfg.DataDir, doc.WorkDir, workspace.DeliverableKey),
|
||
filepath.Join(cfg.DataDir, workspace.WorkspacesDirName, doc.WorkDir, workspace.DeliverableKey),
|
||
}
|
||
for _, p := range candidates {
|
||
if b, err := os.ReadFile(p); err == nil && len(b) > 0 {
|
||
contentPath = p
|
||
break
|
||
}
|
||
}
|
||
}
|
||
if contentPath != "" {
|
||
if b, err := os.ReadFile(contentPath); err == nil && len(b) > 0 {
|
||
_, _ = ctxMgr.IndexApprovedDocument(ctx, doc, string(b))
|
||
}
|
||
} else {
|
||
pattern := filepath.Join(cfg.DataDir, workspace.WorkspacesDirName, "**", doc.TaskKey, workspace.DeliverableKey)
|
||
matches, _ := filepath.Glob(pattern)
|
||
if len(matches) == 0 {
|
||
pattern2 := filepath.Join(cfg.DataDir, "**", doc.TaskKey, workspace.DeliverableKey)
|
||
matches, _ = filepath.Glob(pattern2)
|
||
}
|
||
if len(matches) > 0 {
|
||
if b, err := os.ReadFile(matches[len(matches)-1]); err == nil && len(b) > 0 {
|
||
_, _ = ctxMgr.IndexApprovedDocument(ctx, doc, string(b))
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
if pwSvc != nil && workflowdef.SameTaskKey(doc.TaskKey, "system_reqs") {
|
||
_, _ = pwSvc.CheckAndPropagateStale(ctx, doc.DocID, doc.ContentSHA256, fmt.Sprintf("上游系统需求规格书 V%d 已正式批准，已标记关联下游验证方案为待复核(stale)", doc.Version))
|
||
}
|
||
})
|
||
|
||
return svc, nil
|
||
}
|
||
|
||
// SchemaVersion 返回启动时缓存的数据库 schema 版本（goose）。
|
||
func (s *Service) SchemaVersion() int64 {
|
||
if s == nil {
|
||
return 0
|
||
}
|
||
return s.schemaVersion
|
||
}
|
||
|
||
// LiveSchemaVersion 从数据库实时读取当前 schema 版本；失败时回退缓存值。
|
||
func (s *Service) LiveSchemaVersion(ctx context.Context) int64 {
|
||
if s == nil {
|
||
return 0
|
||
}
|
||
if s.pool == nil {
|
||
return s.schemaVersion
|
||
}
|
||
ver, err := db.CurrentVersion(ctx, s.pool)
|
||
if err != nil {
|
||
return s.schemaVersion
|
||
}
|
||
return ver
|
||
}
|
||
|
||
// DatabaseReady 数据库是否已连接且迁移完成。
|
||
func (s *Service) DatabaseReady() bool {
|
||
return s != nil && s.pool != nil
|
||
}
|
||
|
||
// AgentOutputMaterializer is the production materialize path. Runtime injects
|
||
// *docflow.Commands so candidate-ready/commit envelopes share the unique UoW.
|
||
type AgentOutputMaterializer interface {
|
||
MaterializeAgentOutput(ctx context.Context, req docrevision.AgentMaterializeRequest) (*docrevision.AgentMaterializeResult, error)
|
||
}
|
||
|
||
func (s *Service) SetAgentMaterializer(m AgentOutputMaterializer) {
|
||
if s != nil {
|
||
s.materializer = m
|
||
}
|
||
}
|
||
|
||
func (s *Service) ListTasks() []TaskMeta {
|
||
return s.squad.ListTasks()
|
||
}
|
||
|
||
// resolveDocStage 解析写入文档账本的 Stage 字段。
|
||
// 优先请求透传的冻结实例阶段键；否则按 main 三阶段任务映射；再回落 V0.5_feasibility。
|
||
|
||
func (s *Service) ListStages() []StageMeta {
|
||
return SupportedStages
|
||
}
|
||
|
||
func (s *Service) InspectWorkspace(ctx context.Context, workDir string) (*squad.WorkspaceState, error) {
|
||
return s.squad.InspectWorkspace(withSquadProgress(ctx), workDir)
|
||
}
|
||
|
||
// ResumeOrModifyWorkspace 动态调整任务并继续执行
|
||
func (s *Service) ResumeOrModifyWorkspace(ctx context.Context, req squad.ModifyWorkflowRequest) (*squad.SquadRunResult, error) {
|
||
return s.squad.ResumeOrModifyWorkspace(withSquadProgress(ctx), req)
|
||
}
|