Agent-System/internal/bizdev/service.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),
		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)
			},
		}),
		ctxMgr:          ctxMgr,
		cas:             casStore,
		materialIndex:   matIndex,
		docGov:          docGov,
		docRev:          docRev,
		freeDoc:         freeDoc,
		projectWorkflow: pwSvc,
		pool:            pool,
		schemaVersion:   schemaVersion,
	}
	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)
}