Async runs via priority queue: user jobs HIGH, background learning LOW; single worker; UI polls

Author: dev-pd

backend/app/api/run.py

@@ -1,25 +1,21 @@
-"""Live agent-run endpoint — audit a real URL with A and B.
+"""Async agent-run endpoints — audit a real URL with A and B via the job queue.
 
-Runs both agents live (real page fetch + live SERP + Anthropic) on any URL +
-keyword the user enters, and records the results so a later replay is free. There
-is no answer key for an arbitrary real page, so this returns recommendations and
-cost, not scores — the scored A-vs-B proof lives on the planted benchmark.
+`POST /run` enqueues a HIGH-priority job and returns a `job_id` immediately; the
+single worker runs both agents (B learning on), and the UI polls `GET /run/{job_id}`
+until it is done. Going through the queue means a user run jumps ahead of background
+learning, waiting at most for the one page already in flight.
 """
 
 from __future__ import annotations
 
-import httpx
+import uuid
+
 from fastapi import APIRouter, HTTPException
 from pydantic import BaseModel
 
-from app.agents.agent_a import run_agent_a
-from app.agents.agent_b import run_agent_b
-from app.core.exceptions import ReplayMissError
-from app.memory.janitor import run_janitor
-from app.models import AgentRunResult
-from app.repositories.diary import DiaryRepository
-from app.tools.fetch_page import FetchPageInput, FetchPageTool
-from app.tools.registry import build_default_registry
+from app.jobs.queue import HIGH, Job, get_queue
+from app.models import RunJobRecord
+from app.repositories.run_results import RunResultRepository
 
 router = APIRouter(prefix="/run", tags=["run"])
 
@@ -29,49 +25,29 @@ class RunRequest(BaseModel):
     keyword: str
 
 
-class RunResponse(BaseModel):
-    a: AgentRunResult
-    b: AgentRunResult
-    note: str | None = (
-        None  # set when the page body couldn't be fetched (SERP-only run)
-    )
+class RunAccepted(BaseModel):
+    job_id: str
+
 
+@router.post("", response_model=RunAccepted, status_code=202)
+async def run(req: RunRequest) -> RunAccepted:
+    """Enqueue an A-vs-B audit; the UI polls GET /run/{job_id} for the result."""
+    job_id = uuid.uuid4().hex
+    await RunResultRepository().create(
+        RunJobRecord(job_id=job_id, url=req.url, keyword=req.keyword)
+    )
+    get_queue().enqueue(
+        Job(
+            name="run", args={"job_id": job_id, "url": req.url, "keyword": req.keyword}
+        ),
+        priority=HIGH,
+    )
+    return RunAccepted(job_id=job_id)
 
-@router.post("", response_model=RunResponse)
-async def run(req: RunRequest) -> RunResponse:
-    try:
-        # Probe the page once. A bot-block no longer fails the run — fetch_page degrades
-        # to a SERP-only audit and tells the user via `note`.
-        probe = await FetchPageTool(replay=False).run(FetchPageInput(url=req.url))
-        a = await run_agent_a(
-            req.url,
-            req.keyword,
-            registry=build_default_registry(replay=False),
-            replay=False,
-        )
-        # B learns on every live run: read the playbook, audit, jot one diary lesson…
-        b = await run_agent_b(
-            req.url,
-            req.keyword,
-            registry=build_default_registry(replay=False),
-            replay=False,
-            memory_on=True,
-            learn=True,
-            run_index=await DiaryRepository().next_run(),
-        )
-        # …then consolidate the diary into the playbook so the NEXT run sees it.
-        await run_janitor()
-    except httpx.HTTPError as exc:
-        # A page bot-block degrades gracefully (see fetch_page); this only fires for
-        # other live HTTP failures, e.g. the SERP call.
-        raise HTTPException(
-            status_code=502,
-            detail=f"Live request failed: {exc.__class__.__name__}. Try again or a different page.",
-        ) from exc
-    except ReplayMissError as exc:
-        raise HTTPException(status_code=409, detail=str(exc)) from exc
-    except RuntimeError as exc:
-        # e.g. live SERP needs SERP_API_KEY
-        raise HTTPException(status_code=502, detail=str(exc)) from exc
 
-    return RunResponse(a=a, b=b, note=None if probe.fetch_ok else probe.fetch_note)
+@router.get("/{job_id}", response_model=RunJobRecord)
+async def run_status(job_id: str) -> RunJobRecord:
+    record = await RunResultRepository().get(job_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail="No such run job.")
+    return record

backend/app/jobs/handlers.py

@@ -0,0 +1,82 @@
+"""Job handlers — the work the single worker performs, by job name.
+
+- `run`        : a user's A-vs-B audit (HIGH priority). Fills a RunJobRecord the UI polls.
+- `seed_page`  : learn from one discovered page (LOW priority, background).
+- `janitor`    : consolidate the diary into the playbook (LOW priority).
+
+All are one page / one consolidation each, so a user `run` only ever waits for the
+single background page in flight — never a whole pass.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Awaitable, Callable
+from typing import Any
+
+from app.agents.agent_a import run_agent_a
+from app.agents.agent_b import run_agent_b
+from app.memory.janitor import run_janitor
+from app.models import JobStatus
+from app.repositories.diary import DiaryRepository
+from app.repositories.run_results import RunResultRepository
+from app.tools.fetch_page import FetchPageInput, FetchPageTool
+from app.tools.registry import build_default_registry
+
+logger = logging.getLogger("worker")
+
+
+async def handle_run(args: dict[str, Any]) -> None:
+    """Run both agents on the user's page; B learns; consolidate; record the result."""
+    job_id, url, keyword = args["job_id"], args["url"], args["keyword"]
+    results = RunResultRepository()
+    await results.update(job_id, status=JobStatus.RUNNING)
+    try:
+        probe = await FetchPageTool(replay=False).run(FetchPageInput(url=url))
+        a = await run_agent_a(
+            url, keyword, registry=build_default_registry(replay=False), replay=False
+        )
+        b = await run_agent_b(
+            url,
+            keyword,
+            registry=build_default_registry(replay=False),
+            replay=False,
+            memory_on=True,
+            learn=True,
+            run_index=await DiaryRepository().next_run(),
+        )
+        await run_janitor()
+        await results.update(
+            job_id,
+            status=JobStatus.DONE,
+            a=a.model_dump(),
+            b=b.model_dump(),
+            note=None if probe.fetch_ok else probe.fetch_note,
+        )
+    except Exception as exc:  # surface a clean message to the polling UI
+        logger.exception("run job %s failed", job_id)
+        await results.update(job_id, status=JobStatus.ERROR, error=str(exc))
+
+
+async def handle_seed_page(args: dict[str, Any]) -> None:
+    """Learn from one discovered page (no janitor here — that is its own job)."""
+    await run_agent_b(
+        args["url"],
+        args["keyword"],
+        registry=build_default_registry(replay=False),
+        replay=False,
+        memory_on=True,
+        learn=True,
+        run_index=await DiaryRepository().next_run(),
+    )
+
+
+async def handle_janitor(args: dict[str, Any]) -> None:
+    await run_janitor()
+
+
+HANDLERS: dict[str, Callable[[dict[str, Any]], Awaitable[None]]] = {
+    "run": handle_run,
+    "seed_page": handle_seed_page,
+    "janitor": handle_janitor,
+}

