package main import ( "context" "database/sql" "errors" "flag" "fmt" "log/slog" "net" "net/http" "os" "os/signal" "path/filepath" "strings" "syscall" "time" "github.com/go-sql-driver/mysql" "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/channel" "github.com/tidbcloud/AgentTeam/internal/api/daemon" "github.com/tidbcloud/AgentTeam/internal/api/im" "github.com/tidbcloud/AgentTeam/internal/api/im/fanout" "github.com/tidbcloud/AgentTeam/internal/api/llm" "github.com/tidbcloud/AgentTeam/internal/api/media" "github.com/tidbcloud/AgentTeam/internal/api/org" "github.com/tidbcloud/AgentTeam/internal/api/provider" "github.com/tidbcloud/AgentTeam/internal/api/public" "github.com/tidbcloud/AgentTeam/internal/api/schedule" "github.com/tidbcloud/AgentTeam/internal/api/serviceaccount" "github.com/tidbcloud/AgentTeam/internal/api/task" "github.com/tidbcloud/AgentTeam/internal/api/team" "github.com/tidbcloud/AgentTeam/internal/api/teamworkflow" apitrigger "github.com/tidbcloud/AgentTeam/internal/api/trigger" "github.com/tidbcloud/AgentTeam/internal/api/webagent" bindingapi "github.com/tidbcloud/AgentTeam/internal/api/workflow/binding" invocationapi "github.com/tidbcloud/AgentTeam/internal/api/workflow/invocation" planapi "github.com/tidbcloud/AgentTeam/internal/api/workflow/plan" runapi "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/archive" "github.com/tidbcloud/AgentTeam/internal/config" "github.com/tidbcloud/AgentTeam/internal/coredb" "github.com/tidbcloud/AgentTeam/internal/coredb/testkit" "github.com/tidbcloud/AgentTeam/internal/extension/registry" "github.com/tidbcloud/AgentTeam/internal/mailer" "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/approvalapp" "github.com/tidbcloud/AgentTeam/internal/service/appruntime" "github.com/tidbcloud/AgentTeam/internal/service/authflow" "github.com/tidbcloud/AgentTeam/internal/service/collaborator" "github.com/tidbcloud/AgentTeam/internal/service/extensionflow" messageservice "github.com/tidbcloud/AgentTeam/internal/service/message" organizationcore "github.com/tidbcloud/AgentTeam/internal/service/organization" serviceaccountservice "github.com/tidbcloud/AgentTeam/internal/service/serviceaccount" "github.com/tidbcloud/AgentTeam/internal/service/serviceevent" "github.com/tidbcloud/AgentTeam/internal/service/team/roleproposal" serviceteamworkflow "github.com/tidbcloud/AgentTeam/internal/service/teamworkflow" workspaceservice "github.com/tidbcloud/AgentTeam/internal/service/workspace" "github.com/tidbcloud/AgentTeam/internal/sms" "github.com/tidbcloud/AgentTeam/internal/store" "github.com/tidbcloud/AgentTeam/internal/store/message" "github.com/tidbcloud/AgentTeam/internal/store/organization" "github.com/tidbcloud/AgentTeam/internal/store/plan" schedulestore "github.com/tidbcloud/AgentTeam/internal/store/schedule" "github.com/tidbcloud/AgentTeam/internal/store/taskboard" "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" "github.com/tidbcloud/AgentTeam/internal/store/workflow/run" "github.com/tidbcloud/AgentTeam/internal/trigger" "github.com/tidbcloud/AgentTeam/internal/versioning" workflowagentruntime "github.com/tidbcloud/AgentTeam/internal/workflow/agentruntime" "github.com/tidbcloud/AgentTeam/internal/workflow/bindingmanage" "github.com/tidbcloud/AgentTeam/internal/workflow/calendar" "github.com/tidbcloud/AgentTeam/internal/workflow/invocation" "github.com/tidbcloud/AgentTeam/internal/workflow/invocationmanage" "github.com/tidbcloud/AgentTeam/internal/workflow/orchestrator" "github.com/tidbcloud/AgentTeam/internal/workflow/planmanage" "github.com/tidbcloud/AgentTeam/internal/workflow/runmanage" "github.com/tidbcloud/AgentTeam/internal/workspace/owner" ) var ( e2eFeishuCredentialEncryptionKey = []byte("0123456789abcdef0123456789abcdef") e2eAppCredentialStorageKey = []byte("0123456789abcdef0123456789abcdef") ) func e2eStoreOptions(publicBaseURL string) store.Options { return store.Options{ AppCredentialStorageKey: e2eAppCredentialStorageKey, PublicBaseURL: publicBaseURL, CheckWorkspaceQuotaInTx: func(context.Context, coredb.Tx, string, plan.MetricKey, int64) error { return nil }, CheckWorkspaceAgentOwnerQuotaInTx: func(context.Context, coredb.Tx, string, string, int64) error { return nil }, } } func e2eExtensionRouteDependencies(db coredb.DB) router.ExtensionDependencies { extensionRegistry := registry.New() return router.ExtensionDependencies{ Registry: extensionRegistry, Invoker: extensionflow.New( db, extensionRegistry, e2eAppCredentialStorageKey, extensionflow.Options{}, ), } } func main() { var httpAddress string flag.StringVar(&httpAddress, "http-address", "127.0.0.1:0", "HTTP listen address (host:port)") flag.Parse() ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer stop() logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelWarn})) listener, err := net.Listen("tcp", httpAddress) if err != nil { logger.Error("listen", slog.Any("error", err)) os.Exit(1) } defer listener.Close() baseURL := "http://" + listener.Addr().String() tidbRuntime, err := startTiDB(ctx, logger) if err != nil { logger.Error("start tidb", slog.Any("error", err)) os.Exit(1) } defer tidbRuntime.Close() pool, err := sql.Open("mysql", tidbRuntime.DatabaseDSN) if err != nil { logger.Error("connect tidb", slog.Any("error", err)) os.Exit(1) } defer func() { _ = pool.Close() }() corePool := coredb.WrapSQLDB(pool) pingCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() if err := pool.PingContext(pingCtx); err != nil { logger.Error("ping tidb", slog.Any("error", err)) os.Exit(1) } coreDBAdmin := coredb.NewAdmin() adminDB, err := coreDBAdmin.Open(ctx, tidbRuntime.DatabaseDSN, 5*time.Second) if err != nil { logger.Error("open tidb admin", slog.Any("error", err)) os.Exit(1) } defer adminDB.Close() if err := coreDBAdmin.ApplyMigrations(ctx, adminDB); err != nil { logger.Error("apply migrations", slog.Any("error", err)) os.Exit(1) } archiveRoot := filepath.Join(tidbRuntime.RuntimeDir, "archive") archiveStorage, err := archive.NewFilesystem(archiveRoot) if err != nil { logger.Error("init archive storage", slog.Any("error", err)) os.Exit(1) } stores := store.NewWithOptions(corePool, archiveStorage, nil, 30, e2eStoreOptions(baseURL)) appRunnerService := appruntime.NewRunnerService(appruntime.RunnerServiceOptions{Store: stores.ServiceActor, Authority: stores.App}) identityStore := stores.Identity workspaceStore := stores.Workspace if _, err := stores.Agent.MarkMachinesDisconnectedOnStartup(ctx); err != nil { logger.Error("reconcile machine connections", slog.Any("error", err)) os.Exit(1) } mailbox := newE2EMailbox() mailService := newE2EMailboxMailer(mailer.NewNoop(logger, baseURL), mailbox) mobileVerificationRecorder := newE2EMobileVerificationRecorder() mobileVerificationSender := newE2EMobileVerificationSender(sms.NewNoop(logger), mobileVerificationRecorder) organizationCore := organizationcore.NewCoreBridge(e2eOrganizationCoreDependencies(stores)) organizationService := organization.NewWithDBOptions(corePool, organizationCore, organization.ServiceOptions{ FeishuCredentialEncryptionKey: e2eFeishuCredentialEncryptionKey, CheckOrganizationQuotaInTx: e2eCheckOrganizationQuotaInTx, }) agentRuntime := agentruntime.New(agentruntime.Dependencies{ Agents: stores.Agent, Messages: stores.Message, Recovery: stores.Rewind, WebAgents: stores.App, WebRecovery: stores.Rewind, IsRuntimeOffline: func(err error) bool { return errors.Is(err, runtimecontrol.ErrMachineOffline) }, }) replyRootPreloader := fanout.NewScopeReplyThreadRootPreloader(stores.Message) imStreamHub := im.NewStreamEventHub() imStreamWriter := im.NewPublishingStreamEventWriter(stores.IM, imStreamHub) liveEmitter := im.NewLiveEventEmitter(imStreamHub, stores.App, logger) taskStore := taskboard.NewWithOptions(corePool, taskboard.StoreOptions{ TaskChangeSink: im.NewTaskChangeBridge(liveEmitter). WithTaskScopeLifecycleGuard(stores.Message), }) daemonHub := daemon.New(logger, daemon.NewStore(stores.Agent, stores.Rewind, agentRuntime.PrepareAgentStart, stores.Workspace), baseURL, liveEmitter) agentRuntime.SetRuntimeControl(daemonHub) 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, } deliveryFanout := fanout.NewDeliveryFanout(daemonHub, imStreamWriter, logger). WithReplyThreadRootPreloader(replyRootPreloader). WithScopeFanoutMarker(stores.App). WithTaskScopeLifecycleGuard(stores.Message) stores.App.SetCommittedMessageFanout(deliveryFanout) emailVerificationRequired := true authFlow := authflow.New(authflow.Dependencies{ RegistrationAllowlist: stores.Identity, WorkspaceInvites: stores.App, Organizations: organizationService, }) workspaceFlow := workspaceservice.New(workspaceservice.Dependencies{ Workspaces: workspaceStore, WorkspaceAdmin: stores.App, Machines: stores.App, MachineTokens: stores.Agent, MiniappPairings: stores.Identity, }) scheduleManageService := manage.New(logger, corePool, stores.App, manage.ServiceOptions{}) triggerService := trigger.New(logger, corePool, stores.App, trigger.ServiceOptions{}) workflowPlanStore := planstore.New(corePool) workflowPlanService := planmanage.New( workflowPlanStore, planmanage.WithDraftUpdated(liveEmitter.EmitWorkflowPlanDraftUpdated), ) workflowAgentEditService := planmanage.NewAgentEditService( corePool, workflowPlanStore, stores.Message, stores.App, stores.AgentOps, ) workflowBindingStore := binding.New(corePool) workflowBindingService := bindingmanage.New( workflowBindingStore, workflowPlanService, bindingmanage.WithAgentOwnerVerifier(stores.App), ) workflowInvocationStore := invocationstore.New(corePool) workflowInvocationService := invocationmanage.New(workflowInvocationStore) workflowRunStore := run.New(corePool) if err := workflowRunStore.Validate(); err != nil { logger.Error("build Workflow Run store", slog.Any("error", err)) os.Exit(1) } workflowRunService := runmanage.New(workflowRunStore) workflowCalendarService := calendar.New(calendarstore.New(corePool)) workflowRunCallbacks, err := orchestrator.NewCallbackService(workflowRunStore, orchestrator.CallbackServiceOptions{ ScopeAuthorizer: stores.App, TaskWatchMutator: stores.App, }) if err != nil { logger.Error("build Workflow Run callback service", slog.Any("error", err)) os.Exit(1) } workflowApprovalDecisions, err := orchestrator.NewApprovalDecisionService(workflowRunStore) if err != nil { logger.Error("build Workflow Run approval decision service", slog.Any("error", err)) os.Exit(1) } workflowInvocationScheduleDispatcher, err := invocation.NewScheduleDispatcher( workflowInvocationStore, invocation.ScheduleDispatcherOptions{}, ) if err != nil { logger.Error("build Workflow Invocation schedule dispatcher", slog.Any("error", err)) os.Exit(1) } workflowInvocationScheduleWorker := invocation.NewScheduleWorker( workflowInvocationScheduleDispatcher, invocation.ScheduleWorkerOptions{Logger: logger}, ) serviceEventSubscriptionCoordinator := serviceevent.NewServiceEventSubscriptionCoordinator(stores.App) serviceAccountFlow := serviceaccountservice.New(corePool) teamWorkflows := serviceteamworkflow.New(corePool) teamWorkflowBrowserService := teamworkflow.NewStoreService(corePool, teamWorkflows) approvalAppProjector := approvalapp.NewProjector(stores.App, approvalapp.NewProvisioner(corePool)) if err := approvalAppProjector.Register(); err != nil { logger.Error("register Approval notification contract", slog.Any("error", err)) os.Exit(1) } collaboratorRequestService := collaborator.NewWithStore(corePool, stores.Approval, taskStore) approvalSubjects := approvalapp.NewSubjectRegistry(stores.Approval) if err := approvalSubjects.Register(approvalapp.NewCollaboratorRequestSubjectHandler( stores.Approval, collaboratorRequestService, approvalAppProjector, )); err != nil { logger.Error("register collaborator request Approval subject", slog.Any("error", err)) os.Exit(1) } if err := approvalSubjects.Register(approvalapp.NewServiceAccountCreateSubjectHandler( stores.Approval, serviceAccountFlow, approvalAppProjector, )); err != nil { logger.Error("register service account Approval subject", slog.Any("error", err)) os.Exit(1) } if err := approvalSubjects.Register(approvalapp.NewTeamRoleProposalSubjectHandler( stores.Approval, roleproposal.New(corePool), approvalAppProjector, )); err != nil { logger.Error("register Team role proposal Approval subject", slog.Any("error", err)) os.Exit(1) } approvalSubjects.Seal() agentDeps.ApprovalSubjects = approvalSubjects internalAppHandlers := appruntime.NewInternalHandlerRegistry() approvalHandler := approvalapp.NewHandler( stores.Approval, stores.App, approvalAppProjector, approvalSubjects, ) approvalHandler.SetWorkflowRunDecisionSink(workflowApprovalDecisions) if err := internalAppHandlers.Register(approvalapp.HandlerKey, approvalHandler); err != nil { logger.Error("register Approval internal handler", slog.Any("error", err)) os.Exit(1) } internalAppHandlers.Seal() internalAppDispatcher := appruntime.NewInternalDispatcher(appruntime.InternalDispatcherOptions{ Store: stores.App, Registry: internalAppHandlers, Logger: logger, }) approvalProjectionReconciler := approvalapp.NewProjectionReconciler( stores.Approval, approvalSubjects, owner.NewAllSet(), logger, ) agentAPI := agent.New(logger, agentDeps, stores.Attachment, taskStore, daemonHub, deliveryFanout) agentAPI.TeamWorkflowPlans = teamWorkflows agentAPI.MinDaemonVersion = strings.TrimSpace(os.Getenv("AGENTTEAM_DAEMON_MIN_VERSION")) agentAPI.ScheduleService = scheduleManageService agentAPI.TriggerService = triggerService agentAPI.WorkflowPlans = workflowPlanService agentAPI.WorkflowPlanAgentEditGuard = workflowAgentEditService agentAPI.WorkflowBindings = workflowBindingService agentAPI.WorkflowBindingOwners = stores.App agentAPI.WorkflowBindingWorkspace = workspaceStore agentAPI.WorkflowInvocations = workflowInvocationService agentAPI.WorkflowInvocationOwners = stores.App agentAPI.WorkflowInvocationWorkspace = workspaceStore agentAPI.WorkflowRuns = workflowRunService agentAPI.WorkflowRecovery = workflowRunStore agentAPI.WorkflowRunCallbacks = workflowRunCallbacks agentAPI.WorkflowReplies = workflowAgentEditService scopeFanoutRunner := fanout.NewScopeFanoutRunnerWithOptions( corePool, deliveryFanout, logger, fanout.ScopeFanoutRunnerOptions{WakeStatusEmitter: liveEmitter}, ) messagingService := messageservice.New(messageservice.Dependencies{ Sessions: identityStore, Workspaces: workspaceStore, Creation: stores.App, Queries: stores.App, Search: stores.Message, Logger: logger, }) workflowRunMessageWriter := fanout.NewWorkflowRunMessageWriter(identityStore, stores.App, fanout.Dependencies{ HumanMessages: messagingService, Hub: daemonHub, StreamWriter: imStreamWriter, ReplyThreadRootPreloader: replyRootPreloader, ScopeFanoutMarker: stores.App, TaskScopeLifecycle: stores.Message, Logger: logger, }).WithReceiptResolver(fanout.NewWorkflowRunMessageReceiptStore(corePool)). WithResponderResolver(stores.App) workflowRunApprovalAssignees, err := orchestrator.NewHumanApprovalAssigneeResolver(corePool) if err != nil { logger.Error("build Workflow Run approval assignee resolver", slog.Any("error", err)) os.Exit(1) } workflowRunMaterials, err := orchestrator.NewPinnedMaterialLoader(workflowPlanStore, workflowInvocationStore) if err != nil { logger.Error("build Workflow Run material loader", slog.Any("error", err)) os.Exit(1) } workflowRunReducer, err := orchestrator.NewEngine(workflowRunStore, workflowRunMaterials, orchestrator.EngineOptions{ WorkerID: "workflow-run-e2e-reducer", }) if err != nil { logger.Error("build Workflow Run reducer", slog.Any("error", err)) os.Exit(1) } workflowRunEffects, err := orchestrator.NewEffectDispatcher( workflowRunStore, workflowRunMessageWriter, approvalAppProjector, workflowRunApprovalAssignees, orchestrator.EffectDispatcherOptions{ WorkerID: "workflow-run-e2e-effects", Responders: workflowRunMessageWriter, }, ) if err != nil { logger.Error("build Workflow Run effect dispatcher", slog.Any("error", err)) os.Exit(1) } workflowRunRecovery, err := orchestrator.NewRecoveryRunner( workflowRunStore, workflowRunMessageWriter, orchestrator.RecoveryRunnerOptions{ WorkerID: "workflow-run-e2e-recovery", RuntimeWorking: workflowagentruntime.NewHubWorkingReader(workflowagentruntime.ServiceAdapter{Service: stores.Agent}, daemonHub), }, ) if err != nil { logger.Error("build Workflow Run automatic recovery runner", slog.Any("error", err)) os.Exit(1) } workflowRunWorker := orchestrator.NewWorker( workflowRunReducer, workflowRunEffects, orchestrator.WorkerOptions{Logger: logger, Recovery: workflowRunRecovery}, ) httpClient := &http.Client{Timeout: 10 * time.Second} agentteamLLMRefresher := agentteambridge.NewRefresher(organizationService, httpClient, nil, logger) if _, err := agentteamLLMRefresher.Reload(context.Background()); err != nil { logger.Error("load APR-backed organization providers", slog.Any("error", err)) os.Exit(1) } scheduleStore := schedulestore.New(corePool) scheduleService := execschedule.New(logger, scheduleStore, execschedule.ServiceOptions{ Writer: fanout.NewScheduleEffectWriter(scheduleStore, fanout.Dependencies{ HumanMessages: messagingService, Hub: daemonHub, StreamWriter: imStreamWriter, ReplyThreadRootPreloader: replyRootPreloader, ScopeFanoutMarker: stores.App, TaskScopeLifecycle: stores.Message, Logger: logger, }), }) routerDependencies := router.Dependencies{ Logger: logger, Health: router.HealthDependencies{ CoreDB: adminDB, CoreDBAdmin: coreDBAdmin, }, Agent: agentAPI, Extension: e2eExtensionRouteDependencies(corePool), Daemon: router.DaemonDependencies{ Hub: daemonHub, Auth: stores.Agent, SessionIssuer: identityStore, }, API: router.APIDependencies{ Admin: admin.Dependencies{ AuthStore: identityStore, AdminStore: stores.Admin, RegistrationAllowlist: identityStore, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, Auth: auth.Dependencies{ Store: identityStore, AuthFlow: authFlow, Organizations: organizationService, Mailer: mailService, MobileVerificationSender: mobileVerificationSender, EmailVerificationRequired: emailVerificationRequired, RegistrationPolicy: auth.NewRegistrationPolicy(auth.RegistrationModeOpen, nil), VerificationTokenTTL: 24 * time.Hour, SessionTTL: 30 * 24 * time.Hour, Logger: logger, }, Workspace: workspace.Dependencies{ AuthStore: identityStore, WorkspaceService: workspaceFlow, PlanStore: stores.Plan, Organizations: organizationService, Mailer: mailService, EmailVerificationRequired: emailVerificationRequired, MachineRuntime: daemonHub, LatestDaemonVersion: versioning.NewStaticProvider(versioning.ServerVersion()), Logger: logger, }, Org: org.Dependencies{ Store: identityStore, Organizations: organizationService, PlanStore: stores.Plan, Runtime: daemonHub, LLMRegistry: agentteamLLMRefresher, Mailer: mailService, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, Public: public.Dependencies{ AuthStore: identityStore, MessageSearchStore: stores.Message, ContentReports: stores.Reporting, MigrationInspector: coreDBAdmin, CoreDB: adminDB, PingCAPAuthEnabled: false, OpenRegistration: true, RegistrationMode: auth.RegistrationModeOpen, TeamsEnabled: true, DaemonConnectBaseURL: baseURL, MeshXKnowledgeBasesEnabled: true, MeshXWorkflowsEnabled: true, }, Media: media.Dependencies{ AuthStore: identityStore, WorkspaceStore: stores.Workspace, AgentTokenStore: stores.Agent, AgentStore: stores.App, AttachmentStore: stores.Attachment, PersonalStickerStore: stores.Attachment, EmailVerificationRequired: emailVerificationRequired, }, IM: im.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceStore, EmailVerificationRequired: emailVerificationRequired, Logger: logger, 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, Reactions: stores.Reaction, Pins: stores.Message, StreamEvents: stores.IM, VisibleScopes: message.NewSQLVisibleScopeAuthorizer(corePool), StreamWriter: imStreamWriter, StreamSubscriptions: imStreamHub, StreamCursors: stores.IM, ConversationIndex: stores.IM, }, Channel: channel.Dependencies{ AuthStore: identityStore, WorkspaceStore: stores.Workspace, Channels: stores.Channel, AppStore: stores.App, ChannelPermissions: stores.Channel, ThreadStore: stores.Message, ThreadReviewStore: stores.Review, EmailVerificationRequired: emailVerificationRequired, Realtime: liveEmitter, MessageFanout: deliveryFanout, Hub: daemonHub, Logger: logger, }, Task: task.Dependencies{ AuthStore: identityStore, WorkspaceStore: stores.Workspace, EmailVerificationRequired: emailVerificationRequired, TaskBoard: taskStore, ReviewStore: stores.Review, Store: task.NewTaskParityStoreAdapter(stores.App, stores.PrivateAgents), MessageFanout: deliveryFanout, CollaboratorRequests: collaboratorRequestService, ConversationReadProjector: im.NewTaskConversationProjector(stores.Message, imStreamWriter, nil, liveEmitter), ConversationPolicyProjector: im.NewTaskConversationProjector(stores.Message, imStreamWriter, nil, liveEmitter), Logger: logger, }, Schedule: schedule.Dependencies{ AuthStore: identityStore, WorkspaceStore: stores.Workspace, EmailVerificationRequired: emailVerificationRequired, ScheduleService: scheduleManageService, }, ServiceAccount: e2eServiceAccountRouteDependencies( identityStore, workspaceStore, serviceAccountFlow, approvalAppProjector, emailVerificationRequired, ), Trigger: apitrigger.Dependencies{ AuthStore: identityStore, WorkspaceStore: stores.Workspace, EmailVerificationRequired: emailVerificationRequired, TriggerService: triggerService, }, WorkflowBinding: bindingapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceStore, EmailVerificationRequired: emailVerificationRequired, Service: workflowBindingService, }, WorkflowInvocation: invocationapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceStore, EmailVerificationRequired: emailVerificationRequired, Service: workflowInvocationService, }, WorkflowPlan: planapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceStore, EmailVerificationRequired: emailVerificationRequired, Service: workflowPlanService, AgentEdit: workflowAgentEditService, DirectPublishGuard: teamWorkflows, }, WorkflowRun: runapi.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceStore, EmailVerificationRequired: emailVerificationRequired, Service: workflowRunService, Calendar: workflowCalendarService, }, Team: team.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceStore, Teams: stores.Team, EmailVerificationRequired: emailVerificationRequired, }, TeamWorkflow: teamworkflow.Dependencies{ AuthStore: identityStore, WorkspaceStore: workspaceStore, EmailVerificationRequired: emailVerificationRequired, Service: teamWorkflowBrowserService, }, App: app.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceStore, AppStore: stores.App, AppRuntimeStore: stores.App, AppRunnerService: appRunnerService, FollowupDeliverer: deliveryFanout, ServiceEventSubscriptionBlocker: serviceEventSubscriptionCoordinator, InstallationRequestProjector: approvalAppProjector, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, Provider: provider.Dependencies{ SessionAuth: identityStore, WorkspaceStore: workspaceStore, Store: stores.Provider, GitHubAdminStore: stores.App, GiteaAdminStore: stores.App, InstallationRequestProjector: approvalAppProjector, GitHubConfig: provider.GitHubConfig{ FlowStateKey: []byte("agentteam-e2e-github-flow-key"), BaseURL: "https://github.com", APIBaseURL: "https://api.github.com", }, ServiceEventSubscriptionBlocker: serviceEventSubscriptionCoordinator, HTTPClient: httpClient, PublicBaseURL: baseURL, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, LLM: llm.Dependencies{ APRDispatcher: agentteamLLMRefresher.APRDispatcher(), Logger: logger, ProxyKeyAuthenticator: organizationService, UsageTracker: organizationService, }, WebAgent: webagent.Dependencies{ AgentStore: stores.App, ModelProfiles: stores.Agent, SessionAuth: stores.Identity, WorkspaceStore: stores.Workspace, ReviewStore: stores.Review, ProjectStore: stores.Review, RewindStore: stores.Rewind, RuntimeCommands: agentRuntime, Runtime: daemonHub, Realtime: liveEmitter, EmailVerificationRequired: emailVerificationRequired, Logger: logger, }, }, LiveEmitter: liveEmitter, } if err := routerDependencies.Validate(); err != nil { logger.Error("validate router dependencies", slog.Any("error", err)) os.Exit(1) } httpRouter := router.NewRouter(routerDependencies) mux := http.NewServeMux() mux.Handle("/internal/e2e/mailbox/latest", e2eMailboxLatestHandler(mailbox)) mux.Handle("/internal/e2e/mobile-verification/latest", e2eMobileVerificationLatestHandler(mobileVerificationRecorder)) mux.Handle("/internal/e2e/org/state", e2eOrganizationStateHandler(corePool)) mux.Handle("/", httpRouter) server := &http.Server{ Handler: mux, ReadHeaderTimeout: 5 * time.Second, } fmt.Println(baseURL) errCh := make(chan error, 1) go func() { errCh <- server.Serve(listener) }() go scheduleService.Run(ctx) workflowInvocationScheduleWorker.Start(ctx) workflowRunWorker.Start(ctx) go internalAppDispatcher.Run(ctx, 0) go approvalProjectionReconciler.Run(ctx, 0) scopeFanoutRunner.Start(ctx) select { case <-ctx.Done(): shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 10*time.Second) defer shutdownCancel() _ = server.Shutdown(shutdownCtx) case err := <-errCh: if !errors.Is(err, http.ErrServerClosed) { logger.Error("http server exited", slog.Any("error", err)) os.Exit(1) } } } func e2eCheckOrganizationQuotaInTx(context.Context, coredb.Tx, string, plan.MetricKey, int64) error { return nil } func e2eServiceAccountRouteDependencies( sessionAuth api.SessionAuthenticator, workspaceStore api.WorkspaceResolver, service serviceaccount.Service, projector serviceaccount.PendingRequestProjector, emailVerificationRequired bool, ) serviceaccount.Dependencies { return serviceaccount.Dependencies{ SessionAuth: sessionAuth, WorkspaceStore: workspaceStore, Service: service, PendingRequestProjector: projector, EmailVerificationRequired: emailVerificationRequired, } } type tidbRuntime struct { DatabaseDSN string RuntimeDir string closeFn func() error } func e2eOrganizationCoreDependencies(stores *store.Graph) organizationcore.Dependencies { return organizationcore.Dependencies{ Identity: stores.Identity, Workspace: stores.Workspace, Scopes: stores.PrivateAgents, ServiceAccounts: stores.ServiceAccount, DefaultWorkspaceGroupID: config.DefaultWorkspaceGroupID, } } var ( startUnistoreTiDBWithCleanupFn = testkit.StartUnistoreTiDBWithCleanup createTiDBDatabaseFn = createTiDBDatabase dropTiDBDatabaseFn = dropTiDBDatabase ) func (runtime tidbRuntime) Close() { if runtime.closeFn != nil { _ = runtime.closeFn() } if runtime.RuntimeDir != "" { _ = os.RemoveAll(runtime.RuntimeDir) } } func startTiDB(ctx context.Context, logger *slog.Logger) (tidbRuntime, error) { runtimeDir, err := os.MkdirTemp("", "agentteam-e2e-server-*") if err != nil { return tidbRuntime{}, err } if configuredDSN := configuredE2ETiDBDSN(); configuredDSN != "" { if logger != nil { logger.Warn("using configured AGENTTEAM_E2E_TIDB_DSN") } return tidbRuntime{ DatabaseDSN: configuredDSN, RuntimeDir: runtimeDir, }, nil } if adminDSN := configuredE2ETiDBAdminDSN(); adminDSN != "" { if logger != nil { logger.Warn("using configured AGENTTEAM_E2E_TIDB_ADMIN_DSN") } databaseName := fmt.Sprintf("agentteam_e2e_%d", time.Now().UnixNano()) databaseDSN, err := e2eTiDBDatabaseDSN(adminDSN, databaseName) if err != nil { _ = os.RemoveAll(runtimeDir) return tidbRuntime{}, err } if err := createTiDBDatabaseFn(ctx, adminDSN, databaseName); err != nil { _ = os.RemoveAll(runtimeDir) return tidbRuntime{}, err } return tidbRuntime{ DatabaseDSN: databaseDSN, RuntimeDir: runtimeDir, closeFn: func() error { return dropTiDBDatabaseFn(context.Background(), adminDSN, databaseName) }, }, nil } if logger != nil { logger.Warn("starting local unistore tidb runtime") } rootDSN, cleanupRuntime, err := startUnistoreTiDBWithCleanupFn(ctx) if err != nil { _ = os.RemoveAll(runtimeDir) return tidbRuntime{}, fmt.Errorf("start local unistore tidb runtime: %w", err) } databaseName := fmt.Sprintf("agentteam_e2e_%d", time.Now().UnixNano()) databaseDSN, err := e2eTiDBDatabaseDSN(rootDSN, databaseName) if err != nil { _ = cleanupRuntime(context.Background()) _ = os.RemoveAll(runtimeDir) return tidbRuntime{}, err } if err := createTiDBDatabaseFn(ctx, rootDSN, databaseName); err != nil { _ = cleanupRuntime(context.Background()) _ = os.RemoveAll(runtimeDir) return tidbRuntime{}, err } stopRuntime := func() error { var cleanupErr error if err := dropTiDBDatabaseFn(context.Background(), rootDSN, databaseName); err != nil { cleanupErr = err } if err := cleanupRuntime(context.Background()); cleanupErr == nil && err != nil { cleanupErr = err } return cleanupErr } return tidbRuntime{ DatabaseDSN: databaseDSN, RuntimeDir: runtimeDir, closeFn: stopRuntime, }, nil } func configuredE2ETiDBDSN() string { return firstNonEmptyEnv("AGENTTEAM_E2E_TIDB_DSN", "AGENTTEAM_TIDB_TEST_DSN") } func configuredE2ETiDBAdminDSN() string { return firstNonEmptyEnv("AGENTTEAM_E2E_TIDB_ADMIN_DSN", "AGENTTEAM_TIDB_TEST_ADMIN_DSN") } func firstNonEmptyEnv(names ...string) string { for _, name := range names { if value := strings.TrimSpace(os.Getenv(name)); value != "" { return value } } return "" } func createTiDBDatabase(ctx context.Context, adminDSN string, databaseName string) error { adminDB, err := sql.Open("mysql", adminDSN) if err != nil { return fmt.Errorf("open tidb admin dsn: %w", err) } defer func() { _ = adminDB.Close() }() execCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() if _, err := adminDB.ExecContext(execCtx, fmt.Sprintf("CREATE DATABASE IF NOT EXISTS `%s`", databaseName)); err != nil { return fmt.Errorf("create tidb database %s: %w", databaseName, err) } return nil } func dropTiDBDatabase(ctx context.Context, adminDSN string, databaseName string) error { adminDB, err := sql.Open("mysql", adminDSN) if err != nil { return fmt.Errorf("open tidb admin dsn: %w", err) } defer func() { _ = adminDB.Close() }() execCtx, cancel := context.WithTimeout(ctx, 10*time.Second) defer cancel() if _, err := adminDB.ExecContext(execCtx, fmt.Sprintf("DROP DATABASE IF EXISTS `%s`", databaseName)); err != nil { return fmt.Errorf("drop tidb database %s: %w", databaseName, err) } return nil } func e2eTiDBDatabaseDSN(adminDSN string, databaseName string) (string, error) { cfg, err := mysql.ParseDSN(adminDSN) if err != nil { return "", fmt.Errorf("parse tidb admin dsn: %w", err) } cfg.DBName = databaseName return cfg.FormatDSN(), nil }