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/intentflow"
	"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/ctxmgr"
	"chat-agent-system/internal/docgov"
	"chat-agent-system/internal/docrevision"
	"chat-agent-system/internal/events"
	"chat-agent-system/internal/freedoc"
	"chat-agent-system/internal/intent"
	"chat-agent-system/internal/kb"
	"chat-agent-system/internal/notification"
	"chat-agent-system/internal/observability"
	"chat-agent-system/internal/projectinstance"
	"chat-agent-system/internal/projectworkflow"
	"chat-agent-system/internal/storage"
	"chat-agent-system/internal/taskrun"
	"chat-agent-system/internal/workflowdef"
	"chat-agent-system/internal/workspace"

	"github.com/jackc/pgx/v5/pgxpool"
)

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
	IntentPlans *intentflow.Plans
	Pool        *pgxpool.Pool
	Gov         *docgov.GovernanceEngine
	Rev         *docrevision.Service
	Ctx         *ctxmgr.ContextManager
	CAS         *storage.CAS
	FreeDoc     *freedoc.Service
	KB          *kb.Client

	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(nil, dialogueBackend, cfg.DataDir)
	parts := bizdev.Snapshot(svc)
	pool := parts.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 := parts.PW
	if pwSvc == nil {
		pwSvc = projectworkflow.NewService(projectworkflow.NewStore(nil), nil)
	}

	authzSvc := authz.NewService(pool)
	wfDefSvc := workflowdef.NewService(pool, parts.CAS)
	instStore := projectinstance.NewStore(pool)
	var instSvc *projectinstance.Service
	if wfDefSvc.Available() {
		docChecker := projectinstance.NewDocGovChecker(parts.Gov)
		instSvc = projectinstance.NewService(instStore, wfDefSvc.Store(), pwSvc.Store(), docChecker)
		if pool != nil {
			instSvc.SetActiveMemberLookup(projectinstance.NewPGMemberLookup(pool))
		}
	}
	taskMgr := taskrun.NewManager(pwSvc, parts.Ctx)
	if pool != nil {
		taskMgr.SetStore(taskrun.NewPGStore(pool))
	}
	wsCap := workspace.NewCapability(cfg.DataDir, workspace.NewLeaseTable())
	cmds := &taskrun.Commands{
		Ctx:             parts.Ctx,
		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)
	coordEngine.SetContextServices(parts.Ctx, parts.KB)
	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,
		Pool:       pool,
		Gov:        parts.Gov,
		Rev:        parts.Rev,
		Ctx:        parts.Ctx,
		CAS:        parts.CAS,
		FreeDoc:    parts.FreeDoc,
		KB:         parts.KB,
	}
	rt.DocNotify = &docnotify.Subscriber{Notify: rt.Notify, Docs: parts.Gov}
	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: parts.Gov, Rev: parts.Rev, Auth: docflow.DutyAuthorizer(),
		Environment: envName,
	}
	if pool != nil {
		rt.Docflow.UoW = uow.NewPG(pool)
		rt.Docflow.Events = events.PGSink{}
		rt.IntentPlans = intentflow.NewPlans(rt.Docflow.UoW, intentflow.NewRepository(), intentflow.Options{})
	}
	svc.SetAgentMaterializer(rt.Docflow)
	rt.bindDocumentNotify()
	rt.bindWorkflowApprovalHook()
	return rt, nil
}

func (r *Runtime) bindDocumentNotify() {
	if r == nil || r.notifyBound {
		return
	}
	// Durable notify is live; do not assign OnOutbox in production.
	r.notifyBound = true
}

func (r *Runtime) bindWorkflowApprovalHook() {
	if r == nil || r.hookBound {
		return
	}
	// Durable workflow_advance is live; do not register AddOnApprovedHook in production.
	r.hookBound = true
}

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