fix(ci): add production background job E2E verification

.github/workflows/deploy.yml

@@ -344,7 +344,36 @@ jobs:
           curl -sf "$BASE/api/compilation-schemas" -H "Authorization: Bearer $TOKEN" | jq -e 'type == "array"' || { echo "::error::/api/compilation-schemas did not return array"; exit 1; }
           echo "PASS: compilation-schemas"
 
-          echo "Staging smoke tests complete (10/10 passed)"
+          # 11. Background job E2E — enqueue a ping job, wait for worker to process it.
+          # Requires auth (side-effecting endpoint). Retry with backoff to handle
+          # worker startup delays after deploy.
+          PING_OK=false
+          for attempt in 1 2 3 4 5; do
+            PING_RESULT=$(curl -sf "$BASE/health/job-ping" \
+              -H "Authorization: Bearer $TOKEN") || {
+              echo "Attempt $attempt: /health/job-ping request failed, retrying in $((attempt * 5))s..."
+              sleep $((attempt * 5))
+              continue
+            }
+            PING_STATUS=$(echo "$PING_RESULT" | jq -r '.status')
+            if [ "$PING_STATUS" = "ok" ]; then
+              PING_MS=$(echo "$PING_RESULT" | jq -r '.latency_ms')
+              echo "PASS: background job E2E (${PING_MS}ms, attempt $attempt)"
+              PING_OK=true
+              break
+            else
+              PING_ERR=$(echo "$PING_RESULT" | jq -r '.error // "unknown"')
+              echo "Attempt $attempt: job-ping status=$PING_STATUS error=$PING_ERR, retrying in $((attempt * 5))s..."
+              sleep $((attempt * 5))
+            fi
+          done
+          if [ "$PING_OK" != "true" ]; then
+            echo "::error::Background job E2E failed after 5 attempts"
+            echo "$PING_RESULT" | jq . 2>/dev/null || true
+            exit 1
+          fi
+
+          echo "Staging smoke tests complete (11/11 passed)"
 
       - name: 🚨  Create alert on failure
         if: failure()
@@ -540,6 +569,35 @@ jobs:
           ADMIN=$(curl -s -o /dev/null -w "%{http_code}" "$BASE/api/admin/stats" -H "Authorization: Bearer $TOKEN")
           echo "INFO: admin stats responded $ADMIN"
 
+          # 10. Background job E2E — enqueue a ping job, wait for worker to process it.
+          # Requires auth (side-effecting endpoint). Retry with backoff to handle
+          # worker startup delays after deploy.
+          PING_OK=false
+          for attempt in 1 2 3 4 5; do
+            PING_RESULT=$(curl -sf "$BASE/health/job-ping" \
+              -H "Authorization: Bearer $TOKEN") || {
+              echo "Attempt $attempt: /health/job-ping request failed, retrying in $((attempt * 5))s..."
+              sleep $((attempt * 5))
+              continue
+            }
+            PING_STATUS=$(echo "$PING_RESULT" | jq -r '.status')
+            if [ "$PING_STATUS" = "ok" ]; then
+              PING_MS=$(echo "$PING_RESULT" | jq -r '.latency_ms')
+              echo "PASS: background job E2E (${PING_MS}ms, attempt $attempt)"
+              PING_OK=true
+              break
+            else
+              PING_ERR=$(echo "$PING_RESULT" | jq -r '.error // "unknown"')
+              echo "Attempt $attempt: job-ping status=$PING_STATUS error=$PING_ERR, retrying in $((attempt * 5))s..."
+              sleep $((attempt * 5))
+            fi
+          done
+          if [ "$PING_OK" != "true" ]; then
+            echo "::error::Background job E2E failed after 5 attempts"
+            echo "$PING_RESULT" | jq . 2>/dev/null || true
+            exit 1
+          fi
+
           echo "Production functional verification complete"
 
   # -----------------------------------------------------------

docs/openapi.yaml

@@ -4064,6 +4064,38 @@ paths:
                 additionalProperties: true
                 type: object
                 title: Response Deep Health Check Health Deep Get