backend/app/jobs/queue.py

@@ -112,3 +112,18 @@ def pop(self) -> Job | None:
         job = Job.model_validate_json(raw)
         self._backend.clear_pending(job.dedup_key())  # no longer pending
         return job
+
+
+_queue: JobQueue | None = None
+
+
+def get_queue() -> JobQueue:
+    """The shared Redis-backed queue (built once per process)."""
+    global _queue
+    if _queue is None:
+        import redis
+
+        from app.core.config import settings
+
+        _queue = JobQueue(RedisBackend(redis.from_url(settings.redis_url)))
+    return _queue

backend/app/jobs/scheduler.py

@@ -1,32 +1,39 @@
 """Scheduler entrypoint.
 
-When `SEED_IDLE_ENABLED` is set, this cycles the curated seed jobs on a timer to
-grow Agent B's memory between user jobs (no web roaming — it only works the
-hand-picked seed list). Off by default, so it idles and spends nothing until you
-opt in.
+On a timer, enqueues a batch of background learning jobs: B discovers fetchable
+pages for the given seed topics and queues them (LOW priority) for the worker, plus
+a trailing janitor. The worker drains them one page at a time, always yielding to
+user runs (HIGH). On by default, but only runs when live keys are present, so a
+keyless clone still boots clean and silent.
 """
 
 import asyncio
 import logging
 import time
 
 from app.core.config import settings
-from app.jobs.seed_runner import run_idle_loop
+from app.jobs.seed_runner import run_enqueue_loop
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("scheduler")
 
 
 def main() -> None:
-    # Auto-learn in the background, but only when we can actually run live (key present).
-    if settings.seed_idle_enabled and settings.anthropic_api_key:
+    # Enqueue background learning, but only when we can actually run live.
+    if (
+        settings.seed_idle_enabled
+        and settings.anthropic_api_key
+        and settings.serp_api_key
+    ):
         logger.info(
-            "seed idle loop on; cycling curated jobs every %ds",
+            "seed scheduler on; enqueuing a learning batch every %ds",
             settings.seed_idle_interval_seconds,
         )
-        asyncio.run(run_idle_loop(interval_seconds=settings.seed_idle_interval_seconds))
+        asyncio.run(
+            run_enqueue_loop(interval_seconds=settings.seed_idle_interval_seconds)
+        )
         return
-    logger.info("scheduler idle (no Anthropic key, or seeding disabled).")
+    logger.info("scheduler idle (missing live keys, or seeding disabled).")
     while True:
         time.sleep(3600)
 

backend/app/jobs/seed_runner.py

@@ -20,6 +20,7 @@
 
 from app.agents.agent_b import run_agent_b
 from app.jobs.discovery import discover_pages, page_is_fetchable
+from app.jobs.queue import LOW, Job, JobQueue, get_queue
 from app.jobs.seed_topics import SEED_TOPICS
 from app.memory.janitor import run_janitor
 from app.models import AgentRunResult
@@ -102,11 +103,45 @@ async def run_seed_pass(
     )
 
 
-async def run_idle_loop(*, interval_seconds: int, passes: int = 1) -> None:
-    """Forever: run one seed pass, then sleep. Gated by config; off without a key."""
+async def enqueue_seed_pass(
+    queue: JobQueue | None = None,
+    *,
+    topics: list[str] | None = None,
+    pages_per_topic: int = 4,
+    replay: bool = False,
+    discover: DiscoverFn | None = None,
+    probe: ProbeFn | None = None,
+) -> int:
+    """Discover fetchable pages per topic and enqueue them as LOW-priority jobs.
+
+    The worker does the actual auditing one page at a time, so user runs (HIGH) keep
+    jumping ahead. A trailing janitor job consolidates after the batch.
+    """
+    queue = queue or get_queue()
+    topics = topics if topics is not None else SEED_TOPICS
+    discover = discover or (lambda kw, n: discover_pages(kw, n, replay=replay))
+    probe = probe or (lambda url: page_is_fetchable(url, replay=replay))
+
+    enqueued = 0
+    for topic in topics:
+        for url in await discover(topic, pages_per_topic):
+            if not await probe(url):
+                continue
+            job = Job(
+                name="seed_page", args={"url": url, "keyword": topic}, key=f"seed:{url}"
+            )
+            if queue.enqueue(job, priority=LOW):
+                enqueued += 1
+    queue.enqueue(Job(name="janitor", key="janitor"), priority=LOW)
+    logger.info("enqueued %d seed pages + janitor", enqueued)
+    return enqueued
+
+
+async def run_enqueue_loop(*, interval_seconds: int, pages_per_topic: int = 4) -> None:
+    """Forever: enqueue a background learning batch, then sleep. Off without keys."""
     while True:
         try:
-            await run_seed_pass(passes=passes)
-        except Exception:  # never let one bad pass kill the idle loop
-            logger.exception("seed pass failed; will retry next tick")
+            await enqueue_seed_pass(pages_per_topic=pages_per_topic)
+        except Exception:  # never let one bad tick kill the loop
+            logger.exception("seed enqueue failed; will retry next tick")
         await asyncio.sleep(interval_seconds)

backend/app/jobs/worker.py

@@ -1,21 +1,45 @@
-"""Queue worker entrypoint (stub).
+"""Queue worker — the single drain loop.
 
-The real Redis-backed worker arrives in Phase 4. Until then this only exists so
-`python -m app.jobs.worker` is a valid entrypoint; Compose runs it as an idle
-stub. Run directly, it logs once and idles.
+Pops one job at a time, high priority first, and runs its handler. Because there is
+exactly one worker and it never preempts a running job, a user `run` (HIGH) jumps
+ahead of all pending background `seed_page`/`janitor` work (LOW) — waiting at most
+for the one background page already in flight.
 """
 
+from __future__ import annotations
+
+import asyncio
 import logging
-import time
+
+from app.jobs.handlers import HANDLERS
+from app.jobs.queue import get_queue
 
 logging.basicConfig(level=logging.INFO)
 logger = logging.getLogger("worker")
 
+IDLE_SLEEP_SECONDS = 1.0
 
-def main() -> None:
-    logger.info("worker stub started; queue is wired in Phase 4. Idling.")
+
+async def drain() -> None:
+    queue = get_queue()
+    logger.info("worker started; draining HIGH then LOW.")
     while True:
-        time.sleep(3600)
+        job = queue.pop()
+        if job is None:
+            await asyncio.sleep(IDLE_SLEEP_SECONDS)
+            continue
+        handler = HANDLERS.get(job.name)
+        if handler is None:
+            logger.warning("no handler for job %r; dropping", job.name)
+            continue
+        try:
+            await handler(job.args)
+        except Exception:  # one bad job must not kill the worker
+            logger.exception("job %r failed", job.name)
+
+
+def main() -> None:
+    asyncio.run(drain())
 
 
 if __name__ == "__main__":

backend/app/models/__init__.py

@@ -1,5 +1,6 @@
 """Domain models. Import from here: ``from app.models import Page, Recommendation``."""
 
+from app.models.job import JobStatus, RunJobRecord
 from app.models.page import Heading, Page
 from app.models.proactive import BlindSpot, MonitorAlert
 from app.models.recommendation import Recommendation, RecType, Severity, Stage
@@ -12,10 +13,12 @@
     "BlindSpot",
     "CheckResult",
     "Heading",
+    "JobStatus",
     "MonitorAlert",
     "Page",
     "RecType",
     "Recommendation",
+    "RunJobRecord",
     "SerpEntry",
     "SerpResult",
     "Severity",