Agent-System/internal/app/app.go

package app

import (
	"context"
	"fmt"
	"log/slog"
	"sync"
	"time"

	"chat-agent-system/internal/agent"
	"chat-agent-system/internal/app/chatflow"
	"chat-agent-system/internal/app/docflow"
	"chat-agent-system/internal/app/docnotify"
	"chat-agent-system/internal/app/stageflow"
	"chat-agent-system/internal/app/uow"
	"chat-agent-system/internal/artifactvault"
	"chat-agent-system/internal/auth"
	"chat-agent-system/internal/authz"
	"chat-agent-system/internal/bizdev"
	"chat-agent-system/internal/config"
	"chat-agent-system/internal/coordinator"
	"chat-agent-system/internal/docgov"
	"chat-agent-system/internal/events"
	"chat-agent-system/internal/intent"
	"chat-agent-system/internal/notification"
	"chat-agent-system/internal/observability"
	"chat-agent-system/internal/projectinstance"
	"chat-agent-system/internal/projectworkflow"
	"chat-agent-system/internal/taskrun"
	"chat-agent-system/internal/workflowdef"
	"chat-agent-system/internal/workspace"
)

const defaultReminderPollInterval = 15 * time.Minute

// Runtime is the unique composition root. cmd/server, tests and workers
// construct it once; HTTP adapters must not re-wire domain services.
type Runtime struct {
	Cfg        *config.Config
	Biz        *bizdev.Service
	Engine     *coordinator.CoordinatorEngine
	Auth       *auth.Service
	PW         *projectworkflow.Service
	TaskRuns   *taskrun.Manager
	Authorizer *authz.Authorizer
	Authz      *authz.Service
	WFDef      *workflowdef.Service
	Inst       *projectinstance.Service
	Notify     *notification.Service
	Resolver   *workspace.Resolver
	WSCap      *workspace.Capability
	Commands   *taskrun.Commands
	Rec        intent.Recognizer
	Stageflow  *stageflow.UseCase
	DocNotify  *docnotify.Subscriber
	ChatLog    chatflow.ChatLog
	Assembler  chatflow.ContextAssembler
	Docflow    *docflow.Commands

	mu             sync.Mutex
	running        bool
	subscribers    int
	starts         int
	notifyBound    bool
	hookBound      bool
	reminderCancel context.CancelFunc
}

func New(cfg *config.Config) (*Runtime, error) {
	return Assemble(cfg, nil)
}

// Assemble wires domain services. svc may be injected by tests.
func Assemble(cfg *config.Config, svc *bizdev.Service) (*Runtime, error) {
	if cfg == nil {
		cfg, _ = config.Load("")
	}
	if cfg != nil {
		observability.InitLogger(cfg.Env)
	}
	if svc == nil {
		var err error
		svc, err = bizdev.NewService(cfg)
		if err != nil {
			return nil, err
		}
	}

	rec := intent.NewRecognizer()
	dialogueBackend := agent.NewBackend(cfg.AgentBackend, cfg)
	coordEngine := coordinator.NewCoordinatorEngineWithDialogue(svc, dialogueBackend, cfg.DataDir)
	pool := svc.Pool()
	stores, storeErr := auth.NewStores(context.Background(), pool, cfg)
	if storeErr != nil {
		return nil, fmt.Errorf("auth store: %w", storeErr)
	}
	authSvc := auth.NewService(stores.Auth, stores.Sessions, cfg)
	pwSvc := svc.ProjectWorkflow()
	if pwSvc == nil {
		pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil)
	}

	authzSvc := authz.NewService(pool)
	wfDefSvc := workflowdef.NewService(pool, svc.CAS())
	instStore := projectinstance.NewStore(pool)
	var instSvc *projectinstance.Service
	if wfDefSvc.Available() {
		docChecker := projectinstance.NewDocGovChecker(svc.DocGov())
		instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker)
		if pool != nil {
			instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool))
		}
	}
	taskMgr := taskrun.NewManager(pwSvc, svc.ContextManager())
	if pool != nil {
		taskMgr.SetStore(taskrun.NewPGStore(pool))
	}
	wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable())
	cmds := &taskrun.Commands{
		Ctx:             svc.ContextManager(),
		PW:              pwSvc,
		Env:             cfg.Env,
		StorageIdentity: "env:" + cfg.Env,
	}
	if pool != nil {
		cmds.UoW = uow.NewPG(pool)
		cmds.Fav = &artifactvault.Ingest{Env: cfg.Env}
	}
	taskMgr.SetCommands(cmds)

	stages := stageflow.New(svc)
	coordEngine.SetStageTaskRunner(stages)
	if instSvc != nil && instSvc.Available() {
		coordEngine.SetInstanceService(instSvc)
	}

	rt := &Runtime{
		Cfg:        cfg,
		Biz:        svc,
		Engine:     coordEngine,
		Auth:       authSvc,
		PW:         pwSvc,
		TaskRuns:   taskMgr,
		Authorizer: authz.NewAuthorizer(pool),
		Authz:      authzSvc,
		WFDef:      wfDefSvc,
		Inst:       instSvc,
		Notify:     notification.NewService(pool),
		Resolver:   workspace.NewResolver(cfg.DataDir, pool),
		WSCap:      wsCap,
		Commands:   cmds,
		Rec:        rec,
		Stageflow:  stages,
	}
	rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: svc.DocGov()}
	rt.DocNotify.SkipWrite = func(ctx context.Context) bool {
		return rt.WriterLive(ctx, events.ConsumerNotify)
	}
	envName := ""
	if cfg != nil {
		envName = cfg.Env
	}
	rt.Docflow = &docflow.Commands{
		Gov: svc.DocGov(), Rev: svc.DocRevision(), Auth: docflow.DutyAuthorizer(),
		Environment: envName,
	}
	if pool != nil {
		rt.Docflow.UoW = uow.NewPG(pool)
		rt.Docflow.Events = events.PGSink{}
	}
	svc.SetAgentMaterializer(rt.Docflow)
	rt.bindDocumentNotify()
	rt.bindWorkflowApprovalHook()
	return rt, nil
}