+  /health/job-ping:
+    get:
+      tags:
+      - Admin
+      summary: Job Ping
+      description: 'Enqueue a lightweight ping job and wait for its result.
+
+
+        Proves the full background job pipeline works end-to-end:
+
+        API -> Redis -> ARQ worker -> result.  Requires authentication
+
+        because it enqueues a real job (side effect).
+
+
+        Returns ``{"status": "ok", ...}`` when the job completes within the
+
+        timeout, or ``{"status": "error", ...}`` on failure or timeout.
+
+        In dev mode (no Redis), returns ``{"status": "skipped"}`` since
+
+        there is no ARQ worker to verify.'
+      operationId: job_ping_health_job_ping_get
+      responses:
+        '200':
+          description: Successful Response
+          content:
+            application/json:
+              schema:
+                additionalProperties: true
+                type: object
+                title: Response Job Ping Health Job Ping Get
   /auth/login/{provider}:
     get:
       tags:
@@ -7033,6 +7065,7 @@ components:
       - sync_push
       - sync_pull
       - poll_rss_feeds
+      - ping
       title: JobType
       description: Type of async job.
     LLMSettingsResponse:

src/wikimind/api/routes/health.py

@@ -4,23 +4,31 @@
 availability, stuck source processing jobs, Redis connectivity, and ARQ
 queue depth. Returns a structured JSON response with per-check status
 and overall system health.
+
+The ``/health/job-ping`` endpoint enqueues a lightweight ping job and
+polls for its result to verify the full background job pipeline:
+API -> Redis -> ARQ worker -> result.
 """
 
 from __future__ import annotations
 
+import asyncio
 import time
 from datetime import timedelta
 from typing import Any
 
 import structlog
 from alembic.config import Config as AlembicConfig
 from alembic.script import ScriptDirectory
-from fastapi import APIRouter
+from arq.jobs import Job as ArqJob
+from arq.jobs import JobStatus as ArqJobStatus
+from fastapi import APIRouter, Depends
 from redis.asyncio import Redis
 from sqlalchemy import text as sa_text
 from sqlmodel import select
 
 from wikimind._datetime import utcnow_naive
+from wikimind.api.deps import get_current_user_id
 from wikimind.config import get_settings
 from wikimind.database import get_session_factory
 from wikimind.models import IngestStatus, Source
@@ -186,3 +194,84 @@
         "status": _overall_status(checks),
         "checks": checks,
     }
+
+
+# ---------------------------------------------------------------------------
+# Job ping — end-to-end background job verification
+# ---------------------------------------------------------------------------
+
+# Maximum time to wait for the ping job to complete (seconds).
+_JOB_PING_TIMEOUT = 30
+# Interval between result polls (seconds).
+_JOB_PING_POLL_INTERVAL = 1
+
+
+# NOTE: This endpoint intentionally has side effects — it enqueues a real
+# (lightweight) ARQ job to verify the job queue is functional end-to-end.
+# Requires authentication to prevent unauthenticated abuse.
+@router.get("/health/job-ping")
+async def job_ping(_user_id: str = Depends(get_current_user_id)) -> dict[str, Any]:
+    """Enqueue a lightweight ping job and wait for its result.
+
+    Proves the full background job pipeline works end-to-end:
+    API -> Redis -> ARQ worker -> result.  Requires authentication
+    because it enqueues a real job (side effect).
+
+    Returns ``{"status": "ok", ...}`` when the job completes within the
+    timeout, or ``{"status": "error", ...}`` on failure or timeout.
+    In dev mode (no Redis), returns ``{"status": "skipped"}`` since
+    there is no ARQ worker to verify.
+    """
+    from wikimind.jobs.background import get_background_compiler  # noqa: PLC0415
+
+    compiler = get_background_compiler()
+
+    if not compiler.is_prod:
+        return {"status": "skipped", "reason": "in-process mode (no Redis)"}
+
+    start = time.monotonic()
+    try:
+        job_id = await compiler.schedule_ping()
+        if job_id is None:
+            return {"status": "error", "error": "Failed to enqueue ping job"}
+
+        # Poll ARQ for the job result
+        pool = await compiler._get_pool()
+        for _ in range(int(_JOB_PING_TIMEOUT / _JOB_PING_POLL_INTERVAL)):
+            arq_job = ArqJob(job_id, redis=pool)
+            status = await arq_job.status()
+            if status == ArqJobStatus.complete:
+                info = await arq_job.result_info()
+                latency_ms = round((time.monotonic() - start) * 1000)
+                return {
+                    "status": "ok",
+                    "job_id": job_id,
+                    "result": info.result if info else None,
+                    "latency_ms": latency_ms,
+                }
+            if status == ArqJobStatus.not_found:
+                # Job disappeared — worker may have crashed
+                latency_ms = round((time.monotonic() - start) * 1000)
+                return {
+                    "status": "error",
+                    "job_id": job_id,
+                    "error": "Job not found — worker may not be running",
+                    "latency_ms": latency_ms,
+                }
+            await asyncio.sleep(_JOB_PING_POLL_INTERVAL)
+
+        latency_ms = round((time.monotonic() - start) * 1000)
+        return {
+            "status": "error",
+            "job_id": job_id,
+            "error": f"Ping job did not complete within {_JOB_PING_TIMEOUT}s",
+            "latency_ms": latency_ms,
+        }
+    except Exception as exc:
+        latency_ms = round((time.monotonic() - start) * 1000)
+        log.warning("health: job-ping failed", error=str(exc))
+        return {
+            "status": "error",
+            "error": str(exc),
+            "latency_ms": latency_ms,
+        }

src/wikimind/jobs/background.py

@@ -72,6 +72,35 @@ def is_prod(self) -> bool:
         """Return True when a real Redis URL is configured."""
         return self._redis_url is not None
 
+    async def schedule_ping(self) -> str | None:
+        """Enqueue a lightweight ping job and return the ARQ job ID.
+
+        Returns the ARQ job ID on success (prod mode) or ``None``
+        in dev mode where no Redis is available.
+
+        Returns:
+            The ARQ job ID string, or None in dev mode.
+        """
+        if not self.is_prod:
+            return None
+
+        pool = await self._get_pool()
+        job = await pool.enqueue_job("ping")
+        return job.job_id if job else None
+
+    async def _get_pool(self) -> ArqRedis:
+        """Return the cached ARQ Redis pool, creating it lazily.
+
+        Returns:
+            The ArqRedis connection pool.
+        """
+        if self._arq_pool is None:
+            async with self._pool_lock:
+                if self._arq_pool is None:
+                    settings = RedisSettings.from_dsn(self._redis_url)  # type: ignore[arg-type]
+                    self._arq_pool = await create_pool(settings)
+        return self._arq_pool
+
     async def schedule_compile(
         self,
         source_id: str,
@@ -198,12 +227,8 @@ async def _enqueue_arq(self, func_name: str, *args: object) -> None:
         overhead. An asyncio.Lock guards initialization to prevent
         duplicate pools under concurrent calls.
         """
