Merge pull request #527 from lbedner/streaming-worker

Streaming worker

Author: lbedner

aegis/core/post_gen_tasks.py

@@ -337,7 +337,11 @@ def is_enabled(key: str) -> bool:
                 for taskiq_file in queues_dir.glob("*_taskiq.py"):
                     taskiq_file.unlink()
 
-                # arq: Remove TaskIQ pool, API, and registry files
+                # arq: Remove TaskIQ middleware, pool, API, and registry files
+                middleware_taskiq = worker_dir / "middleware_taskiq.py"
+                if middleware_taskiq.exists():
+                    middleware_taskiq.unlink()
+
                 pools_taskiq = worker_dir / "pools_taskiq.py"
                 if pools_taskiq.exists():
                     pools_taskiq.unlink()

aegis/templates/copier-aegis-project/{{ project_slug }}/app/cli/load_test.py.jinja

@@ -54,7 +54,7 @@ class QueueChoice(str, Enum):
 @app.command("run")
 def run_load_test(
     num_tasks: int = typer.Option(
-        100, "--tasks", "-n", help="Number of tasks to spawn", min=10, max=10000
+        100, "--tasks", "-n", help="Number of tasks to spawn", min=1
     ),
     task_type: LoadTestTypes = typer.Option(
         LoadTestTypes.CPU_INTENSIVE, "--type", "-t", help="Type of load test to run"
@@ -131,7 +131,7 @@ def run_load_test(
 @app.command("cpu")
 def quick_cpu_test_cmd(
     num_tasks: int = typer.Option(
-        50, "--tasks", "-n", help="Number of CPU tasks", min=10, max=1000
+        50, "--tasks", "-n", help="Number of CPU tasks", min=1
     ),
     wait: bool = typer.Option(True, "--wait/--no-wait", help="Wait for completion"),
 ) -> None:
@@ -178,7 +178,7 @@ def quick_cpu_test_cmd(
 @app.command("io")
 def quick_io_test_cmd(
     num_tasks: int = typer.Option(
-        100, "--tasks", "-n", help="Number of I/O tasks", min=10, max=1000
+        100, "--tasks", "-n", help="Number of I/O tasks", min=1
     ),
     wait: bool = typer.Option(True, "--wait/--no-wait", help="Wait for completion"),
 ) -> None:
@@ -227,7 +227,7 @@ def quick_io_test_cmd(
 @app.command("memory")
 def quick_memory_test_cmd(
     num_tasks: int = typer.Option(
-        200, "--tasks", "-n", help="Number of memory tasks", min=10, max=1000
+        200, "--tasks", "-n", help="Number of memory tasks", min=1
     ),
     wait: bool = typer.Option(True, "--wait/--no-wait", help="Wait for completion"),
 ) -> None:

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/backend/api/events.py.jinja

@@ -0,0 +1,100 @@
+{%- if include_worker %}
+"""Server-Sent Events endpoint for real-time worker event streaming."""
+
+import asyncio
+import json
+from collections.abc import AsyncGenerator
+
+import redis.asyncio as aioredis
+from fastapi import APIRouter, Request
+from starlette.responses import StreamingResponse
+
+from app.components.worker.events import WORKER_EVENT_STREAM, read_queue_totals
+from app.core.config import settings
+from app.core.log import logger
+
+router = APIRouter()
+
+# XREAD block timeout in milliseconds (5 seconds)
+_XREAD_BLOCK_MS = 5000
+
+# Maximum messages to read per XREAD call — high enough to drain the stream
+# faster than workers produce events during load tests
+_XREAD_COUNT = 500
+
+
+
+@router.get("/worker/stream")
+async def worker_event_stream(request: Request) -> StreamingResponse:
+    """
+    Stream real-time worker events via Server-Sent Events (SSE).
+
+    On connect, sends a "totals" event with absolute counters read from
+    arq's Redis keys. Then tails the Redis Stream and forwards individual
+    events as deltas. The frontend is responsible for batching UI updates.
+    """
+
+    async def event_generator() -> AsyncGenerator[str, None]:
+        redis_url = (
+            settings.redis_url_effective
+            if hasattr(settings, "redis_url_effective")
+            else settings.REDIS_URL
+        )
+        redis_client: aioredis.Redis = aioredis.from_url(
+            redis_url, decode_responses=True
+        )
+        try:
+            # Send initial baseline totals (read from Redis ONCE)
+            totals = await read_queue_totals(redis_client)
+            yield f"data: {json.dumps({'type': 'totals', 'queues': totals})}\n\n"
+            logger.debug(f"SSE: sent initial totals baseline: {totals}")
+
+            # Start from latest entries only (no replay of old events)
+            last_id = "$"
+
+            while True:
+                # Check if client disconnected
+                if await request.is_disconnected():
+                    break
+
+                results = await redis_client.xread(
+                    {WORKER_EVENT_STREAM: last_id},
+                    block=_XREAD_BLOCK_MS,
+                    count=_XREAD_COUNT,
+                )
+
+                if not results:
+                    # No events within the block timeout — send keepalive
+                    yield ": heartbeat\n\n"
+                    continue
+
+                # Forward every stream event as-is
+                msg_count = 0
+                for _stream_name, messages in results:
+                    last_id = messages[-1][0]
+                    msg_count += len(messages)
+                    for _msg_id, data in messages:
+                        yield f"data: {json.dumps(data)}\n\n"
+
+                logger.debug(f"SSE: forwarded {msg_count} events")
+
+                # No sleep needed — XREAD(block=5s) naturally waits
+                # when the stream is empty. Events are forwarded instantly.
+
+        except asyncio.CancelledError:
+            # Client disconnected — normal SSE lifecycle
+            pass
+        except Exception as e:
+            logger.warning(f"SSE stream error: {e}")
+        finally:
+            await redis_client.aclose()
+
+    return StreamingResponse(
+        event_generator(),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "X-Accel-Buffering": "no",
+        },
+    )
+{%- endif %}

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/backend/api/models.py.jinja

@@ -101,8 +101,7 @@ class LoadTestRequest(BaseModel):
     num_tasks: int = Field(
         100,
         description="Number of tasks to spawn for the load test",
-        ge=10,
-        le=10000
+        ge=1,
     )
     task_type: LoadTestTypes = Field(
         LoadTestTypes.CPU_INTENSIVE,

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/backend/api/routing.py.jinja

@@ -2,6 +2,7 @@ from fastapi import FastAPI
 
 from app.components.backend.api import health
 {%- if include_worker %}
+from app.components.backend.api import events as worker_events
 from app.components.backend.api import worker
 {%- endif %}
 {%- if scheduler_backend != "memory" %}
@@ -32,6 +33,7 @@ def include_routers(app: FastAPI) -> None:
     app.include_router(health.router, prefix="/health", tags=["health"])
     {%- if include_worker %}
     app.include_router(worker.router, prefix="/api/v1", tags=["worker"])
+    app.include_router(worker_events.router, prefix="/events", tags=["events"])
     {%- endif %}
     {%- if scheduler_backend != "memory" %}
     app.include_router(scheduler.router, prefix="/api/v1", tags=["scheduler"])

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/backend/api/worker.py.jinja

@@ -15,6 +15,7 @@ from app.components.backend.api.models import (
     TaskStatusResponse,
 )
 from app.components.worker.constants import LoadTestTypes
+from app.components.worker.events import publish_event
 from app.components.worker.pools import get_queue_pool
 from app.components.worker.tasks import get_task_by_name, list_available_tasks
 from app.core.config import get_default_queue
@@ -93,11 +94,18 @@ async def enqueue_task(task_request: TaskRequest) -> TaskResponse:
         if task_request.delay_seconds:
             estimated_start = queued_at + timedelta(seconds=task_request.delay_seconds)
 
-        await pool.aclose()
-
         if job is None:
+            await pool.aclose()
             raise HTTPException(status_code=500, detail="Failed to enqueue task")
 
+        # Publish enqueue event for real-time dashboard updates
+        await publish_event(
+            pool, "job.enqueued", task_request.queue_type,
+            {"job_id": job.job_id, "task": task_request.task_name},
+        )
+
+        await pool.aclose()
+
         logger.info(f"Task enqueued: {job.job_id} ({task_request.task_name})")
 
         return TaskResponse(

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/backend/startup/database_init.py.jinja

@@ -1,7 +1,7 @@
 """
 Database initialization startup hook.
 
-Runs Alembic migrations and {% if database_engine == "sqlite" %}ensures directory structure exists{% else %}verifies connectivity{% endif %}
+{% if database_engine == "sqlite" %}Ensures directory structure exists and creates tables{% else %}Runs Alembic migrations and verifies connectivity{% endif %}
 when the backend starts up (only when database component is included).
 """
 {% if include_database %}
@@ -16,7 +16,7 @@ from app.core.log import logger
 from app.models.user import User  # noqa: F401
 {% endif %}
 
-
+{% if database_engine == "postgres" %}
 def _check_and_stamp_existing_tables() -> None:
     """
     Detect when tables for a pending migration already exist.
@@ -155,16 +155,19 @@ def _run_migrations() -> bool:
     except Exception as e:
         logger.error(f"Failed to run migrations: {e}")
         return False
+{% endif %}
 
 
 async def startup_database_init() -> None:
     """
     Initialize database and run migrations.
 
     This hook runs when the backend starts to:
-    1. {% if database_engine == "sqlite" %}Ensure database directory exists{% else %}Verify database connectivity{% endif %}
+    1. {% if database_engine == "sqlite" %}Ensure database directory exists and create tables{% else %}Verify database connectivity{% endif %}
+{% if database_engine == "postgres" %}
     2. Run Alembic migrations (idempotent - safe to run multiple times)
     3. Verify database schema is ready
+{%- endif %}
     {%- if include_ai and database_engine == "postgres" %}
     4. Sync LLM catalog (AI service, postgres only)
     {%- endif %}
@@ -175,33 +178,47 @@ async def startup_database_init() -> None:
         from app.core.db import DATABASE_PATH
         db_path = Path(DATABASE_PATH)
         db_path.parent.mkdir(parents=True, exist_ok=True)
+
+        # Create tables via SQLModel (no Alembic needed for SQLite)
+        from sqlmodel import SQLModel
+
+        from app.core.db import engine
+        SQLModel.metadata.create_all(engine)
+        logger.info("Database tables created/verified (SQLite)")
 {% endif %}
 
+{% if database_engine == "postgres" %}
         # Run Alembic migrations (idempotent)
         migrations_ok = _run_migrations()
 
         if not migrations_ok:
             logger.warning("Migrations failed - database may not be fully initialized")
             # Continue anyway to allow debugging
+{% endif %}
 
         # Verify database connectivity
         try:
-            from app.core.db import db_session
-            from sqlmodel import text
             from sqlalchemy import inspect
+            from sqlmodel import text
+
+            from app.core.db import db_session
 
             with db_session(autocommit=False) as session:
                 # Basic connectivity check
                 session.exec(text("SELECT 1"))
 
-                # Verify alembic_version table exists
                 inspector = inspect(session.connection())
                 table_names = inspector.get_table_names()
+{% if database_engine == "postgres" %}
 
+                # Verify alembic_version table exists
                 if "alembic_version" in table_names:
                     logger.info(f"Database ready with {len(table_names)} tables")
                 else:
                     logger.warning("alembic_version table missing after migrations")
+{% else %}
+                logger.info(f"Database ready with {len(table_names)} tables")
+{% endif %}
 
         except Exception as e:
             logger.warning(f"Database verification failed: {e}")
@@ -232,4 +249,4 @@ async def startup_database_init() -> None:
 
 # Export the startup hook function
 startup_hook = startup_database_init
-{% endif %}
+{% endif %}

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/frontend/dashboard/cards/card_utils.py.jinja

@@ -8,7 +8,9 @@ No inheritance or ABC complexity - just common functionality extracted.
 from collections.abc import Callable
 
 import flet as ft
+{%- if include_observability %}
 import logfire
+{%- endif %}
 from app.components.frontend.controls import (
     H3Text,
     LabelText,
@@ -434,7 +436,9 @@ def create_progress_indicator(
     )
 
 
+{%- if include_observability %}
 @logfire.instrument("overseer.modal.create {component_name}")
+{%- endif %}
 def create_modal_for_component(
     component_name: str, component_data: ComponentStatus, page: ft.Page
 ) -> ft.Container | None:
@@ -526,7 +530,9 @@ def create_modal_for_component(
     return None
 
 
+{%- if include_observability %}
 @logfire.instrument("overseer.modal.open {component_name}")
+{%- endif %}
 def _open_modal(
     component_name: str, component_data: ComponentStatus, page: ft.Page
 ) -> None:
@@ -538,6 +544,10 @@ def _open_modal(
         if popup:
             modal_cache[component_name] = popup
             page.overlay.append(popup)
+    else:
+        # Refresh cached modal with latest health check data
+        if hasattr(popup, "update_data"):
+            popup.update_data(component_data)
     if popup:
         popup.show()
         page.update()

aegis/templates/copier-aegis-project/{{ project_slug }}/app/components/frontend/dashboard/diagram/diagram_node.py

@@ -10,7 +10,7 @@
 from app.services.system.models import ComponentStatus, ComponentStatusType
 from app.services.system.ui import get_component_label
 
-from ..cards.card_utils import create_modal_for_component, get_ai_engine_display
+from ..cards.card_utils import _open_modal, get_ai_engine_display
 
 # Component configuration: display names and modal routing
 # Subtitles are generated dynamically via get_component_label()
@@ -207,15 +207,7 @@ def _handle_click(self, e: ft.ControlEvent) -> None:
         if not e.page:
             return
 
-        popup = create_modal_for_component(
-            self._modal_name,
-            self._component_data,
-            e.page,
-        )
-        if popup:
-            e.page.overlay.append(popup)
-            popup.show()
-            e.page.update()
+        _open_modal(self._modal_name, self._component_data, e.page)
 
     def update_data(self, component_data: ComponentStatus) -> None:
         """