package server import ( "context" "errors" "fmt" "log/slog" "net/http" "strings" "time" "github.com/google/uuid" "github.com/tidbcloud/AgentTeam/internal/api" "github.com/tidbcloud/AgentTeam/internal/api/admin" "github.com/tidbcloud/AgentTeam/internal/api/agent" "github.com/tidbcloud/AgentTeam/internal/api/app" "github.com/tidbcloud/AgentTeam/internal/api/auth" "github.com/tidbcloud/AgentTeam/internal/api/billing" "github.com/tidbcloud/AgentTeam/internal/api/channel" "github.com/tidbcloud/AgentTeam/internal/api/daemon" "github.com/tidbcloud/AgentTeam/internal/api/im" "github.com/tidbcloud/AgentTeam/internal/api/im/fanout" knowledgeqaapi "github.com/tidbcloud/AgentTeam/internal/api/knowledgeqa" "github.com/tidbcloud/AgentTeam/internal/api/llm" "github.com/tidbcloud/AgentTeam/internal/api/media" apiOnboarding "github.com/tidbcloud/AgentTeam/internal/api/onboarding" "github.com/tidbcloud/AgentTeam/internal/api/org" "github.com/tidbcloud/AgentTeam/internal/api/provider" "github.com/tidbcloud/AgentTeam/internal/api/public" scheduleapi "github.com/tidbcloud/AgentTeam/internal/api/schedule" serviceaccountapi "github.com/tidbcloud/AgentTeam/internal/api/serviceaccount" "github.com/tidbcloud/AgentTeam/internal/api/task" "github.com/tidbcloud/AgentTeam/internal/api/team" teamworkflowapi "github.com/tidbcloud/AgentTeam/internal/api/teamworkflow" triggerapi "github.com/tidbcloud/AgentTeam/internal/api/trigger" "github.com/tidbcloud/AgentTeam/internal/api/webagent" workflowbindingapi "github.com/tidbcloud/AgentTeam/internal/api/workflow/binding" workflowinvocationapi "github.com/tidbcloud/AgentTeam/internal/api/workflow/invocation" workflowplanapi "github.com/tidbcloud/AgentTeam/internal/api/workflow/plan" run "github.com/tidbcloud/AgentTeam/internal/api/workflow/run" "github.com/tidbcloud/AgentTeam/internal/api/workspace" "github.com/tidbcloud/AgentTeam/internal/apr/agentteambridge" "github.com/tidbcloud/AgentTeam/internal/apr/types" "github.com/tidbcloud/AgentTeam/internal/billing/alipay" "github.com/tidbcloud/AgentTeam/internal/config" "github.com/tidbcloud/AgentTeam/internal/coredb" "github.com/tidbcloud/AgentTeam/internal/extension/bootstrap" "github.com/tidbcloud/AgentTeam/internal/extension/providers" mailapi "github.com/tidbcloud/AgentTeam/internal/extension/providers/mail/mailapi" mailprotocol "github.com/tidbcloud/AgentTeam/internal/extension/providers/mail/protocol" mailruntime "github.com/tidbcloud/AgentTeam/internal/extension/providers/mail/mailruntime" mailstore "github.com/tidbcloud/AgentTeam/internal/extension/providers/mail/mailstore" mailtransport "github.com/tidbcloud/AgentTeam/internal/extension/providers/mail/transport" searchservice "github.com/tidbcloud/AgentTeam/internal/extension/providers/searchengine/runtime" "github.com/tidbcloud/AgentTeam/internal/extension/registry" extensionruntime "github.com/tidbcloud/AgentTeam/internal/extension/runner" "github.com/tidbcloud/AgentTeam/internal/forwarder" "github.com/tidbcloud/AgentTeam/internal/kbgovernance" "github.com/tidbcloud/AgentTeam/internal/maas/capability" "github.com/tidbcloud/AgentTeam/internal/maas/availability" "github.com/tidbcloud/AgentTeam/internal/meshxassets" observabilitymetrics "github.com/tidbcloud/AgentTeam/internal/observability/metrics" "github.com/tidbcloud/AgentTeam/internal/router" "github.com/tidbcloud/AgentTeam/internal/runtimecontrol" execschedule "github.com/tidbcloud/AgentTeam/internal/schedule" "github.com/tidbcloud/AgentTeam/internal/schedule/manage" "github.com/tidbcloud/AgentTeam/internal/service/agentruntime" "github.com/tidbcloud/AgentTeam/internal/service/agentapproval" "github.com/tidbcloud/AgentTeam/internal/service/agentmemory" "github.com/tidbcloud/AgentTeam/internal/service/approvalapp" "github.com/tidbcloud/AgentTeam/internal/service/appruntime" "github.com/tidbcloud/AgentTeam/internal/service/authflow" "github.com/tidbcloud/AgentTeam/internal/service/channeladmin" "github.com/tidbcloud/AgentTeam/internal/service/collaborator" extensionflow "github.com/tidbcloud/AgentTeam/internal/service/extensionflow" knowledgeqaservice "github.com/tidbcloud/AgentTeam/internal/service/knowledgeqa" "github.com/tidbcloud/AgentTeam/internal/service/maas" messageservice "github.com/tidbcloud/AgentTeam/internal/service/message" "github.com/tidbcloud/AgentTeam/internal/service/organization" "github.com/tidbcloud/AgentTeam/internal/service/prwatch" "github.com/tidbcloud/AgentTeam/internal/service/role/agents" "github.com/tidbcloud/AgentTeam/internal/service/role/execution" serviceaccountservice "github.com/tidbcloud/AgentTeam/internal/service/serviceaccount" "github.com/tidbcloud/AgentTeam/internal/service/serviceevent" "github.com/tidbcloud/AgentTeam/internal/service/team/roleproposal" "github.com/tidbcloud/AgentTeam/internal/service/teamworkflow" workspaceservice "github.com/tidbcloud/AgentTeam/internal/service/workspace" "github.com/tidbcloud/AgentTeam/internal/store/agentops" appstore "github.com/tidbcloud/AgentTeam/internal/store/app" "github.com/tidbcloud/AgentTeam/internal/store/knowledgeqa" "github.com/tidbcloud/AgentTeam/internal/store/message" orgstore "github.com/tidbcloud/AgentTeam/internal/store/organization" schedulestore "github.com/tidbcloud/AgentTeam/internal/store/schedule" "github.com/tidbcloud/AgentTeam/internal/store/taskboard" bindingstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/binding" calendarstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/calendar" invocationstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/invocation" planstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/plan" runstore "github.com/tidbcloud/AgentTeam/internal/store/workflow/run" "github.com/tidbcloud/AgentTeam/internal/trigger" "github.com/tidbcloud/AgentTeam/internal/unread" "github.com/tidbcloud/AgentTeam/internal/versioning" workflowagentruntime "github.com/tidbcloud/AgentTeam/internal/workflow/agentruntime" bindingmanage "github.com/tidbcloud/AgentTeam/internal/workflow/bindingmanage" calendar "github.com/tidbcloud/AgentTeam/internal/workflow/calendar" invocation "github.com/tidbcloud/AgentTeam/internal/workflow/invocation" invocationmanage "github.com/tidbcloud/AgentTeam/internal/workflow/invocationmanage" planmanage "github.com/tidbcloud/AgentTeam/internal/workflow/planmanage" runmanage "github.com/tidbcloud/AgentTeam/internal/workflow/runmanage" "github.com/tidbcloud/AgentTeam/internal/workflow/orchestrator" "github.com/tidbcloud/AgentTeam/internal/workspace/owner" ) type daemonTransportMetricsProvider interface { DaemonTransportMetrics() *observabilitymetrics.DaemonTransport } // wireSide holds side products from wiring that the Server needs beyond // router.Dependencies (e.g. background workers). type wireSide struct { scheduleService *execschedule.Service workflowInvocationSchedules *invocation.ScheduleWorker scopeFanoutRunner *fanout.ScopeFanoutRunner unreadRuntime *unread.Service assetAccessWorker *meshxassets.Worker workflowRunWorker *orchestrator.Worker internalAppDispatcher *appruntime.InternalDispatcher approvalProjectionReconciler *approvalapp.ProjectionReconciler platformAccounts *serviceaccountservice.PlatformService knowledgeQAWorker *knowledgeqaservice.Worker operationWorker *agentops.OperationWorker roleManagementWorker *execution.Worker wikiGenerationApprovals kbgovernance.WikiGenerationApprovalGateway wikiGenerationTerminals kbgovernance.WikiGenerationTerminalGateway publicationApprovals kbgovernance.PublicationApprovalGateway closers []func() } func schedulerWorkspaceGroupIDForConfig(cfg config.ServerConfig) string { if cfg.ProfileOrDefault() != config.ServerProfileWorkspace { return "" } return strings.TrimSpace(cfg.WorkspaceGroupID) } // profileOwnsWorkspaceDataPlane reports whether the profile owns the workspace // data-plane: the background workers, the MeshX runtime, and the // workspace-sharded stores. Only the all and workspace profiles do. The central // profile delegates the data-plane, and the extension profile is a narrow // runtime that MUST NOT inherit workspace workers (spec §6.1). Worker // constructors use the negation to short-circuit construction for central and // extension alike. func profileOwnsWorkspaceDataPlane(cfg config.ServerConfig) bool { switch cfg.ProfileOrDefault() { case config.ServerProfileAll, config.ServerProfileWorkspace: return true default: return false } } // buildExtensionRegistry constructs the immutable stage-2 Registry from the // trusted Builtin provider registrations (spec §5.1): the aggregated provider // Extensions are validated (catalog/executor/auth/distribution consistency) // and registered in one fail-closed startup pass. func buildExtensionRegistry(deps providers.Deps) (*registry.Registry, error) { reg := registry.New() list, err := providers.Extensions(deps) if err != nil { return nil, err } if err := bootstrap.Register(reg, list); err != nil { return nil, err } return reg, nil } // providerCredentialSource adapts the App store onto the runtime credential // and prepared-connection seams used by authenticated provider executors. type providerCredentialSource struct { store *appstore.Store } func (source providerCredentialSource) ResolveProviderCredentials( ctx context.Context, workspaceID string, installationID string, ) (extensionruntime.Credentials, error) { credentials, _, err := source.store.ResolveExtensionProviderRuntimeConnection( ctx, workspaceID, installationID, appstore.ExtensionProviderTarget{}, extensionruntime.ConnectionConfigContract{}, ) if err != nil { return nil, err } return extensionruntime.Credentials(credentials), nil } func (source providerCredentialSource) ResolveProviderConnection( ctx context.Context, workspaceID string, installationID string, configContract extensionruntime.ConnectionConfigContract, ) (extensionruntime.ResolvedConnection, error) { credentials, config, err := source.store.ResolveExtensionProviderRuntimeConnection( ctx, workspaceID, installationID, appstore.ExtensionProviderTarget{}, configContract, ) if err != nil { return extensionruntime.ResolvedConnection{}, err } runtimeConnection := extensionruntime.ResolvedConnection{ Credentials: extensionruntime.Credentials(credentials), Config: extensionruntime.ConnectionConfig(config), } if err := configContract.Validate(runtimeConnection.Config); err != nil { return extensionruntime.ResolvedConnection{}, err } return runtimeConnection, nil } // extensionRegistryForProfile bootstraps the Extension Registry only for // profiles that register Extension routes (all|extension). central and // workspace never register /runtime/ext, so they MUST be insulated from a // broken provider Extension: they skip the bootstrap entirely and // return a nil Registry (spec §6.1) rather than failing to start. func extensionRegistryForProfile( profile router.Profile, deps providers.Deps, ) (*registry.Registry, error) { if profile != router.ProfileAll && profile != router.ProfileExtension { return nil, nil } return buildExtensionRegistry(deps) } // wireRouterDeps builds all HTTP-layer services and assembles the // router.Dependencies struct from foundational infrastructure. func wireRouterDeps(ctx context.Context, cfg config.Config, logger *slog.Logger, inf *infra, httpWrapper httpClientWrapper, profile router.Profile, managedWorkspaces owner.ManagedWorkspaceSet, embeddedMeshXTransport http.RoundTripper, kbGovernanceAuthority kbgovernance.Authority) (router.Dependencies, wireSide, error) { meshXBaseURL, meshXHTTPClient := effectiveMeshXEndpoint(cfg, httpWrapper, profile, embeddedMeshXTransport) ownsMeshXRuntime := profileComposesMeshX(profile, cfg.MeshX.Enabled) stores := inf.stores appRunnerService := appruntime.NewRunnerService(appruntime.RunnerServiceOptions{Store: stores.ServiceActor, Authority: stores.App}) internalAppHandlers := appruntime.NewInternalHandlerRegistry() agentLogger := logger.With(slog.String("domain", "agent")) authLogger := logger.With(slog.String("domain", "auth")) channelLogger := logger.With(slog.String("domain", "channel")) knowledgeLogger := logger.With(slog.String("domain", "kb")) messageLogger := logger.With(slog.String("domain", "message")) roleLogger := logger.With(slog.String("domain", "role")) scheduleLogger := logger.With(slog.String("domain", "schedule")) taskLogger := logger.With(slog.String("domain", "task")) workflowLogger := logger.With(slog.String("domain", "workflow")) workspaceLogger := logger.With(slog.String("domain", "workspace")) providerWatchService := prwatch.New(stores.App) mailPersistence := mailstore.NewWithNamedApps( inf.coreDataDB, stores.App, ) mailConnector := &mailtransport.Dialer{ Resolver: mailtransport.NetResolver{}, } mailIMAP := mailprotocol.NewIMAPClient(mailprotocol.ClientOptions{ Connector: mailConnector, }) mailSMTP := mailprotocol.NewSMTPClient(mailprotocol.ClientOptions{ Connector: mailConnector, }) approvalAppProjector := approvalapp.NewProjector( stores.App, approvalapp.NewProvisioner(inf.coreDataDB), ) if err := approvalAppProjector.Register(); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "register Approval notification contract: %w", err, ) } mailAttachmentAuthority := mailAttachmentAuthority{ attachments: stores.Attachment, app: stores.App, } // The fresh-authorization fence starts default-deny and is bound to the // authoritative extensionflow resolver later in the wiring order, once // the profile's catalog service exists. mailAuthorizer := &mailInvocationAuthorizer{} mailExecutionService := mailruntime.New( inf.coreDataDB, inf.appCredentialKey, mailruntime.Options{ Connections: mailPersistence, LegacyCredentialKeys: inf.legacyAppCredentialKeys, IMAP: mailIMAP, SMTP: mailSMTP, Attachments: mailAttachmentAuthority, Materializer: mailAttachmentAuthority, Authorizer: mailAuthorizer, }, ) mailConnectionService := mailapi.NewService( mailIMAP, mailSMTP, mailPersistence, inf.appCredentialKey, ) searchEngineService := searchservice.New(stores.App) providerConnections := providerCredentialSource{store: stores.App} providerDeps := providers.Deps{ WatchService: providerWatchService, SearchService: searchEngineService, MailService: mailExecutionService, Credentials: providerConnections, Connections: providerConnections, } extensionProviderIndex, err := newExtensionProviderIndex(stores.App, providerDeps) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "build extension provider index: %w", err) } identityStore := stores.Identity workspaceStore := stores.Workspace cachedWorkspaceResolver := api.NewCachedWorkspaceResolver(workspaceStore, api.DefaultWorkspaceResolverCacheTTL) workspaceSummaryInvalidator := workspaceSummaryCacheInvalidator(cachedWorkspaceResolver) workspaceResolver := api.NewOwnedWorkspaceResolver(cachedWorkspaceResolver, managedWorkspaces) unreadRuntime, err := newUnreadRuntimeForConfig(cfg, unreadRuntimeStoreAdapter{store: stores.UnreadRuntime}, messageLogger.With(slog.String("component", "unread_runtime"))) if err != nil { return router.Dependencies{}, wireSide{}, err } // Runtime/control-plane live events now flow through the IM streaming hub // (StreamEventHub) via LiveEventEmitter; the legacy Socket.IO realtime // service is no longer wired in production. See events_ws_handler for the // /events/stream bootstrap that replaces realtime.Service bootstrap. imStreamHub := im.NewStreamEventHub() liveEmitter := im.NewLiveEventEmitter(imStreamHub, stores.App, messageLogger.With(slog.String("component", "live_event_emitter"))) if unreadRuntime != nil { liveEmitter.SetUnreadCurrentDeltaReader(unreadRuntime) unreadRuntime.SetDeltaSink(liveEmitter) } realtimeTaskSink := im.NewTaskChangeBridge(liveEmitter). WithUnreadRuntime(unreadRuntime, liveEmitter). WithTaskScopeLifecycleGuard(stores.Message) taskUnreadBridge := newTaskUnreadRuntimeBridge(unreadRuntime, liveEmitter) taskChangeSink := newTaskChangeFanout(realtimeTaskSink, taskUnreadBridge) stores.App.SetTaskChangeSink(taskChangeSink) taskStore := taskboard.NewWithOptions(inf.coreDataDB, taskboard.StoreOptions{ TaskChangeSink: taskChangeSink, }) organizationCore := organization.NewCoreBridge(organization.Dependencies{ Identity: stores.Identity, Workspace: stores.Workspace, Scopes: stores.PrivateAgents, ServiceAccounts: stores.ServiceAccount, DefaultWorkspaceGroupID: cfg.Server.DefaultWorkspaceGroupIDOrDefault(), }) organizationService := orgstore.NewWithDBOptions(inf.coreDataDB, organizationCore, orgstore.ServiceOptions{ FeishuAppID: cfg.Feishu.AppID, FeishuAppSecret: cfg.Feishu.AppSecret, FeishuHTTPClient: newOutboundHTTPClient(httpWrapper, "feishu", 10*time.Second, nil), FeishuCredentialEncryptionKey: inf.feishuCredentialKey, CheckOrganizationQuotaInTx: stores.Plan.CheckOrganizationQuotaInTx, }) workspaceSummaryOrganizations := withWorkspaceSummaryInvalidatingOrganizationProvider(organizationService, workspaceSummaryInvalidator, liveEmitter) llmProxyTransporter, err := llmProxyTransporter(cfg) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("build llm proxy transporter: %w", err) } billingAlipayClient := alipay.NewClient(alipay.Config{ AppID: cfg.Billing.Alipay.AppID, AppPrivateKey: cfg.Billing.Alipay.AppPrivateKey, AlipayPublicKey: cfg.Billing.Alipay.AlipayPublicKey, GatewayURL: cfg.Billing.Alipay.GatewayURL, }, newOutboundHTTPClient(httpWrapper, "billing.alipay", cfg.Billing.Alipay.RequestTimeout, nil)) billingMaaSClient := billing.NewMaaSClient( cfg.LLM.MaaSBaseURL, cfg.LLM.MaaSInternalBearerToken, newOutboundHTTPClient(httpWrapper, "billing.maas", 10*time.Second, nil), ) var hostedAvailabilityClient workspace.HostedAvailabilityClient var hostedPricingClient workspace.HostedPricingClient var hostedClient *availability.Client if strings.TrimSpace(cfg.LLM.MaaSBaseURL) != "" && strings.TrimSpace(cfg.LLM.MaaSInternalBearerToken) != "" { hostedClient = availability.NewClient( cfg.LLM.MaaSBaseURL, cfg.LLM.MaaSInternalBearerToken, newOutboundHTTPClient(httpWrapper, "workspace.maas_hosted", 10*time.Second, nil), ) hostedAvailabilityClient = hostedClient hostedPricingClient = workspaceHostedPricingClient{next: billingMaaSClient} } var hostedAgentValidator *maas.Validator if hostedAvailabilityClient != nil { hostedAgentValidator = maas.NewValidator(maas.Dependencies{ Organizations: organizationService, Availability: hostedAvailabilityClient, }) } providerHTTPClient := newOutboundHTTPClient(httpWrapper, "provider.api", 10*time.Second, llmProxyTransporter) llmUpstreamHTTPClient := newOutboundHTTPClient(httpWrapper, "llm.upstream", 10*time.Minute, llmProxyTransporter) llmSetupHTTPClient := newOutboundHTTPClient(httpWrapper, "llm.setup", 30*time.Second, llmProxyTransporter) agentteamLLMRefresher := agentteambridge.NewRefresherWithOptions(organizationService, agentteambridge.RefresherOptions{ HTTPClients: agentteambridge.ProviderHTTPClients{ Upstream: llmUpstreamHTTPClient, Setup: llmSetupHTTPClient, }, Transporter: llmProxyTransporter, Logger: logger, }) if _, err := agentteamLLMRefresher.Reload(ctx); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("load APR-backed organization providers: %w", err) } teamRoleProposals := roleproposal.New(inf.coreDataDB) teamWorkflows := teamworkflow.New(inf.coreDataDB, teamworkflow.ServiceOptions{ LifecycleEvents: liveEmitter, }) teamWorkflowBrowserService := teamworkflowapi.NewStoreService(inf.coreDataDB, teamWorkflows) channelManagement := channeladmin.New(channeladmin.Dependencies{ DB: inf.coreDataDB, Channels: stores.Channel, Apps: appstore.NewChannelManagementPort(stores.App), Owners: stores.App, Workspaces: stores.App, Realtime: liveEmitter, Unread: unreadRuntime, }) agentDeps := agent.StorePortsDeps{ Agents: stores.Agent, App: stores.App, Channel: stores.Channel, Identity: stores.Identity, Messages: stores.Message, Reactions: stores.Reaction, Rewind: stores.Rewind, Review: stores.Review, Teams: stores.Team, ApprovalSubjects: nil, SourceContexts: stores.App, ChannelManagement: channelManagement, } agentRuntime := agentruntime.New(agentruntime.Dependencies{ Agents: stores.Agent, Messages: stores.Message, Recovery: stores.Rewind, WebAgents: stores.App, WebRecovery: stores.Rewind, HostedAgents: hostedAgentValidator, IsRuntimeOffline: func(err error) bool { return errors.Is(err, runtimecontrol.ErrMachineOffline) }, MaaSRuntimeBaseURL: cfg.LLM.MaaSRuntimeBaseURL, }) daemonHub := daemon.NewWithOptions(agentLogger.With(slog.String("component", "daemon_hub")), daemon.NewStore(stores.Agent, stores.Rewind, agentRuntime.PrepareAgentStart, stores.Workspace), cfg.PublicBaseURL, liveEmitter, daemon.HubOptions{ MaaSRuntimeBaseURL: cfg.LLM.MaaSRuntimeBaseURL, ElasticRequestLinkConfig: daemon.ElasticRequestLinkManagerConfig{ Metrics: observabilityDaemonTransportMetrics(httpWrapper), }, HintReader: daemon.NewCatalogHintReader(inf.coreDataDB), }, stores.AgentOps) agentRuntime.SetRuntimeControl(daemonHub) // Start the Extension catalog delivery agentteam for profiles that own daemon // clients (all|central — NOT the split ProfileExtension which has no // WebSocket clients). The agentteam claims pending catalog updates, resolves the // agent's machine, pushes via SyncMachineCatalog, and marks // delivered/requeue/fail (spec §8.1, stage 6). if profile == router.ProfileAll || profile == router.ProfileCentral { // The delivery worker re-reads authoritative state at delivery time and does // NOT depend on the in-memory Registry, so it runs on every profile that owns // WebSocket clients (all|central) regardless of whether /runtime/ext routes // are registered. central has no registry yet still owns clients, so it must // not be gated on extReg != nil (stage-6 final-review blocker 4). extReg, _ := extensionRegistryForProfile( profile, providerDeps, ) extensionflow.New( inf.coreDataDB, extReg, nil, extensionflow.Options{}, ).StartCatalogDeliveryLoop(ctx, daemonHub, 5*time.Second) } replyRootPreloader := fanout.NewScopeReplyThreadRootPreloader(stores.Message) imStreamWriter := im.NewPublishingStreamEventWriter(stores.IM, imStreamHub) knowledgeQATurnUpdates := newKnowledgeQATurnStreamNotifier(imStreamWriter, knowledgeLogger.With(slog.String("component", "turn_updates"))) deliveryFanout := fanout.NewDeliveryFanout(daemonHub, imStreamWriter, messageLogger.With(slog.String("component", "delivery_fanout"))). WithReplyThreadRootPreloader(replyRootPreloader). WithScopeFanoutMarker(stores.App). WithTaskScopeLifecycleGuard(stores.Message). WithUnreadRuntime(unreadRuntime). WithUnreadDeltaSink(liveEmitter) stores.App.SetCommittedMessageFanout(deliveryFanout) channelManagement.MessageFanout = deliveryFanout workflowRunStore := runstore.NewWithOptions(inf.coreDataDB, runstore.StoreOptions{ WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(cfg.Server), }) if err := workflowRunStore.Validate(); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("build Workflow Run store: %w", err) } workflowApprovalDecisions, err := orchestrator.NewApprovalDecisionService(workflowRunStore) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("build Workflow Run approval decision service: %w", err) } serviceAccountDeletionObserver := newServiceAccountUnreadRuntimeBridge(unreadRuntime, liveEmitter) serviceAccountFlow := serviceaccountservice.New(inf.coreDataDB) serviceAccountFlow.SetLocalAccountDeletionObserver(serviceAccountDeletionObserver) platformServiceAccounts := serviceaccountservice.NewPlatformService(inf.coreDataDB, serviceaccountservice.PlatformServiceOptions{ ControlPlane: cfg.Server.ProfileOrDefault() != config.ServerProfileWorkspace, ManagedWorkspaces: managedWorkspaces, MessageWriter: stores.App, LocalAccountDeletionObserver: serviceAccountDeletionObserver, }) emailVerificationRequired := cfg.EmailVerificationRequired() apiForwarderDeps := forwarder.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, BaseURL: meshXBaseURL, RequestTimeout: cfg.MeshX.RequestTimeout, TrustedJWTAudience: "meshx", HTTPClient: meshXHTTPClient, EmailVerificationRequired: emailVerificationRequired, PasswordAuthDisabled: cfg.Auth.PasswordAuthDisabled, Logger: logger, AgentThreadSessions: stores.App, AgentStore: stores.App, MeshXEnabled: ownsMeshXRuntime, BillingEnabled: cfg.Billing.Enabled, PlanStore: stores.Plan, ManagedWorkspaces: managedWorkspaces, EnforceRouteWorkspaceOwner: profile == router.ProfileWorkspace, } var roleOperationGateway *forwarder.RoleOperationGateway var roleReadGateway *forwarder.RoleReadGateway var skillGateway *forwarder.SkillGateway if ownsMeshXRuntime && strings.TrimSpace(apiForwarderDeps.BaseURL) != "" && cfg.Server.ProfileOrDefault() != config.ServerProfileCentral { roleOperationGateway, err = forwarder.NewRoleOperationGateway( apiForwarderDeps, ) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "build Role management gateway: %w", err, ) } roleReadGateway, err = forwarder.NewRoleReadGateway( apiForwarderDeps, ) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "build Role read gateway: %w", err, ) } skillGateway, err = forwarder.NewSkillGateway(apiForwarderDeps) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "build Skill gateway: %w", err, ) } } roleManagementTerminalObserver := approvalapp.NewRoleManagementTerminalObserver( stores.Approval, approvalAppProjector, ) roleManagementRuntime, err := newRoleManagementRuntimeForConfig( roleManagementRuntimeOptions{ Config: cfg, DB: inf.coreDataDB, Approvals: stores.Approval, WorkspaceLister: stores.Workspace, Gateway: roleOperationGateway, LoadProjectedTerminal: roleManagementTerminalObserver.LoadProjectedTerminal, ProjectTerminal: roleManagementTerminalObserver.ProjectTerminal, Logger: roleLogger, }, ) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "build Role management runtime: %w", err, ) } approvalSubjects := approvalapp.NewSubjectRegistry(stores.Approval) var wikiGenerationApprovals kbgovernance.WikiGenerationApprovalGateway var wikiGenerationTerminals kbgovernance.WikiGenerationTerminalGateway var publicationApprovals kbgovernance.PublicationApprovalGateway if kbGovernanceAuthority != nil { if err := approvalSubjects.Register(approvalapp.NewWikiGenerationSubjectHandler( stores.Approval, kbGovernanceAuthority, inf.coreDataDB, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("register Wiki generation Approval subject: %w", err) } wikiGenerationApprovals = approvalapp.NewWikiGenerationApprovalGateway( approvalSubjects, stores.Approval, inf.coreDataDB, approvalAppProjector, ) wikiGenerationTerminals = approvalapp.NewWikiGenerationTerminalGateway( kbGovernanceAuthority, inf.coreDataDB, approvalAppProjector, ) if err := approvalSubjects.Register(approvalapp.NewPublicationSubjectHandler( stores.Approval, kbGovernanceAuthority, inf.coreDataDB, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("register publication Approval subject: %w", err) } publicationApprovals = approvalapp.NewPublicationApprovalGateway( approvalSubjects, stores.Approval, inf.coreDataDB, approvalAppProjector, ) } if err := approvalSubjects.Register(approvalapp.NewServiceAccountCreateSubjectHandler( stores.Approval, serviceAccountFlow, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("register service account Approval subject: %w", err) } if err := approvalSubjects.Register(approvalapp.NewTeamRoleProposalSubjectHandler( stores.Approval, teamRoleProposals, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("register Team role proposal Approval subject: %w", err) } agentManagement := agentapproval.New(inf.coreDataDB, stores.Approval, stores.AgentOps) if err := approvalSubjects.Register(approvalapp.NewAgentManagementSubjectHandler( stores.Approval, agentManagement, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("register agent management Approval subject: %w", err) } if skillGateway != nil { skillDeleteAuthority := agent.NewAgentSkillDeleteAuthority( stores.App, workspaceResolver, skillGateway, ) if err := approvalSubjects.Register(approvalapp.NewSkillDeleteSubjectHandler( stores.Approval, skillDeleteAuthority, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "register Skill delete Approval subject: %w", err, ) } } roleManagementSide := wireSide{} if roleManagementRuntime != nil { roleManagementService := roleManagementRuntime.service roleManagementWorker := roleManagementRuntime.worker if err := approvalSubjects.Register(approvalapp.NewRoleManagementSubjectHandler( stores.Approval, roleManagementService, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "register Role management Approval subject: %w", err, ) } roleManagementSide = wireSide{ roleManagementWorker: roleManagementWorker, } } // Human collaborator requests: taskboard retains the durable request row and // the generic Approval (approver = execution owner, 7-day validity) is the // workflow source of truth. Approved requests grant via the same // transaction-local primitive as direct owner-add. collaboratorRequestService := collaborator.NewWithStore(inf.coreDataDB, stores.Approval, taskStore) if err := approvalSubjects.Register(approvalapp.NewCollaboratorRequestSubjectHandler( stores.Approval, collaboratorRequestService, approvalAppProjector, )); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("register collaborator request Approval subject: %w", err) } approvalSubjects.Seal() agentDeps.ApprovalSubjects = approvalSubjects agentDeps.AgentRuntime = agentRuntime agentDeps.AgentOps = stores.AgentOps approvalHandler := approvalapp.NewHandler( stores.Approval, stores.App, approvalAppProjector, approvalSubjects, liveEmitter, ) approvalHandler.SetWorkflowRunDecisionSink(workflowApprovalDecisions) if err := internalAppHandlers.Register(approvalapp.HandlerKey, approvalHandler); err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("register Approval internal handler: %w", err) } internalAppHandlers.Seal() internalAppDispatcher := appruntime.NewInternalDispatcher(appruntime.InternalDispatcherOptions{ Store: stores.App, Registry: internalAppHandlers, Logger: logger, TargetAllowed: func(workspaceID string) bool { return owner.IsManaged(managedWorkspaces, workspaceID) }, WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(cfg.Server), }) approvalProjectionReconciler := approvalapp.NewProjectionReconciler( stores.Approval, approvalSubjects, managedWorkspaces, roleLogger.With(slog.String("component", "approval_projection")), ) agentAPI := agent.New(agentLogger, agentDeps, stores.Attachment, taskStore, daemonHub, deliveryFanout) if roleManagementRuntime != nil && roleReadGateway != nil { agentAPI.RoleManagers = roleManagementRuntime.manager agentAPI.RoleReads = roleReadGateway agentAPI.RoleRequests = roleManagementRuntime.service } if skillGateway != nil { agentAPI.SkillOwners = stores.App agentAPI.SkillWorkspace = workspaceResolver agentAPI.Skills = skillGateway } agentAPI.TeamWorkflowPlans = teamWorkflows llmProxySessions, err := llmProxySessionsForConfig(cfg, httpWrapper, stores.Agent, hostedAgentValidator, daemonHub) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("build agent llm proxy sessions: %w", err) } agentAPI.LLMProxySessions = llmProxySessions agentAPI.MinDaemonVersion = cfg.Daemon.MinDaemonVersion authOrganizations := authflow.OrganizationProvider(organizationService) authFlow := authflow.New(authflow.Dependencies{ RegistrationAllowlist: identityStore, WorkspaceInvites: stores.App, Organizations: authOrganizations, DefaultWorkspacePostCommit: defaultWorkspaceBillingProvisioner{ hosted: hostedClient, credit: billingMaaSClient, }, RequireDefaultWorkspacePostCommit: cfg.Billing.Enabled, }) workspaceSummaryAuthFlow := withWorkspaceSummaryInvalidatingAuthFlow( authFlow, workspaceSummaryInvalidator, workspaceSummaryOrganizations, liveEmitter, ) workspaceFlow := workspaceservice.New(workspaceservice.Dependencies{ Workspaces: stores.Workspace, WorkspaceAdmin: stores.App, Machines: stores.App, MachineTokens: stores.Agent, MiniappPairings: identityStore, }) workspaceSummaryWorkspaceFlow := withWorkspaceSummaryInvalidatingWorkspaceService(workspaceFlow, workspaceSummaryInvalidator) workspaceSummaryAuthStore := withWorkspaceSummaryInvalidatingAuthStore(identityStore, workspaceSummaryInvalidator) registrationPolicy := auth.NewRegistrationPolicy(inf.registrationMode, cfg.Auth.RegistrationAllowlistEmails) meshXPublicBaseURL := cfg.PublicBaseURL scheduleStore := schedulestore.NewWithOptions(inf.coreDataDB, schedulestore.StoreOptions{WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(cfg.Server)}) scheduleManageService := manage.New(scheduleLogger, inf.coreDataDB, stores.App, manage.ServiceOptions{}) triggerService := trigger.New(workflowLogger, inf.coreDataDB, stores.App, trigger.ServiceOptions{}) workflowPlanStore := planstore.New(inf.coreDataDB) workflowPlanService := planmanage.New( workflowPlanStore, planmanage.WithDraftUpdated(liveEmitter.EmitWorkflowPlanDraftUpdated), ) workflowAgentEditService := planmanage.NewAgentEditService( inf.coreDataDB, workflowPlanStore, stores.Message, stores.App, stores.AgentOps, ) agentAPI.WorkflowReplies = workflowAgentEditService agentAPI.WorkflowPlanAgentEditGuard = workflowAgentEditService workflowBindingStore := bindingstore.New(inf.coreDataDB) workflowBindingService := bindingmanage.New( workflowBindingStore, workflowPlanService, bindingmanage.WithAgentOwnerVerifier(stores.App), ) workflowInvocationStore := invocationstore.NewWithOptions( inf.coreDataDB, invocationstore.StoreOptions{WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(cfg.Server)}, ) workflowInvocationService := invocationmanage.New( workflowInvocationStore, invocationmanage.WithInputAuthority(stores.App), ) workflowRunService := runmanage.New(workflowRunStore) workflowCalendarService := calendar.New(calendarstore.New(inf.coreDataDB)) workflowRunCallbacks, err := orchestrator.NewCallbackService(workflowRunStore, orchestrator.CallbackServiceOptions{ ScopeAuthorizer: stores.App, TaskWatchMutator: stores.App, }) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("build Workflow Run callback service: %w", err) } workflowInvocationScheduleWorker, err := newWorkflowInvocationScheduleWorkerForConfig( cfg, scheduleLogger.With(slog.String("component", "workflow_invocation_worker")), workflowInvocationStore, ) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("build Workflow Invocation schedule worker: %w", err) } agentAPI.WorkflowPlans = workflowPlanService agentAPI.WorkflowBindings = workflowBindingService agentAPI.WorkflowBindingOwners = stores.App agentAPI.WorkflowBindingWorkspace = workspaceResolver agentAPI.WorkflowInvocations = workflowInvocationService agentAPI.WorkflowInvocationOwners = stores.App agentAPI.WorkflowInvocationWorkspace = workspaceResolver agentAPI.WorkflowRuns = workflowRunService agentAPI.WorkflowRecovery = workflowRunStore agentAPI.WorkflowRunCallbacks = workflowRunCallbacks serviceEventSubscriptionCoordinator := serviceevent.NewServiceEventSubscriptionCoordinator(stores.App) latestDaemonVersion := latestDaemonVersionProvider(cfg.Daemon, httpWrapper) agentAPI.ScheduleService = scheduleManageService agentAPI.TriggerService = triggerService scopeFanoutRunner := fanout.NewScopeFanoutRunnerWithOptions(inf.coreDataDB, deliveryFanout, messageLogger.With(slog.String("component", "scope_fanout")), fanout.ScopeFanoutRunnerOptions{ WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(cfg.Server), WakeStatusEmitter: liveEmitter, }) messagingService := messageservice.New(messageservice.Dependencies{ Sessions: identityStore, Workspaces: workspaceStore, Creation: stores.App, Queries: stores.App, Search: stores.Message, Logger: messageLogger, }) workflowRunMessageWriter := fanout.NewWorkflowRunMessageWriter(identityStore, stores.App, fanout.Dependencies{ HumanMessages: messagingService, Hub: daemonHub, StreamWriter: imStreamWriter, ReplyThreadRootPreloader: replyRootPreloader, ScopeFanoutMarker: stores.App, TaskScopeLifecycle: stores.Message, UnreadRuntime: unreadRuntime, UnreadDeltaSink: liveEmitter, Logger: workflowLogger, }).WithReceiptResolver(fanout.NewWorkflowRunMessageReceiptStore(inf.coreDataDB)). WithResponderResolver(stores.App) workflowRunApprovalAssignees, err := orchestrator.NewHumanApprovalAssigneeResolver(inf.coreDataDB) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("build Workflow Run approval assignee resolver: %w", err) } workflowRunWorker, err := newWorkflowRunWorkerForConfig( cfg, workflowLogger, workflowRunStore, workflowPlanStore, workflowInvocationStore, workflowRunMessageWriter, approvalAppProjector, workflowRunApprovalAssignees, workflowagentruntime.NewHubWorkingReader(workflowagentruntime.ServiceAdapter{Service: stores.Agent}, daemonHub), ) if err != nil { return router.Dependencies{}, wireSide{}, err } scheduleService := execschedule.New(scheduleLogger, scheduleStore, execschedule.ServiceOptions{ Writer: fanout.NewScheduleEffectWriter(scheduleStore, fanout.Dependencies{ HumanMessages: messagingService, Hub: daemonHub, StreamWriter: imStreamWriter, ReplyThreadRootPreloader: replyRootPreloader, ScopeFanoutMarker: stores.App, TaskScopeLifecycle: stores.Message, UnreadRuntime: unreadRuntime, UnreadDeltaSink: liveEmitter, Logger: scheduleLogger, }), }) assetAccessWorker, err := newMeshXAssetAccessWorkerForConfig(cfg, meshXBaseURL, meshXHTTPClient, logger, daemonHub, stores.Workspace) if err != nil { return router.Dependencies{}, wireSide{}, err } if strings.TrimSpace(apiForwarderDeps.BaseURL) != "" { agentThreadGateway, err := forwarder.NewAgentThreadGateway(apiForwarderDeps) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("agent thread gateway: %w", err) } agentAPI.AgentThreadGateway = agentThreadGateway } knowledgeQAWorker, knowledgeQAGateway, err := newKnowledgeQAWorkerForConfig( cfg, knowledgeLogger.With(slog.String("component", "knowledge_qa_worker")), inf, stores.KnowledgeQA, stores.App, deliveryFanout, knowledgeQATurnUpdates, apiForwarderDeps, ) if err != nil { return router.Dependencies{}, wireSide{}, err } var roleRealizer agents.RoleRealizer if ownsMeshXRuntime { roleRealizer = agents.NewMeshXRealizer(agents.MeshXRealizerOptions{ BaseURL: meshXBaseURL, HTTPClient: apiForwarderDeps.HTTPClient, TokenTTL: cfg.MeshX.ServiceTokenTTL, }) } var cloudMemoryProvisioner webagent.CloudMemoryProvisioner if ownsMeshXRuntime { cloudMemoryProvisioner = agentmemory.NewMeshXProvisioner(agentmemory.MeshXProvisionerOptions{ BaseURL: meshXBaseURL, HTTPClient: apiForwarderDeps.HTTPClient, TokenTTL: cfg.MeshX.ServiceTokenTTL, }) } roleAgentCreate := agents.NewService(stores.App, roleRealizer, agents.ServiceOptions{ BillingEnabled: cfg.Billing.Enabled, QuotaStore: stores.Plan, }) if ownsMeshXRuntime && roleAgentCreate == nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("role agent creation requires configured MeshX role realization") } if ownsMeshXRuntime && cloudMemoryProvisioner == nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("agent cloud memory creation requires configured MeshX memory provisioning") } var agentResourceCopier webagent.AgentResourceCopier if ownsMeshXRuntime && meshXBaseURL != "" { agentResourceCopier, err = webagent.NewForwardedAgentResourceCopier(apiForwarderDeps) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("agent resource copier: %w", err) } } operationAgentService := webagent.NewAgentService( stores.App, stores.Rewind, daemonHub, liveEmitter, agentLogger, stores.App, organizationService, hostedAgentValidator, agentResourceCopier, cfg.MeshX.Enabled && cfg.MeshX.KnowledgeBasesEnabled, stores.Agent, cloudMemoryProvisioner, ) operationExecutor := webagent.NewOperationExecutor( operationAgentService, webagent.WithOperationAgentInventory(stores.App), webagent.WithOperationOwnerIdentityFinder(stores.Identity), ) operationWorker, err := agentops.NewOperationWorker( stores.AgentOps, operationExecutor, approvalapp.NewAgentManagementTerminalObserver( stores.Approval, approvalAppProjector, ), 0, agentLogger.With(slog.String("component", "operation_worker")), agentops.WithOperationWorkspaceGroupID(schedulerWorkspaceGroupIDForConfig(cfg.Server)), ) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "build Agent operation worker: %w", err, ) } extensionRegistry, err := extensionRegistryForProfile( profile, providerDeps, ) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf("bootstrap extension registry: %w", err) } // Catalog registry + service is reused for /runtime/ext runtime Access and // App selection/authorization catalog hooks. Workspace owns the hook but not // the runtime routes, so it loads only the App-hosted Builtins. catalogHookRegistry := extensionRegistry if catalogHookRegistry == nil && profile == router.ProfileWorkspace { catalogHookRegistry, err = buildExtensionRegistry(providerDeps) if err != nil { return router.Dependencies{}, wireSide{}, fmt.Errorf( "build App-hosted Builtin registry: %w", err, ) } } var catalogService *extensionflow.Service if catalogHookRegistry != nil { var credentialKey []byte if profile == router.ProfileAll || profile == router.ProfileExtension { credentialKey = inf.appCredentialKey } catalogService = extensionflow.New( inf.coreDataDB, catalogHookRegistry, credentialKey, extensionflow.Options{}, ) mailAuthorizer.Bind(catalogService) } // The access adapter delegates to the shared catalog projection (extension + // app-kind/default-allow via the Builtin registry) so all authorized surfaces // appear in /runtime/ext. (stage-6 final-review blocker 1b.) extensionAccessAdapter := newExtensionAccessAdapter(inf.coreDataDB, catalogService) // Wire the app Enable/Disable → Extension Catalog same-tx hook (stage-6 // final-review blocker 2a). Built only when this profile bootstraps the // Extension registry; the hook recomputes + applies the catalog digest within // the app authorization transaction so an app enable/disable advances the // catalog atomically. // Catalog hook + backfill: wired for all|workspace (profiles with App // Enable/Disable routes), independent of the /runtime/ext registry gate. Reuse // the bootstrap registry for all|extension; build one for workspace (the // /runtime/ext gate leaves workspace nil, but the catalog hook needs the Builtin // registrations for the app-kind digest). stage-6 final-review blocker 2a. if catalogService != nil && (profile == router.ProfileAll || profile == router.ProfileWorkspace) { stores.App.SetAgentAppCatalogChangeHook(func(ctx context.Context, tx coredb.Tx, workspaceID, agentID string, now time.Time) error { return catalogService.ApplyAgentCatalogChange(ctx, tx, workspaceID, agentID, now) }) stores.App.SetWorkspaceAppCatalogReconcileHook(func(ctx context.Context, tx coredb.Tx, workspaceID string, now time.Time) error { return catalogService.EnqueueWorkspaceCatalogReconcile(ctx, tx, workspaceID, now) }) // Drain coalesced Workspace dirty jobs. Idle ticks touch only the // reconcile-job table; source mutations enqueue the affected Workspace and // the worker processes Agents in bounded pages with digest no-op writes. catalogService.StartWorkspaceCatalogReconcileWorker(ctx, 5*time.Second, func(err error) { logger.Warn("extension catalog reconciliation failed", "error", err) }) } routerDeps := router.Dependencies{ Logger: logger, Health: router.HealthDependencies{ CoreDB: inf.coreDBConn, CoreDBAdmin: inf.coreDBAdmin, DependencyCheckTimeout: cfg.HTTP.EffectiveReadyzDependencyTimeout(), }, Agent: agentAPI, ManagedWorkspaces: managedWorkspaces, Web: router.WebDependencies{ // The extension profile is an API runtime and must not serve the web // SPA dist even if AGENTTEAM_WEB_SERVE_ENABLED is set (spec §6.1). ServeWeb: cfg.Web.ServeWeb && profile != router.ProfileExtension, DistDir: cfg.Web.DistDir, }, Extension: router.ExtensionDependencies{ Registry: extensionRegistry, Access: extensionAccessAdapter, Snapshot: extensionAccessAdapter, Invoker: catalogService, }, Daemon: router.DaemonDependencies{ Hub: daemonHub, Auth: stores.Agent, SessionIssuer: identityStore, LatestDesktopVersion: latestDaemonVersion, MinDesktopVersion: cfg.Daemon.MinDaemonVersion, }, API: router.APIDependencies{ Onboarding: apiOnboarding.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceResolver, Store: stores.Onboarding, MockExchangeStore: stores.Onboarding, Agents: stores.App, Runtime: daemonHub, MockMessages: stores.App, MockAttachments: stores.Attachment, MockFanout: deliveryFanout, BillingEnabled: cfg.Billing.Enabled, EmailVerificationRequired: emailVerificationRequired, }, KnowledgeQA: knowledgeqaapi.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceResolver, Sessions: knowledgeqaapi.NewConversationService(stores.KnowledgeQA), Policies: stores.KnowledgeQA, Organizations: knowledgeqaapi.NewOrganizationResolver(workspaceSummaryOrganizations), HostedAvailability: hostedAvailabilityClient, Queue: knowledgeQAWorker, TurnUpdates: knowledgeQATurnUpdates, Citations: knowledgeQAGateway, Feedbacks: knowledgeqaapi.NewFeedbackService(stores.Reporting), Sharing: knowledgeqaapi.NewSharingService(stores.KnowledgeQA), PublicBaseURL: cfg.PublicBaseURL, EmailVerificationRequired: emailVerificationRequired, }, Admin: admin.Dependencies{ AuthStore: identityStore, AdminStore: stores.Admin, RegistrationAllowlist: identityStore, HostedCreditGrants: adminHostedCreditGrantClient{next: billingMaaSClient}, DatabasePool: inf.coreDBPool, WorkspacePlacementAdmin: stores.Workspace, PlatformServiceAccounts: platformServiceAccounts, InternalAdminToken: cfg.Server.InternalAdminToken, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, Workspace: workspace.Dependencies{ AuthStore: identityStore, WorkspaceService: workspaceSummaryWorkspaceFlow, PlanStore: stores.Plan, Organizations: workspaceSummaryOrganizations, HostedAvailability: hostedAvailabilityClient, HostedPricing: hostedPricingClient, Mailer: inf.mailService, PublicBaseURL: cfg.PublicBaseURL, EmailVerificationRequired: emailVerificationRequired, MachineRuntime: daemonHub, Realtime: liveEmitter, LatestDaemonVersion: latestDaemonVersion, Logger: workspaceLogger, ManagedWorkspaces: managedWorkspaces, UnreadRuntime: unreadRuntime, MeshXEnabled: cfg.MeshX.Enabled, }, Org: org.Dependencies{ Store: identityStore, Organizations: workspaceSummaryOrganizations, PlanStore: stores.Plan, Runtime: daemonHub, LLMRegistry: agentteamLLMRefresher, HostedAvailability: hostedAvailabilityClient, KnowledgeQAPolicies: stores.KnowledgeQA, Mailer: inf.mailService, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, Auth: auth.Dependencies{ Store: workspaceSummaryAuthStore, AuthFlow: workspaceSummaryAuthFlow, Realtime: liveEmitter, Organizations: organizationService, Mailer: inf.mailService, MobileVerificationSender: inf.mobileVerificationSender, EmailVerificationRequired: emailVerificationRequired, RegistrationPolicy: registrationPolicy, VerificationTokenTTL: cfg.Auth.VerificationTokenTTL, SessionTTL: cfg.Auth.SessionTTL, PasswordAuthDisabled: cfg.Auth.PasswordAuthDisabled, PingCAPAuthConfig: auth.PingCAPAuthConfig{ Enabled: cfg.PingCAPAuth.Enabled, Issuer: cfg.PingCAPAuth.Issuer, ClientID: cfg.PingCAPAuth.ClientID, ClientSecret: cfg.PingCAPAuth.ClientSecret, WebClientID: cfg.PingCAPAuth.WebClientID, WebClientSecret: cfg.PingCAPAuth.WebClientSecret, DesktopClientID: cfg.PingCAPAuth.DesktopClientID, DesktopClientSecret: cfg.PingCAPAuth.DesktopClientSecret, Scopes: []string{"openid", "profile", "email", "phone"}, }, HTTPClient: newOutboundHTTPClient(httpWrapper, "oidc", 30*time.Second, nil), Logger: authLogger, }, Billing: billing.Dependencies{ SessionAuth: identityStore, Organizations: organizationService, BillingClient: billingMaaSClient, SeatBilling: billing.NewSeatBillingStoreAdapter(stores.SeatBilling), Alipay: billingAlipayClient, Enabled: cfg.Billing.Enabled, PublicBaseURL: cfg.PublicBaseURL, ReturnBaseURL: cfg.Billing.ReturnBaseURL, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, Public: public.Dependencies{ AuthStore: identityStore, MessageSearchStore: stores.Message, ContentReports: stores.Reporting, MigrationInspector: inf.coreDBAdmin, CoreDB: inf.coreDBConn, Environment: cfg.Environment, PingCAPAuthEnabled: cfg.PingCAPAuth.Enabled, PasswordAuthDisabled: cfg.Auth.PasswordAuthDisabled, OpenRegistration: openRegistration(inf.registrationMode), RegistrationMode: inf.registrationMode, TeamsEnabled: cfg.Teams.Enabled, AgentToolsSettingsEnabled: cfg.AgentTools.SettingsEnabled, OrganizationDirectoryEnabled: cfg.Organization.DirectoryEnabled, OrganizationBYOKEnabled: cfg.Organization.BYOKEnabled, BillingEnabled: cfg.Billing.Enabled, DaemonConnectBaseURL: cfg.Daemon.ConnectBaseURLOrDefault(cfg.PublicBaseURL), MeshXEnabled: cfg.MeshX.Enabled, MeshXKnowledgeBasesEnabled: cfg.MeshX.Enabled && cfg.MeshX.KnowledgeBasesEnabled, MeshXWorkflowsEnabled: cfg.MeshX.Enabled && cfg.MeshX.WorkflowsEnabled, MeshXMiniappPairingEnabled: cfg.MeshX.Enabled && cfg.MeshX.MiniappPairingEnabled, MeshXRolesEnabled: cfg.MeshX.Enabled, MeshXBaseURL: meshXPublicBaseURL, }, Media: media.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, ControlWorkspaceStore: cachedWorkspaceResolver, AgentTokenStore: stores.Agent, AgentStore: stores.App, AttachmentStore: stores.Attachment, PersonalStickerStore: stores.Attachment, EmailVerificationRequired: emailVerificationRequired, }, IM: im.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, Logger: messageLogger, Messages: messagingService, MessageQueries: messagingService, Channels: stores.Channel, ChannelAccess: stores.Channel, ChannelMembers: stores.App, MessageSearch: stores.Message, HumanMessages: messagingService, ScopeSnapshotTaskBoard: im.NewScopeSnapshotTaskBoard(taskStore), AgentDeliveries: stores.App, AgentNotifier: daemonHub, ReplyThreadRootPreloader: replyRootPreloader, ScopeFanoutMarker: stores.App, TaskScopeLifecycle: stores.Message, RewindStore: stores.Rewind, RewindHub: daemonHub, DMs: stores.App, OwnerContexts: stores.App, ChannelNotifications: stores.PrivateAgents, ThreadNotifications: stores.PrivateAgents, Conversations: stores.Message, ConversationSidebarThreads: stores.Message, Reactions: stores.Reaction, Pins: stores.Message, StreamEvents: stores.IM, VisibleScopes: message.NewSQLVisibleScopeAuthorizer(inf.coreDataDB), StreamWriter: imStreamWriter, StreamSubscriptions: imStreamHub, StreamCursors: stores.IM, ConversationIndex: stores.IM, UnreadRuntime: unreadRuntime, MessageUnreadRuntime: unreadRuntime, UnreadDeltaSink: liveEmitter, RuntimeBootstrap: daemonHub, LiveEmitter: liveEmitter, ScopeMembershipSync: imStreamHub, }, Channel: channel.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, Channels: stores.Channel, AppStore: stores.App, ChannelPermissions: stores.Channel, ThreadStore: stores.Message, ThreadSubsystemChannels: stores.App, ThreadReviewStore: stores.Review, EmailVerificationRequired: emailVerificationRequired, Realtime: liveEmitter, MessageFanout: deliveryFanout, UnreadRuntime: unreadRuntime, Hub: daemonHub, Logger: channelLogger, }, APIForwarder: apiForwarderDeps, Task: task.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, TaskBoard: taskStore, ReviewStore: stores.Review, Store: task.NewTaskParityStoreAdapter(stores.App, stores.PrivateAgents), MessageFanout: deliveryFanout, UnreadRuntime: unreadRuntime, ConversationReadProjector: im.NewTaskConversationProjector(stores.Message, imStreamWriter, unreadRuntime, liveEmitter), ConversationPolicyProjector: im.NewTaskConversationProjector(stores.Message, imStreamWriter, unreadRuntime, liveEmitter), CollaboratorRequests: collaboratorRequestService, Logger: taskLogger, }, Team: team.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, Teams: stores.Team, EmailVerificationRequired: emailVerificationRequired, }, TeamWorkflow: teamworkflowapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, Service: teamWorkflowBrowserService, }, Schedule: scheduleapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, ScheduleService: scheduleManageService, }, ServiceAccount: serviceaccountapi.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceResolver, Service: serviceAccountFlow, PendingRequestProjector: approvalAppProjector, EmailVerificationRequired: emailVerificationRequired, }, Trigger: triggerapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, TriggerService: triggerService, }, WorkflowBinding: workflowbindingapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, Service: workflowBindingService, }, WorkflowInvocation: workflowinvocationapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, Service: workflowInvocationService, }, WorkflowPlan: workflowplanapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, Service: workflowPlanService, AgentEdit: workflowAgentEditService, DirectPublishGuard: teamWorkflows, }, WorkflowRun: run.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceResolver, EmailVerificationRequired: emailVerificationRequired, Service: workflowRunService, Calendar: workflowCalendarService, }, App: app.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceResolver, AppStore: stores.App, AppRuntimeStore: stores.App, AppRunnerService: appRunnerService, SearchEngineConnections: stores.App, SearchEngineTester: searchEngineService, ExtensionProviders: extensionProviderIndex, FollowupDeliverer: deliveryFanout, ServiceEventSubscriptionBlocker: serviceEventSubscriptionCoordinator, InstallationRequestProjector: approvalAppProjector, Realtime: liveEmitter, EmailVerificationRequired: emailVerificationRequired, Logger: logger, ManagedWorkspaces: managedWorkspaces, }, Mail: mailapi.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceResolver, Service: mailConnectionService, Realtime: liveEmitter, EmailVerificationRequired: emailVerificationRequired, }, Provider: provider.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceResolver, Store: stores.Provider, GitHubAdminStore: stores.App, GiteaAdminStore: stores.App, InstallationRequestProjector: approvalAppProjector, GitHubConfig: provider.GitHubConfig{ FlowStateKey: inf.feishuCredentialKey, BaseURL: githubBaseURL, APIBaseURL: githubAPIBaseURL, }, ServiceEventSubscriptionBlocker: serviceEventSubscriptionCoordinator, HTTPClient: providerHTTPClient, PublicBaseURL: cfg.PublicBaseURL, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, LLM: llm.Dependencies{ APRDispatcher: agentteamLLMRefresher.APRDispatcher(), Logger: logger, ProxyKeyAuthenticator: organizationService, UsageTracker: organizationService, ManagedWorkspaces: managedWorkspaces, }, WebAgent: webagent.Dependencies{ AgentStore: stores.App, RoleRealizer: roleRealizer, RoleAgentCreate: roleAgentCreate, CloudMemory: cloudMemoryProvisioner, ModelProfiles: stores.Agent, SessionAuth: stores.Identity, WorkspaceStore: workspaceResolver, ReviewStore: stores.Review, ProjectStore: stores.Review, RewindStore: stores.Rewind, RuntimeCommands: agentRuntime, Organizations: organizationService, HostedAgents: hostedAgentValidator, ResourceCopy: agentResourceCopier, KnowledgeBasesEnabled: ownsMeshXRuntime && cfg.MeshX.KnowledgeBasesEnabled, Runtime: daemonHub, Realtime: liveEmitter, PublicBaseURL: cfg.PublicBaseURL, EmailVerificationRequired: emailVerificationRequired, Logger: agentLogger, }, }, LiveEmitter: liveEmitter, } return routerDeps, wireSide{ scheduleService: scheduleService, workflowInvocationSchedules: workflowInvocationScheduleWorker, scopeFanoutRunner: scopeFanoutRunner, unreadRuntime: unreadRuntime, assetAccessWorker: assetAccessWorker, workflowRunWorker: workflowRunWorker, internalAppDispatcher: internalAppDispatcher, approvalProjectionReconciler: approvalProjectionReconciler, platformAccounts: platformServiceAccounts, knowledgeQAWorker: knowledgeQAWorker, operationWorker: operationWorker, roleManagementWorker: roleManagementSide.roleManagementWorker, wikiGenerationApprovals: wikiGenerationApprovals, wikiGenerationTerminals: wikiGenerationTerminals, publicationApprovals: publicationApprovals, closers: []func(){agentteamLLMRefresher.Close}, }, nil } func effectiveMeshXEndpoint(cfg config.Config, wrapper httpClientWrapper, profile router.Profile, embedded http.RoundTripper) (string, *http.Client) { client := newOutboundHTTPClient(wrapper, "api.forwarder", apiForwarderTimeout(cfg.MeshX.RequestTimeout), nil) if embedded != nil { client.Transport = embedded return "http://embedded.meshx", client } return "", nil } func newKnowledgeQAWorkerForConfig( cfg config.Config, logger *slog.Logger, inf *infra, knowledgeQAStore *knowledgeqa.Store, messages knowledgeqaservice.WorkerMessages, messageFanout knowledgeqaservice.WorkerMessageFanout, turnUpdates knowledgeqaservice.TurnUpdateNotifier, forwarderDeps forwarder.Dependencies, ) (*knowledgeqaservice.Worker, *forwarder.KnowledgeQARuntimeGateway, error) { if !cfg.MeshX.Enabled || !cfg.MeshX.KnowledgeBasesEnabled || strings.TrimSpace(forwarderDeps.BaseURL) == "" || strings.TrimSpace(cfg.LLM.MaaSCapabilityPrivateKeyFile) == "" { return nil, nil, nil } runtimeForwarderDeps := forwarderDeps runtimeForwarderDeps.Logger = logger if forwarderDeps.HTTPClient != nil { client := *forwarderDeps.HTTPClient client.Timeout = knowledgeqaservice.DefaultTurnExecutionTimeout runtimeForwarderDeps.HTTPClient = &client } runtimeGateway, err := forwarder.NewKnowledgeQARuntimeGateway(runtimeForwarderDeps) if err != nil { return nil, nil, fmt.Errorf("build Knowledge Q&A MeshX runtime gateway: %w", err) } signer, err := capability.LoadSignerFromFile(cfg.LLM.MaaSCapabilityPrivateKeyFile) if err != nil { return nil, nil, fmt.Errorf("load Knowledge Q&A MaaS capability signer: %w", err) } worker, err := knowledgeqaservice.NewWorker(knowledgeqaservice.WorkerOptions{ Store: knowledgeQAStore, Policies: knowledgeQAStore, Runtime: runtimeGateway, Signer: signer, Provisioner: knowledgeqaservice.NewProvisioner(inf.coreDataDB), Messages: messages, MessageFanout: messageFanout, TurnUpdates: turnUpdates, Logger: logger, }) if err != nil { return nil, nil, err } return worker, runtimeGateway, nil } func newWorkflowInvocationScheduleWorkerForConfig( cfg config.Config, logger *slog.Logger, materializer invocation.ScheduledMaterializer, ) (*invocation.ScheduleWorker, error) { if err := cfg.Server.Validate(); err != nil { return nil, err } if !profileOwnsWorkspaceDataPlane(cfg.Server) { return nil, nil } dispatcher, err := invocation.NewScheduleDispatcher(materializer, invocation.ScheduleDispatcherOptions{}) if err != nil { return nil, err } return invocation.NewScheduleWorker(dispatcher, invocation.ScheduleWorkerOptions{ Logger: logger, }), nil } // newWorkflowRunWorkerForConfig composes the first-class Run reducer and its // post-transaction effect dispatcher. The central profile owns neither Run // leases nor external effects; workspace/all profiles use the same placement // fence as Invocation materialization. func newWorkflowRunWorkerForConfig( cfg config.Config, logger *slog.Logger, repository *runstore.Store, plans *planstore.Store, invocations *invocationstore.Store, messages orchestrator.MessageWriter, approvals orchestrator.ApprovalWriter, assignees orchestrator.ApprovalAssigneeResolver, runtimeWorking orchestrator.RuntimeWorkingReader, ) (*orchestrator.Worker, error) { if err := cfg.Server.Validate(); err != nil { return nil, err } if !profileOwnsWorkspaceDataPlane(cfg.Server) { return nil, nil } if repository == nil { return nil, fmt.Errorf("build Workflow Run worker: Run store is required") } if err := repository.Validate(); err != nil { return nil, fmt.Errorf("build Workflow Run worker: %w", err) } if plans == nil || invocations == nil { return nil, fmt.Errorf("build Workflow Run worker: PlanVersion and Invocation stores are required") } if messages == nil || approvals == nil || assignees == nil { return nil, fmt.Errorf("build Workflow Run worker: message, approval, and assignee adapters are required") } materials, err := orchestrator.NewPinnedMaterialLoader(plans, invocations) if err != nil { return nil, fmt.Errorf("build Workflow Run material loader: %w", err) } instanceID := uuid.NewString() reducer, err := orchestrator.NewEngine(repository, materials, orchestrator.EngineOptions{ WorkerID: "workflow-run-reducer:" + instanceID, }) if err != nil { return nil, fmt.Errorf("build Workflow Run reducer: %w", err) } effects, err := orchestrator.NewEffectDispatcher( repository, messages, approvals, assignees, orchestrator.EffectDispatcherOptions{ WorkerID: "workflow-run-effects:" + instanceID, WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(cfg.Server), Responders: workflowRunCallbackResponders(messages), }, ) if err != nil { return nil, fmt.Errorf("build Workflow Run effect dispatcher: %w", err) } // The automatic recovery runner shares the same store, message writer, and // real clock as the reducer/effect dispatcher, and runs inside the worker's // tick (RecoverOnce per pass) so the production lifecycle owns exactly one // goroutine with no duplicate runner or shutdown race. It never touches the // execution lease or the callback queue; it only nudges frozen awaiting // callbacks whose canonical Agent runtime working-state has lapsed (P0 #1/#2). // The working-state judgment comes from the same daemon/Hub overlay that drives Web Agent Activity, resolved per responder. workflowRunRecovery, err := orchestrator.NewRecoveryRunner( repository, messages, orchestrator.RecoveryRunnerOptions{ WorkerID: "workflow-run-recovery:" + instanceID, WorkspaceGroupID: schedulerWorkspaceGroupIDForConfig(cfg.Server), Logger: logger.With(slog.String("component", "recovery_runner")), RuntimeWorking: runtimeWorking, }, ) if err != nil { return nil, fmt.Errorf("build Workflow Run automatic recovery runner: %w", err) } return orchestrator.NewWorker(reducer, effects, orchestrator.WorkerOptions{ Logger: logger.With(slog.String("component", "run_worker")), Reconciler: effects, Recovery: workflowRunRecovery, ArtifactMaintenance: repository, }), nil } // workflowRunCallbackResponders returns the trusted pre-delivery callback // responder resolver when the configured message writer implements it. The // resolver is optional so non-message writer adapters degrade gracefully. func workflowRunCallbackResponders(messages orchestrator.MessageWriter) orchestrator.CallbackResponderResolver { if messages == nil { return nil } resolver, ok := messages.(orchestrator.CallbackResponderResolver) if !ok { return nil } return resolver } type meshXAssetAccessWorkspaceLister interface { ListRoutableWorkspaceIDsForGroup(ctx context.Context, workspaceGroupID string) ([]string, error) } func newMeshXAssetAccessWorkerForConfig(cfg config.Config, baseURL string, httpClient *http.Client, logger *slog.Logger, notifier meshxassets.Notifier, lister meshXAssetAccessWorkspaceLister) (*meshxassets.Worker, error) { if !cfg.MeshX.Enabled || strings.TrimSpace(baseURL) == "" || !profileOwnsWorkspaceDataPlane(cfg.Server) { return nil, nil } return meshxassets.NewWorker(meshxassets.Options{ BaseURL: baseURL, HTTPClient: httpClient, Notifier: notifier, WorkspaceIDs: meshXAssetAccessWorkspaceIDs(cfg.Server, lister), Logger: logger, RequestTTL: apiForwarderTimeout(cfg.MeshX.RequestTimeout), TokenTTL: cfg.MeshX.ServiceTokenTTL, }) } func meshXAssetAccessWorkspaceIDs(cfg config.ServerConfig, lister meshXAssetAccessWorkspaceLister) meshxassets.WorkspaceIDSource { if cfg.ProfileOrDefault() != config.ServerProfileWorkspace { return nil } return func(ctx context.Context) ([]string, error) { if lister == nil { return nil, fmt.Errorf("meshx asset access workspace lister is not configured") } return lister.ListRoutableWorkspaceIDsForGroup(ctx, strings.TrimSpace(cfg.WorkspaceGroupID)) } } func newUnreadRuntimeForConfig(cfg config.Config, store unread.SnapshotStore, logger *slog.Logger) (*unread.Service, error) { if !profileOwnsWorkspaceDataPlane(cfg.Server) { return nil, nil } runtime, err := unread.New(unread.Config{ SnapshotFlushInterval: cfg.UnreadRuntime.SnapshotFlushInterval, SnapshotFlushBatchSize: cfg.UnreadRuntime.SnapshotFlushBatchSize, HydrateMaxConcurrency: cfg.UnreadRuntime.HydrateMaxConcurrency, HydrateBatchSize: cfg.UnreadRuntime.HydrateBatchSize, }) if err != nil { return nil, fmt.Errorf("initialize unread runtime: %w", err) } service, err := unread.NewService(runtime, store, logger) if err != nil { return nil, fmt.Errorf("initialize unread runtime service: %w", err) } return service, nil } func daemonConnectBaseURL(cfg config.Config) string { return cfg.Daemon.ConnectBaseURLOrDefault(cfg.PublicBaseURL) } func latestDaemonVersionProvider(cfg config.DaemonConfig, httpWrapper httpClientWrapper) versioning.Provider { if override := strings.TrimSpace(cfg.LatestVersionOverride); override != "" { return versioning.NewStaticProvider(override) } return versioning.NewStableManifestProvider(versioning.StableManifestProviderConfig{ StableURL: cfg.LatestVersionURL, FallbackURL: versioning.DefaultFallbackStableManifestURL, CacheTTL: cfg.LatestVersionCacheTTL, HTTPClient: newOutboundHTTPClient(httpWrapper, "versioning.daemon", 10*time.Second, nil), }) } func observabilityDaemonTransportMetrics(wrapper httpClientWrapper) daemon.DaemonTransportMetrics { provider, ok := wrapper.(daemonTransportMetricsProvider) if !ok || provider == nil { return nil } return provider.DaemonTransportMetrics() } func llmProxyTransporter(cfg config.Config) (types.HTTPProxy, error) { proxyURL, err := cfg.LLM.OutboundProxyURL() if err != nil { return nil, err } if proxyURL == nil { return nil, nil } return types.NewURLTransporter(proxyURL), nil } func llmProxySessionsForConfig(cfg config.Config, httpWrapper httpClientWrapper, scopes agent.MaaSLLMProxyScopeStore, hostedAgents agent.HostedAgentValidator, runtimeSessions agent.RuntimeSessionLookup) (agent.LLMProxySessionStore, error) { if !cfg.LLM.MaaSDirectConfigured() { return nil, nil } signer, err := capability.LoadSignerFromFile(cfg.LLM.MaaSCapabilityPrivateKeyFile) if err != nil { return nil, err } projectionClient := capability.NewProjectionClient( cfg.LLM.MaaSBaseURL, cfg.LLM.MaaSInternalBearerToken, newOutboundHTTPClient(httpWrapper, "maas.control", 10*time.Second, nil), ) return agent.NewMaaSLLMProxySessionService(agent.MaaSLLMProxySessionConfig{ Scopes: scopes, HostedAgents: hostedAgents, RuntimeSessions: runtimeSessions, Signer: signer, ProjectionPush: projectionClient, RuntimeBaseURL: cfg.LLM.MaaSRuntimeBaseURL, }), nil } func openRegistration(registrationMode string) bool { return strings.ToLower(strings.TrimSpace(registrationMode)) != auth.RegistrationModeInviteOnly } func apiForwarderTimeout(configured time.Duration) time.Duration { if configured <= 0 { return 30 * time.Second } return configured }