Add a watchdog that repairs missed schedules and stalled jobs

Every daily and weekly source missed two days in November 2026: Procrastinate
skips a periodic slot that is more than ten minutes overdue, and jobs of
killed workers stayed "doing" forever. Every ten minutes the watchdog fails
jobs of dead workers, closes abandoned run records and defers any scheduled
task whose last slot passed without a run.

Also: EP vote references of joint motions and Council documents, and the
loader now refreshes a vote's reference and stage.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

backend/politrack/core/loader.py

@@ -254,6 +254,7 @@ class Loader:
                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
                ON CONFLICT (event_key) DO UPDATE SET result = EXCLUDED.result, raw_text = EXCLUDED.raw_text,
                    roll_call = EXCLUDED.roll_call, item_title = EXCLUDED.item_title, part_vote = EXCLUDED.part_vote,
+                   item_ref = EXCLUDED.item_ref, stage = EXCLUDED.stage,
                    protocol_url = COALESCE(EXCLUDED.protocol_url, vote_event.protocol_url)
                RETURNING id""",
             (rec.key, chamber, rec.item_ref, rec.title, rec.stage, rec.voted_on, rec.result, rec.roll_call, rec.url,

backend/politrack/sources/eu.py

@@ -200,7 +200,7 @@ class EuropeanParliament(Adapter):
 
 LD_JSON = "application/ld+json"
 EP_DECISIONS_PAGE = 500
-EP_REFERENCE = re.compile(r"^[AB]\d+-\d+/\d{4}$")
+EP_REFERENCE = re.compile(r"^(?:RC-)?[A-C]\d+-\d+/\d{4}$")  # A10-0272/2026, RC-B9-0253/2024, C9-0010/2024
 EP_OUTCOME = {"ADOPTED": ("accepted", "angenommen", "adopted"), "REJECTED": ("rejected", "abgelehnt", "rejected")}
 EP_VOTERS = (("had_voter_favor", "for", "Stimmte für", "Voted for"),
              ("had_voter_against", "against", "Stimmte gegen", "Voted against"),
@@ -233,7 +233,7 @@ class ParliamentVotes(Adapter):
     description = "Roll-call votes of Austrian MEPs in the European Parliament plenary, including amendments"
     schedule = "45 5 * * *"
     queue = "scrape"
-    parser_version = 2  # amendment and part votes
+    parser_version = 3  # 2: amendment and part votes; 3: references of joint motions and Council documents
 
     def start(self, ctx: RunContext) -> Iterator[Request]:
         ep_ids = []

backend/politrack/worker.py

@@ -9,21 +9,38 @@ One periodic task is generated per registered adapter/enricher from its `schedul
 
 Periodic tasks are deferred by every running worker and de-duplicated by Procrastinate; queueing locks
 prevent two runs of the same adapter at once. Workers can run on other machines that reach the DB.
+
+Procrastinate only remembers in memory when it last deferred a task, and skips a slot that is more than
+ten minutes overdue. In November 2026 every daily and weekly source silently missed two days that way,
+and jobs of killed workers stayed "doing" forever. The watchdog below makes both self-healing.
 """
 
 from __future__ import annotations
 
+import asyncio
 import logging
+from datetime import datetime, timedelta, timezone
 
+from croniter import croniter
 from procrastinate import App, PsycopgConnector
+from procrastinate.exceptions import AlreadyEnqueued
+from procrastinate.jobs import Status
 
 from politrack.config import get_settings
 from politrack.core import registry
 from politrack.core.runner import run
+from politrack.db import connection
 
 logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
+log = logging.getLogger("politrack.watchdog")
 
 app = App(connector=PsycopgConnector(conninfo=get_settings().database_url))
+TASKS: dict = {}
+SCHEDULES: dict[str, str] = {}
+
+STALLED_AFTER = 300                       # seconds without a worker heartbeat
+OVERDUE_GRACE = timedelta(minutes=15)     # normal deferral happens within a minute of the slot
+ABANDONED_RUN = timedelta(hours=6)        # CLI runs have no job; don't close them while they may still run
 
 
 def _make_task(name: str, queue: str, schedule: str | None) -> None:
@@ -32,10 +49,65 @@ def _make_task(name: str, queue: str, schedule: str | None) -> None:
 
     job.__name__ = f"run_{name.replace('.', '_')}"
     task = app.task(name=f"politrack.run.{name}", queue=queue, queueing_lock=name, pass_context=False)(job)