-        if self._arq_pool is None:
-            async with self._pool_lock:
-                if self._arq_pool is None:  # double-check after acquiring lock
-                    settings = RedisSettings.from_dsn(self._redis_url)  # type: ignore[arg-type]
-                    self._arq_pool = await create_pool(settings)
-        await self._arq_pool.enqueue_job(func_name, *args)
+        pool = await self._get_pool()
+        await pool.enqueue_job(func_name, *args)
 
     async def close(self) -> None:
         """Close the cached ARQ Redis pool, if any."""

src/wikimind/jobs/worker.py

@@ -226,6 +226,16 @@ async def _on_chunk_progress(message: str) -> None:
     await sweep_wikilinks(ctx, user_id=user_id)
 
 
+async def ping(_ctx) -> str:
+    """Lightweight no-op job that proves the full ARQ pipeline works.
+
+    Returns ``"pong"`` immediately. Used by the production deploy
+    verification step to confirm API -> Redis -> ARQ worker -> result.
+    """
+    log.info("ping job executed")
+    return "pong"
+
+
 async def compile_source(
     ctx,
     source_id: str,
@@ -705,7 +715,7 @@ def get_redis_settings() -> RedisSettings:
 class WorkerSettings:
     """ARQ worker configuration for production (requires Redis)."""
 
-    functions: ClassVar[list] = [compile_source, lint_wiki, recompile_article, sweep_wikilinks]
+    functions: ClassVar[list] = [ping, compile_source, lint_wiki, recompile_article, sweep_wikilinks]
     redis_settings = get_redis_settings()
     _worker_cfg = get_settings().worker
     max_jobs = _worker_cfg.max_jobs

src/wikimind/models/enums.py

@@ -122,6 +122,7 @@ class JobType(StrEnum):
     SYNC_PUSH = "sync_push"
     SYNC_PULL = "sync_pull"
     POLL_RSS_FEEDS = "poll_rss_feeds"
+    PING = "ping"
 
 
 class JobStatus(StrEnum):