func (r *Runtime) bindDocumentNotify() {
	if r == nil || r.notifyBound {
		return
	}
	if r.Biz != nil && r.Biz.DocRevision() != nil {
		if docnotify.Bind(r.Biz.DocRevision(), r.DocNotify) {
			r.subscribers++
		}
	}
	r.notifyBound = true
}

func (r *Runtime) bindWorkflowApprovalHook() {
	if r == nil || r.hookBound || r.Biz == nil || r.Biz.DocGov() == nil || r.Inst == nil || !r.Inst.Available() {
		r.hookBound = true
		return
	}
	r.Biz.DocGov().AddOnApprovedHook(r.onWorkflowDocumentApproved)
	r.hookBound = true
	r.subscribers++
}

// WriterLive reports whether the durable consumer already owns the visible write.
func (r *Runtime) WriterLive(ctx context.Context, consumer string) bool {
	if r == nil || r.Biz == nil {
		return false
	}
	env := ""
	if r.Cfg != nil {
		env = r.Cfg.Env
	}
	return events.WriterLive(ctx, r.Biz.Pool(), env, consumer)
}

func (r *Runtime) onWorkflowDocumentApproved(ctx context.Context, doc *docgov.DocumentRecord) {
	if r.WriterLive(ctx, events.ConsumerWorkflowAdvance) {
		return
	}
	if doc == nil || doc.TaskKey == "" || doc.ProjectID <= 0 || r.Inst == nil {
		return
	}
	result, err := r.Inst.HandleDocumentApproved(ctx, doc.ProjectID, doc.TaskKey)
	if err != nil || result == nil || !result.StageReady {
		return
	}
	if r.Notify == nil || !r.Notify.Available() {
		return
	}
	if r.PW == nil {
		return
	}
	project, err := r.PW.GetProject(ctx, doc.ProjectID, 0)
	if err != nil || project == nil || project.CreatorUserID <= 0 {
		return
	}
	_, _ = r.Notify.EmitStageReadyToAdvance(ctx, project.CreatorUserID, notification.StageReadyPayload{
		ProjectID:          doc.ProjectID,
		WorkflowInstanceID: result.InstanceID,
		StageInstanceID:    result.StageInstanceID,
		StageKey:           result.StageKey,
		StageName:          result.StageName,
		ConfigRevision:     result.ConfigRevision,
		RequiredDocCount:   result.RequiredDocCount,
		ApprovedDocCount:   result.ApprovedDocCount,
	})
}

// Start launches background workers. Repeat calls on a running runtime are no-ops.
func (r *Runtime) Start(_ context.Context) error {
	if r == nil {
		return fmt.Errorf("app: nil runtime")
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.running {
		return nil
	}
	r.bindDocumentNotify()
	r.bindWorkflowApprovalHook()
	ctx, cancel := context.WithCancel(context.Background())
	r.reminderCancel = cancel
	go r.runReminderLoop(ctx)
	r.running = true
	r.starts++
	if r.subscribers == 0 {
		r.subscribers = 1
	}
	return nil
}

// Stop cancels workers. A later Start recreates the reminder loop without
// double-registering document subscribers.
func (r *Runtime) Stop(_ context.Context) error {
	if r == nil {
		return nil
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	if !r.running {
		return nil
	}
	if r.reminderCancel != nil {
		r.reminderCancel()
		r.reminderCancel = nil
	}
	r.running = false
	return nil
}

func (r *Runtime) SubscriberCount() int {
	if r == nil {
		return 0
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.subscribers
}

func (r *Runtime) StartCount() int {
	if r == nil {
		return 0
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.starts
}

func (r *Runtime) Running() bool {
	if r == nil {
		return false
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.running
}

func (r *Runtime) runReminderLoop(ctx context.Context) {
	r.runStageAdvanceReminders(ctx)
	ticker := time.NewTicker(defaultReminderPollInterval)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
			r.runStageAdvanceReminders(ctx)
		}
	}
}

func (r *Runtime) runStageAdvanceReminders(ctx context.Context) {
	if r.Notify == nil || !r.Notify.Available() {
		return
	}
	n, err := r.Notify.ProcessStageAdvanceReminders(ctx)
	if err != nil {
		slog.Warn("stage advance reminder scan failed", "error", err)
		return
	}
	if n > 0 {
		slog.Info("stage advance 24h reminders sent", "count", n)
	}
}