+    TASKS[name] = task
     if schedule:
+        SCHEDULES[name] = schedule
         app.periodic(cron=schedule, periodic_id=name)(task)
 
 
 registry.load_builtin()
 for _name, _cls in {**registry.ADAPTERS, **registry.ENRICHERS}.items():
     _make_task(_name, _cls.queue, _cls.schedule)
+
+
+def overdue_tasks(now: datetime) -> list[str]:
+    """Scheduled tasks whose latest slot passed without a run, a waiting job or a running job."""
+    due: list[str] = []
+    with connection() as conn:
+        for name, schedule in SCHEDULES.items():
+            slot = croniter(schedule, now).get_prev(datetime)
+            if now - slot < OVERDUE_GRACE:
+                continue
+            seen = conn.execute(
+                """SELECT EXISTS (SELECT 1 FROM run WHERE adapter = %(name)s AND started_at >= %(slot)s)
+                       OR EXISTS (SELECT 1 FROM procrastinate_jobs WHERE task_name = %(task)s
+                                  AND status IN ('todo', 'doing')) AS seen""",
+                {"name": name, "slot": slot, "task": f"politrack.run.{name}"},
+            ).fetchone()["seen"]
+            if not seen:
+                due.append(name)
+    return due
+
+
+def close_abandoned_runs() -> int:
+    """Runs still marked running although no job for them is in progress (the worker was stopped)."""
+    with connection() as conn:
+        closed = conn.execute(
+            """UPDATE run r SET status = 'error', finished_at = now(),
+                   error = 'abandoned: the worker stopped before the run finished (closed by the watchdog)'
+               WHERE r.status = 'running' AND r.started_at < now() - %s
+                 AND NOT EXISTS (SELECT 1 FROM procrastinate_jobs j
+                                 WHERE j.status = 'doing' AND j.task_name = 'politrack.run.' || r.adapter)""",
+            (ABANDONED_RUN,),
+        ).rowcount
+        conn.commit()
+    return closed
+
+
+@app.periodic(cron="*/10 * * * *", periodic_id="watchdog")
+@app.task(name="politrack.watchdog", queue="social", queueing_lock="watchdog", pass_context=False)
+async def watchdog(timestamp: int | None = None) -> dict:
+    stats = {"stalled_failed": 0, "overdue_deferred": 0}
+    for job in await app.job_manager.get_stalled_jobs(seconds_since_heartbeat=STALLED_AFTER):
+        await app.job_manager.finish_job(job, status=Status.FAILED, delete_job=False)
+        stats["stalled_failed"] += 1
+        log.warning("failed stalled job %s (%s): its worker is gone", job.id, job.task_name)
+    stats["abandoned_runs_closed"] = await asyncio.to_thread(close_abandoned_runs)
+    for name in await asyncio.to_thread(overdue_tasks, datetime.now(timezone.utc)):
+        try:
+            await TASKS[name].defer_async()
+            stats["overdue_deferred"] += 1
+            log.warning("deferred %s: its last scheduled slot passed without a run", name)
+        except AlreadyEnqueued:
+            pass
+    return stats

backend/pyproject.toml

@@ -17,6 +17,7 @@ dependencies = [
     "python-slugify>=8.0",
     "rapidfuzz>=3.9",
     "procrastinate>=2.14,<4",
+    "croniter>=2.0",
     "typer>=0.12",
     "pyyaml>=6.0",
     "mcp>=1.9,<2",

backend/tests/Dockerfile.debian

@@ -13,7 +13,7 @@ RUN apt-get update -qq && apt-get install -y -qq --no-install-recommends eatmyda
     && eatmydata apt-get install -y -qq --no-install-recommends \
         python3 python3-psycopg python3-psycopg-pool python3-pydantic-settings python3-fastapi python3-httpx \
         python3-protego python3-feedparser python3-bs4 python3-slugify python3-rapidfuzz python3-typer \
-        python3-yaml python3-uvicorn python3-pytest python3-openpyxl ca-certificates >/dev/null \
+        python3-yaml python3-uvicorn python3-pytest python3-openpyxl python3-croniter ca-certificates >/dev/null \
     && rm -rf /var/lib/apt/lists/*
 
 # Keeps the agent image's user, entrypoint and /workspace as they are.