internal/app/application.go
Rama main · réplica de lectura del repositorio original.
package app
import (
"context"
"net/http"
"strings"
"sync"
"time"
"github.com/yeomyeonggeori/blueclaw/internal/acpsession"
"github.com/yeomyeonggeori/blueclaw/internal/agentruntime"
"github.com/yeomyeonggeori/blueclaw/internal/backup"
"github.com/yeomyeonggeori/blueclaw/internal/config"
"github.com/yeomyeonggeori/blueclaw/internal/connectors"
apiconnector "github.com/yeomyeonggeori/blueclaw/internal/connectors/api"
"github.com/yeomyeonggeori/blueclaw/internal/filejournal"
"github.com/yeomyeonggeori/blueclaw/internal/harnessdriver"
"github.com/yeomyeonggeori/blueclaw/internal/httpserver"
"github.com/yeomyeonggeori/blueclaw/internal/learning"
"github.com/yeomyeonggeori/blueclaw/internal/memory"
"github.com/yeomyeonggeori/blueclaw/internal/protocolidentity"
runtimelogging "github.com/yeomyeonggeori/blueclaw/internal/runtime"
"github.com/yeomyeonggeori/blueclaw/internal/runtimecontrol"
"github.com/yeomyeonggeori/blueclaw/internal/scheduler"
"github.com/yeomyeonggeori/blueclaw/internal/security"
"github.com/yeomyeonggeori/blueclaw/internal/store/postgres"
"github.com/yeomyeonggeori/blueclaw/internal/task"
)
const databaseInitializationTimeout = 240 * time.Second
const backgroundLoopStopGrace = 10 * time.Second
type Application struct {
httpServer *http.Server
connectorRuntime *connectors.ConnectorRuntime
taskRunService *task.TaskRunService
interruptedTaskResumer interruptedTaskResumer
runtimeLogger *runtimelogging.PersistentLogger
terminalService *security.ShellService
backgroundLoops sync.WaitGroup
database postgres.Database
startupError error
connectorRuntimeCancel context.CancelFunc
interruptedTaskResumeCancel context.CancelFunc
scheduleCancel context.CancelFunc
logRetentionCancel context.CancelFunc
memoryMaintenanceCancel context.CancelFunc
learningCancel context.CancelFunc
taskRetentionCancel context.CancelFunc
staleTaskCancel context.CancelFunc
schedulePoller *scheduler.SchedulePoller
taskRetentionSweeper *scheduler.TaskRetentionSweeper
memoryStores *memory.Stores
heldCircles func() map[string]bool
isOnTheRoster func(personID string) bool
learningCoordinator *learning.Coordinator
schedulePollSecond int
taskRetentionIntervalMinute int
interruptedTaskResumeDelay time.Duration
languageModelConfigured bool
languageModelError error
protocolIdentityChecker protocolidentity.Checker
protocolIdentityExpected protocolidentity.Identity
protocolIdentityStatus *protocolidentity.Result
protocolIdentityCheckOnce sync.Once
protocolIdentityCheckError error
refreshSkillIndex func(context.Context)
acpSessionServer *acpsession.Server
acpSessionCancel context.CancelFunc
trustedClock *runtimecontrol.TrustedClock
clockWatchCancel context.CancelFunc
}
type InboundOptions struct {
ACPSocketPath string
InboundPath string
}
const (
InboundPathConnectors = "connectors"
InboundPathACP = "acp"
)
func (options InboundOptions) admitsConnectorHTTPEvent() bool {
return strings.TrimSpace(options.InboundPath) != InboundPathACP
}
type interruptedTaskResumer interface {
CanResumeInterruptedTaskRun(task.TaskRun) bool
ResumeInterruptedTaskRun(context.Context, task.TaskRun) (connectors.ConnectorRuntimeResult, error)
FailUnresumedInterruptedTaskRun(context.Context, task.TaskRun, string) bool
}
type applicationComponents struct {
runtimeConfiguration config.RuntimeConfiguration
policyPath string
foundation runtimeFoundation
directory identityDirectory
services taskServices
kernel agentKernel
memory memoryComponents
backupCoordinator *backup.Coordinator
trustedClock *runtimecontrol.TrustedClock
taskIntakeController *runtimecontrol.TaskIntakeController
toolCatalogBuilder *agentruntime.ToolCatalogBuilder
taskLauncher *agentruntime.TaskLauncher
schedulePoller *scheduler.SchedulePoller
taskRetentionSweeper *scheduler.TaskRetentionSweeper
connectorRuntime *connectors.ConnectorRuntime
agentReplyStore *apiconnector.ReplyStore
connectorEventHandler *httpserver.ConnectorEventHandler
protocolIdentity protocolIdentityComponents
acpSessionServer *acpsession.Server
router http.Handler
startupError error
learningStore *learning.Store
learningCoordinator *learning.Coordinator
}
type ApplicationOption func(*applicationOptions)
type applicationOptions struct {
bundledACPFactory harnessdriver.ACPFactory
bundledToolSelectorMake BundledToolSelectorFactory
questionWorderMake QuestionWorderFactory
fileJournalSocketPath string
}
func WithFileJournalSocket(socketPath string) ApplicationOption {
return func(options *applicationOptions) { options.fileJournalSocketPath = socketPath }
}
func WithQuestionWorder(newQuestionWorder QuestionWorderFactory) ApplicationOption {
return func(options *applicationOptions) { options.questionWorderMake = newQuestionWorder }
}
func WithBundledToolSelector(newBundledToolSelector BundledToolSelectorFactory) ApplicationOption {
return func(options *applicationOptions) { options.bundledToolSelectorMake = newBundledToolSelector }
}
func WithBundledACPFactory(bundledACPFactory harnessdriver.ACPFactory) ApplicationOption {
return func(options *applicationOptions) { options.bundledACPFactory = bundledACPFactory }
}
func NewApplication(runtimeConfiguration config.RuntimeConfiguration, policyPath string, inbound InboundOptions, options ...ApplicationOption) *Application {
return newApplication(newApplicationComponents(runtimeConfiguration, policyPath, inbound, options))
}
func newApplicationComponents(runtimeConfiguration config.RuntimeConfiguration, policyPath string, inbound InboundOptions, applicationOptionList []ApplicationOption) applicationComponents {
options := applicationOptions{}
for _, option := range applicationOptionList {
option(&options)
}
components := applicationComponents{runtimeConfiguration: runtimeConfiguration, policyPath: policyPath}
components.foundation = newRuntimeFoundation(runtimeConfiguration, policyPath)
logger := components.foundation.logger
components.directory = newIdentityDirectory(components.foundation.database, components.foundation.policyDocument, logger)
components.services = newTaskServices(runtimeConfiguration, components.foundation.database, components.directory.companyProvider, logger)
components.kernel = newAgentKernel(runtimeConfiguration, options.bundledACPFactory, options.bundledToolSelectorMake, options.questionWorderMake, components.services, logger)
components.memory = newMemoryComponents(runtimeConfiguration, components.kernel, components.services, components.directory.identityService, logger)
useMemoryMergingPersonRepository(&components)
components.learningStore, components.startupError = openLearningStore(runtimeConfiguration.Terminal.WorkspaceRootPath)
learningCoordinator, learningError := newLearningCoordinator(runtimeConfiguration, components.learningStore, components.kernel.taskTierLanguageModels.Low, logger)
components.learningCoordinator = learningCoordinator
components.startupError = firstNonNilError(components.startupError, learningError)
components.backupCoordinator = backup.NewCoordinator(buildBackupManifest(runtimeConfiguration, components.foundation.database))
components.trustedClock = runtimecontrol.NewTrustedClock(languageModelEndpoints(runtimeConfiguration.LanguageModel), companyZoneOf(components.directory.companyProvider), logger)
components.taskIntakeController = runtimecontrol.NewTaskIntakeController(components.trustedClock)
components.taskIntakeController.SetQuiesced(components.kernel.languageModelError != nil)
components.toolCatalogBuilder = newToolCatalogBuilder(runtimeConfiguration, components.kernel, components.services, components.directory, components.memory, logger)
if components.learningStore != nil {
components.toolCatalogBuilder.UseLearnedSkillLoader(func(audience string) []learning.Skill {
return components.learningStore.List(audience, false)
})
}
if components.learningCoordinator != nil {
components.learningCoordinator.UseAvailableTools(func() []string {
return components.toolCatalogBuilder.AvailableToolNames()
})
components.learningCoordinator.Report = func(errorValue error) {
logger.Error("learning.review_failed", "error", errorValue.Error())
}
}
components.toolCatalogBuilder.UseToolSelector(components.kernel.toolSelector)
components.taskLauncher = newTaskLauncher(runtimeConfiguration, components.foundation, components.directory, components.kernel, components.services, components.toolCatalogBuilder)
components.taskLauncher.UseTaskObserver(learningTaskObserver(components.learningCoordinator, components.services.taskRunService))
components.schedulePoller = newSchedulePoller(runtimeConfiguration, components.services, components.directory.identityService, components.taskLauncher, components.taskIntakeController, logger)
configureMorningBriefing(components.schedulePoller, runtimeConfiguration, components.directory, components.kernel, logger)
logger.Info("application.initializing", "stage", "connector_runtime")
components.taskRetentionSweeper = newTaskRetentionSweeper(runtimeConfiguration, components.services, logger)
components.connectorRuntime = newConnectorRuntime(runtimeConfiguration, components.foundation, components.directory, components.kernel, components.services, components.taskLauncher, components.backupCoordinator, components.taskIntakeController)
if options.fileJournalSocketPath != "" {
components.connectorRuntime.UseFileKeeper(filejournal.Keeper(options.fileJournalSocketPath))
}
if !runtimeConfiguration.Memory.ExtractionDisabled {
components.connectorRuntime.UseHeardMemory(components.memory.stores)
}
installThreadAsker(components.connectorRuntime, runtimeConfiguration, inbound, components.kernel.toolCatalog.approvalGate, components.directory)
registerChatdAdapters(components.connectorRuntime, runtimeConfiguration, logger)
components.agentReplyStore = newAgentReplyStore(runtimeConfiguration)
components.connectorRuntime.RegisterAdapter(apiconnector.NewAdapter(components.directory.identityService, components.agentReplyStore))
components.connectorEventHandler = httpserver.NewConnectorEventHandler(components.connectorRuntime, inbound.admitsConnectorHTTPEvent())
components.acpSessionServer = newACPSessionServer(inbound, components.kernel, components.directory, components.taskLauncher, components.connectorRuntime, components.services.taskRunService, components.trustedClock, logger)
logger.Info("application.initializing", "stage", "router")
components.protocolIdentity = newProtocolIdentity(runtimeConfiguration, components.kernel.capabilityClient)
components.startupError = firstNonNilError(components.foundation.startupError, components.kernel.startupError, components.startupError)
components.router = httpserver.NewRouter(newRouterDependencies(components))
return components
}
func newApplication(components applicationComponents) *Application {
components.foundation.logger.Info("application.initializing", "stage", "ready")
return &Application{
httpServer: &http.Server{
Addr: deriveListenAddress(components.runtimeConfiguration.BaseURL),
Handler: components.router,
},
connectorRuntime: components.connectorRuntime,
taskRunService: components.services.taskRunService,
interruptedTaskResumer: components.connectorRuntime,
runtimeLogger: components.foundation.runtimeLogger,
terminalService: components.kernel.terminalService,
database: components.foundation.database,
startupError: components.startupError,
schedulePoller: components.schedulePoller,
taskRetentionSweeper: components.taskRetentionSweeper,
memoryStores: components.memory.stores,
heldCircles: components.directory.identityService.HeldCircles,
isOnTheRoster: func(personID string) bool {
_, isFound := components.directory.identityService.FindPersonAccess(personID)
return isFound
},
learningCoordinator: components.learningCoordinator,
schedulePollSecond: components.runtimeConfiguration.Scheduler.SchedulePollIntervalSecond,
taskRetentionIntervalMinute: components.runtimeConfiguration.Scheduler.RetentionCheckIntervalMinute,
interruptedTaskResumeDelay: 2 * time.Second,
languageModelConfigured: components.kernel.taskTierLanguageModels.High != nil,
languageModelError: components.kernel.languageModelError,
protocolIdentityChecker: components.protocolIdentity.checker,
protocolIdentityExpected: components.protocolIdentity.expected,
protocolIdentityStatus: components.protocolIdentity.status,
refreshSkillIndex: components.kernel.refreshSkillIndex,
acpSessionServer: components.acpSessionServer,
trustedClock: components.trustedClock,
}
}
func firstNonNilError(errorValues ...error) error {
for _, errorValue := range errorValues {
if errorValue != nil {
return errorValue
}
}
return nil
}
func firstNonEmptyString(values ...string) string {
for _, value := range values {
if strings.TrimSpace(value) != "" {
return strings.TrimSpace(value)
}
}
return ""
}
// useMemoryMergingPersonRepository hands the person repository the memory it
// has to move when two records turn out to be the same person. It runs after
// the memory components because they are what opens those files.
func useMemoryMergingPersonRepository(components *applicationComponents) {
if components.foundation.database.SQL == nil || components.memory.stores == nil {
return
}
repository := postgres.NewPersonRepositoryMergingMemory(components.foundation.database, components.memory.stores)
components.services.repositories.person = repository
components.services.repositories.personReferenceCanonicalizer = repository
}