wip multiple workers, created_by, refactor values

Signed-off-by: Radek Ježek <[email protected]>

Author: jezekra1

apps/beeai-cli/src/beeai_cli/commands/platform/base_driver.py

@@ -110,7 +110,6 @@ async def deploy(
         values = {
             **{svc: {"service": {"type": "LoadBalancer"}} for svc in ["collector", "docling", "ui", "phoenix"]},
             "hostNetwork": True,
-            "localDockerRegistry": {"enabled": True},
             "externalRegistries": {"public_github": str(Configuration().agent_registry)},
             "encryptionKey": "Ovx8qImylfooq4-HNwOzKKDcXLZCB3c_m0JlB9eJBxc=",
             "features": {

apps/beeai-sdk/src/beeai_sdk/platform/provider_build.py

@@ -27,6 +27,7 @@ class ProviderBuild(pydantic.BaseModel):
     status: BuildState
     source: ResolvedGithubUrl
     destination: str
+    created_by: str
 
     @staticmethod
     async def create(*, location: str, client: PlatformClient | None = None) -> ProviderBuild:

apps/beeai-server/src/beeai_server/api/routes/provider_builds.py

@@ -22,11 +22,11 @@
 
     @router.post("")
     async def create_provider_build(
-        _: Annotated[AuthorizedUser, Depends(RequiresPermissions(provider_builds={"write"}))],
+        user: Annotated[AuthorizedUser, Depends(RequiresPermissions(provider_builds={"write"}))],
         request: CreateProviderBuildRequest,
         provider_build_service: ProviderBuildServiceDependency,
     ) -> ProviderBuild:
-        return await provider_build_service.create_build(location=request.location)
+        return await provider_build_service.create_build(location=request.location, user=user.user)
 
     @router.get("/{id}")
     async def get_provider_build(

apps/beeai-server/src/beeai_server/configuration.py

@@ -46,8 +46,8 @@ def validate_level(cls, v: str | int | None):
 class OCIRegistryConfiguration(BaseModel, extra="allow"):
     username: str | None = None
     password: Secret[str] | None = None
-    insecure: bool = False
     auth_header: Secret[str] | None = None
+    insecure: bool = False
 
     @property
     def protocol(self):
@@ -168,6 +168,7 @@ class DockerConfigJsonAuth(BaseModel, extra="allow"):
     auth: Secret[str] | None = None
     username: str | None = None
     password: Secret[str] | None = None
+    insecure: bool = False
 
 
 class DockerConfigJson(BaseModel):
@@ -241,7 +242,11 @@ def _oci_registry_defaultdict(self):
         for docker_config_json in self.oci_registry_docker_config_json.values():
             try:
                 for registry, conf in docker_config_json.auths.items():
-                    registry_short = AnyUrl(registry).host if "://" in registry else registry.strip("/")
+                    if "://" in registry:
+                        url = AnyUrl(registry)
+                        registry_short = f"{url.host}:{url.port}" if url.port else url.host
+                    else:
+                        registry_short = registry.strip("/")
                     # For some reason dockerhub registry has a weird url
                     assert registry_short
                     aliases = [registry_short]
@@ -251,6 +256,7 @@ def _oci_registry_defaultdict(self):
                         self.oci_registry[alias].username = conf.username
                         self.oci_registry[alias].password = conf.password
                         self.oci_registry[alias].auth_header = conf.auth
+                        self.oci_registry[alias].insecure = conf.insecure
             except ValueError as e:
                 logger.error(f"Failed to parse .dockerconfigjson: {e}. Some agent images might not work correctly.")
         if not self.provider_build.oci_build_registry_prefix and len(self.oci_registry):

apps/beeai-server/src/beeai_server/domain/models/provider_build.py

@@ -31,3 +31,4 @@ class ProviderBuild(BaseModel):
     status: BuildState
     source: ResolvedGithubUrl
     destination: DockerImageID
+    created_by: UUID

apps/beeai-server/src/beeai_server/infrastructure/persistence/migrations/alembic/versions/b0388daeb831_.py

@@ -5,35 +5,40 @@
 Create Date: 2025-09-22 12:42:16.598519
 
 """
-from typing import Sequence, Union
 
-from alembic import op
-import sqlalchemy as sa
+from collections.abc import Sequence
 
+import sqlalchemy as sa
+from alembic import op
 
 # revision identifiers, used by Alembic.
-revision: str = 'b0388daeb831'
-down_revision: Union[str, None] = '7b933a4a8cfc'
-branch_labels: Union[str, Sequence[str], None] = None
-depends_on: Union[str, Sequence[str], None] = None
+revision: str = "b0388daeb831"
+down_revision: str | None = "7b933a4a8cfc"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
 
 
 def upgrade() -> None:
     """Upgrade schema."""
     # ### commands auto generated by Alembic - please adjust! ###
-    op.create_table('provider_builds',
-    sa.Column('id', sa.UUID(), nullable=False),
-    sa.Column('source', sa.JSON(), nullable=False),
-    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
-    sa.Column('status', sa.Enum('missing', 'in_progress', 'completed', 'failed', name='buildstate'), nullable=False),
-    sa.Column('destination', sa.String(length=512), nullable=False),
-    sa.PrimaryKeyConstraint('id')
+    op.create_table(
+        "provider_builds",
+        sa.Column("id", sa.UUID(), nullable=False),
+        sa.Column("source", sa.JSON(), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("created_by", sa.UUID(), nullable=False),
+        sa.Column(
+            "status", sa.Enum("missing", "in_progress", "completed", "failed", name="buildstate"), nullable=False
+        ),
+        sa.Column("destination", sa.String(length=512), nullable=False),
+        sa.ForeignKeyConstraint(["created_by"], ["users.id"], ondelete="CASCADE"),
+        sa.PrimaryKeyConstraint("id"),
     )
     # ### end Alembic commands ###
 
 
 def downgrade() -> None:
     """Downgrade schema."""
     # ### commands auto generated by Alembic - please adjust! ###
-    op.drop_table('provider_builds')
+    op.drop_table("provider_builds")
     # ### end Alembic commands ###

apps/beeai-server/src/beeai_server/infrastructure/persistence/repositories/provider_build.py

@@ -5,7 +5,7 @@
 from typing import Any
 from uuid import UUID
 
-from sqlalchemy import JSON, Column, DateTime, Row, String, Table
+from sqlalchemy import JSON, Column, DateTime, ForeignKey, Row, String, Table
 from sqlalchemy import UUID as SQL_UUID
 from sqlalchemy.ext.asyncio import AsyncConnection
 from sqlalchemy.sql import select
@@ -23,6 +23,8 @@
     Column("id", SQL_UUID, primary_key=True),
     Column("source", JSON, nullable=False),
     Column("created_at", DateTime(timezone=True), nullable=False),
+    # The CASCADE might leave some k8s jobs orphaned without cancellation, but jobs have timeout and self-deletion
+    Column("created_by", ForeignKey("users.id", ondelete="CASCADE"), nullable=True),
     Column("status", sql_enum(BuildState), nullable=False),
     Column("destination", String(512), nullable=False),
 )
@@ -50,6 +52,7 @@ def _to_row(self, provider_build: ProviderBuild) -> dict[str, Any]:
             "source": provider_build.source.model_dump(mode="json"),
             "created_at": provider_build.created_at,
             "status": provider_build.status,
+            "created_by": provider_build.created_by,
             "destination": str(provider_build.destination),
         }
 
@@ -60,6 +63,7 @@ def _to_provider_build(self, row: Row) -> ProviderBuild:
                 "source": row.source,
                 "destination": row.destination,
                 "created_at": row.created_at,
+                "created_by": row.created_by,
                 "status": row.status,
             }
         )

apps/beeai-server/src/beeai_server/jobs/crons/cleanup.py

@@ -5,6 +5,7 @@
 from kink import inject
 from procrastinate import Blueprint, JobContext, builtin_tasks
 
+from beeai_server.jobs.queues import Queues
 from beeai_server.service_layer.services.contexts import ContextService
 
 blueprint = Blueprint()
@@ -13,7 +14,7 @@
 
 
 @blueprint.periodic(cron="5 * * * *")
-@blueprint.task(queueing_lock="cleanup_expired_vector_stores", queue="cron:cleanup")
+@blueprint.task(queueing_lock="cleanup_expired_vector_stores", queue=str(Queues.CRON_CLEANUP))
 @inject
 async def cleanup_expired_context_resources(timestamp: int, context: ContextService) -> None:
     """Delete resources of contexts that haven't been used for several days."""
@@ -22,7 +23,7 @@ async def cleanup_expired_context_resources(timestamp: int, context: ContextServ
 
 
 @blueprint.periodic(cron="*/10 * * * *")
-@blueprint.task(queueing_lock="remove_old_jobs", queue="cron:cleanup", pass_context=True)
+@blueprint.task(queueing_lock="remove_old_jobs", queue=str(Queues.CRON_CLEANUP), pass_context=True)
 async def remove_old_jobs(context: JobContext, timestamp: int):
     return await builtin_tasks.remove_old_jobs(
         context,

apps/beeai-server/src/beeai_server/jobs/crons/mcp.py

@@ -12,6 +12,7 @@
 
 from beeai_server.configuration import get_configuration
 from beeai_server.domain.models.mcp_provider import McpProviderDeploymentState
+from beeai_server.jobs.queues import Queues
 from beeai_server.service_layer.services.mcp import McpService
 from beeai_server.utils.utils import extract_messages
 
@@ -22,7 +23,7 @@
 if get_configuration().mcp.auto_remove_enabled:
 
     @blueprint.periodic(cron="* * * * * */5")
-    @blueprint.task(queueing_lock="auto_remove_mcp_providers", queue="cron:mcp_provider")
+    @blueprint.task(queueing_lock="auto_remove_mcp_providers", queue=str(Queues.CRON_MCP_PROVIDER))
     @inject
     async def auto_remove_providers(timestamp: int, mcp_service: McpService):
         providers = await mcp_service.list_providers()

apps/beeai-server/src/beeai_server/jobs/crons/provider.py

@@ -16,6 +16,7 @@
 from beeai_server.configuration import Configuration
 from beeai_server.domain.models.provider import ProviderLocation
 from beeai_server.exceptions import EntityNotFoundError
+from beeai_server.jobs.queues import Queues
 from beeai_server.service_layer.services.provider import ProviderService
 from beeai_server.service_layer.unit_of_work import IUnitOfWorkFactory
 from beeai_server.utils.utils import extract_messages
@@ -26,15 +27,15 @@
 
 
 @blueprint.periodic(cron="*/1 * * * *")
-@blueprint.task(queueing_lock="scale_down_providers", queue="cron:provider")
+@blueprint.task(queueing_lock="scale_down_providers", queue=str(Queues.CRON_PROVIDER))
 @inject
 async def scale_down_providers(timestamp: int, service: ProviderService):
     await service.scale_down_providers()
 
 
 # TODO: Can't use DI here because it's not initialized yet
 @blueprint.periodic(cron=get_configuration().agent_registry.sync_period_cron)
-@blueprint.task(queueing_lock="check_registry", queue="cron:provider")
+@blueprint.task(queueing_lock="check_registry", queue=str(Queues.CRON_PROVIDER))
 @inject
 async def check_registry(timestamp: int, configuration: Configuration, provider_service: ProviderService):
     if not configuration.agent_registry.locations:
@@ -102,7 +103,7 @@ async def check_registry(timestamp: int, configuration: Configuration, provider_
 if get_configuration().provider.auto_remove_enabled:
 
     @blueprint.periodic(cron="* * * * * */5")
-    @blueprint.task(queueing_lock="auto_remove_providers", queue="cron:provider")
+    @blueprint.task(queueing_lock="auto_remove_providers", queue=str(Queues.CRON_PROVIDER))
     @inject
     async def auto_remove_providers(timestamp: int, uow_f: IUnitOfWorkFactory, provider_service: ProviderService):
         async with uow_f() as uow: