src/endopath/worker.py

"""The in-process worker that drains the Intake follow-up queue.

Intake enqueues report induction (#92) and dictionary compilation (#91) for each submitted
source-and-dictionary pair (`jobs.py`). This module claims those jobs from the durable queue and runs
them one at a time, in a background thread started from the API lifespan. It matches the app's
local-first, single-process posture: no external queue or worker process. The claim goes through
`storage.claim_next_job`, a conditional update, so even if a second process ran this loop a job would
still run once.

An executor runs one job to completion or raises; the worker records done or failed around it, so a
single job's failure never stops the loop and always leaves a recorded reason. The executors register
themselves here: report induction with #107, dictionary compilation with #108.

Tests drive `process_next_job` directly and leave the background thread off, so the fast suite never
starts a thread or spends the Anthropic key.
"""

from __future__ import annotations

import logging
import os
import threading
from pathlib import Path
from typing import Callable, Optional

from sqlalchemy import Connection

from endopath import storage

logger = logging.getLogger(__name__)

# kind -> executor. Populated by the executor tickets. An executor gets the connection and the claimed
# job row (dict), reads what it needs from the row (dictionary_id, source_kind, counts), and writes its
# artifact; it returns nothing and raises on failure.
Executor = Callable[[Connection, dict], None]
EXECUTORS: dict[str, Executor] = {}


def register_executor(kind: str, executor: Executor) -> None:
    """Register the executor for a job kind. Called at import by the executor modules."""
    EXECUTORS[kind] = executor


def process_next_job(conn: Connection) -> Optional[dict]:
    """Claim and run the next enqueued job, returning the claimed job row with its final status, or None
    when the queue is drained. An executor that raises marks the job failed with the error text; a kind
    with no registered executor fails the same way rather than silently completing."""
    job = storage.claim_next_job(conn)
    if job is None:
        return None
    executor = EXECUTORS.get(job["kind"])
    try:
        if executor is None:
            raise RuntimeError(f"no executor registered for job kind {job['kind']!r}")
        executor(conn, job)
    except Exception as exc:  # noqa: BLE001 - a failed job records its reason, it does not crash the loop
        logger.exception("job %s (%s) failed", job["id"], job["kind"])
        storage.finish_job(conn, job_id=job["id"], status="failed", error=str(exc))
        job["status"] = "failed"
        job["error"] = str(exc)
        return job
    storage.finish_job(conn, job_id=job["id"], status="done")
    job["status"] = "done"
    return job


def drain(conn: Connection) -> int:
    """Run every currently enqueued job, returning how many ran. For a synchronous drain in a test or a
    CLI, where the background thread is not wanted."""
    count = 0
    while process_next_job(conn) is not None:
        count += 1
    return count


# --- The background thread -----------------------------------------------------

POLL_SECONDS = 2.0
WORKER_DISABLED_ENV = "ENDOPATH_WORKER_DISABLED"

_thread: Optional[threading.Thread] = None
_stop = threading.Event()


def _loop(db_path: Path | str) -> None:
    conn = storage.connect(db_path)
    try:
        while not _stop.is_set():
            try:
                ran = process_next_job(conn)
            except Exception:  # noqa: BLE001 - a storage hiccup must not kill the loop
                logger.exception("worker loop error")
                ran = None
            if ran is None:
                # Queue drained (or errored): wait before polling again, and wake immediately on stop.
                _stop.wait(POLL_SECONDS)
    finally:
        conn.close()


def start(db_path: Path | str) -> bool:
    """Start the background worker thread, returning whether one was started. Disabled under pytest and
    when ENDOPATH_WORKER_DISABLED is set, so tests drive process_next_job directly and never start a
    thread. A no-op if a worker thread is already running."""
    global _thread
    if os.environ.get(WORKER_DISABLED_ENV) or "PYTEST_CURRENT_TEST" in os.environ:
        return False
    if _thread is not None and _thread.is_alive():
        return False
    # Importing the executors module registers report induction (#107) and dictionary compilation
    # (#108) into EXECUTORS. Imported here, at the point the worker starts, so an import cycle never
    # forms at module load (executors imports worker) and the heavy extraction stack loads only when a
    # deployment actually runs jobs.
    from endopath import executors  # noqa: F401

    _stop.clear()
    _thread = threading.Thread(target=_loop, args=(db_path,), name="endopath-worker", daemon=True)
    _thread.start()
    return True


def stop() -> None:
    """Signal the worker to stop and wait briefly for it. For a clean shutdown and for tests."""
    _stop.set()
    if _thread is not None:
        _thread.join(timeout=5.0)