feat(runner): add pull-dispatch coordination layer

runner/__init__.py

@@ -0,0 +1,5 @@
+"""Pull-based runner coordination layer.
+
+Registration, heartbeat, HTTP client and active-job tracking used by the
+runner process to talk to the backend runner API. No docker, no Flask.
+"""

runner/active_jobs.py

@@ -0,0 +1,31 @@
+import threading
+
+
+class ActiveJobTracker:
+    """Thread-safe set of the job ids this runner currently holds.
+
+    The snapshot is sent with every heartbeat to renew the backend leases
+    (spec §7.2); jobs absent from the snapshot stop being renewed and become
+    orphans once their lease expires.
+    """
+
+    def __init__(self):
+        self._lock = threading.Lock()
+        self._jobs = set()
+
+    def add(self, job_id):
+        with self._lock:
+            self._jobs.add(job_id)
+
+    def remove(self, job_id):
+        # Idempotent: removing an absent id is a no-op.
+        with self._lock:
+            self._jobs.discard(job_id)
+
+    def snapshot(self):
+        with self._lock:
+            return list(self._jobs)
+
+    def __len__(self):
+        with self._lock:
+            return len(self._jobs)

runner/client.py

@@ -0,0 +1,116 @@
+import logging
+from dataclasses import dataclass, field
+
+import requests
+
+from .config import (
+    REQUEST_TIMEOUT,
+    DEFAULT_HEARTBEAT_INTERVAL_SEC,
+    DEFAULT_POLL_INTERVAL_SEC,
+    DEFAULT_MAX_CONCURRENT_JOBS,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class BackendAPIError(Exception):
+    """A backend runner API call returned an unexpected non-2xx status."""
+
+    def __init__(self, message, status_code):
+        super().__init__(message)
+        self.status_code = status_code
+
+
+class BackendAuthError(BackendAPIError):
+    """The backend rejected our credentials (HTTP 401)."""
+
+
+@dataclass
+class RunnerConfig:
+    heartbeat_interval_sec: int
+    poll_interval_sec: int
+    max_concurrent_jobs: int
+
+    @classmethod
+    def from_dict(cls, data):
+        data = data or {}
+        return cls(
+            heartbeat_interval_sec=data.get(
+                'heartbeat_interval_sec',
+                DEFAULT_HEARTBEAT_INTERVAL_SEC,
+            ),
+            poll_interval_sec=data.get(
+                'poll_interval_sec',
+                DEFAULT_POLL_INTERVAL_SEC,
+            ),
+            max_concurrent_jobs=data.get(
+                'max_concurrent_jobs',
+                DEFAULT_MAX_CONCURRENT_JOBS,
+            ),
+        )
+
+
+@dataclass
+class RunnerIdentity:
+    runner_id: str
+    # Memory-only credential (ADR-0004); never leak it via repr.
+    token: str = field(repr=False)
+    config: RunnerConfig
+
+
+class BackendClient:
+    """HTTP client for the backend runner API (spec §7)."""
+
+    def __init__(self, base_url, session=None, timeout=REQUEST_TIMEOUT):
+        self.base_url = base_url.rstrip('/')
+        self.session = session if session is not None else requests.Session()
+        self.timeout = timeout
+
+    def register(self, registration_token, name):
+        """POST /runners/register -> RunnerIdentity.
+
+        201 -> parsed identity; 401 -> BackendAuthError; other non-2xx ->
+        BackendAPIError. Network errors propagate as requests.RequestException.
+        """
+        resp = self.session.post(
+            f'{self.base_url}/runners/register',
+            json={
+                'registration_token': registration_token,
+                'name': name,
+            },
+            timeout=self.timeout,
+        )
+        if resp.status_code == 201:
+            body = resp.json()
+            return RunnerIdentity(
+                runner_id=body['runner_id'],
+                token=body['token'],
+                config=RunnerConfig.from_dict(body.get('config')),
+            )
+        self._raise_for_status(resp, 'register')
+
+    def heartbeat(self, identity, active_job_ids):
+        """POST /runners/<rn>/heartbeat. Expect 204."""
+        resp = self.session.post(
+            f'{self.base_url}/runners/{identity.runner_id}/heartbeat',
+            json={'active_job_ids': list(active_job_ids)},
+            headers={'Authorization': f'Bearer {identity.token}'},
+            timeout=self.timeout,
+        )
+        if resp.status_code == 204:
+            return
+        self._raise_for_status(resp, 'heartbeat')
+
+    @staticmethod
+    def _raise_for_status(resp, action):
+        # Never include tokens in the message: only the status code is logged.
+        status = resp.status_code
+        if status == 401:
+            raise BackendAuthError(
+                f'runner {action} rejected with 401',
+                status_code=status,
+            )
+        raise BackendAPIError(
+            f'runner {action} failed with status {status}',
+            status_code=status,
+        )

runner/config.py

@@ -0,0 +1,29 @@
+import os
+import socket
+
+# Backend base URL for the pull-based runner API. Coexists with the legacy
+# BACKEND_API (dispatcher/config.py) until the keystone slice removes the
+# old push path.
+BACKEND_URL = os.getenv(
+    'BACKEND_URL',
+    'http://web:8080',
+)
+# Shared secret presented to POST /runners/register.
+RUNNER_REGISTRATION_TOKEN = os.getenv(
+    'RUNNER_REGISTRATION_TOKEN',
+    '',
+)
+# Stable, human-readable name for logs / admin display.
+RUNNER_NAME = os.getenv('RUNNER_NAME') or socket.gethostname()
+
+# Registration retry backoff (seconds). After the schedule is exhausted the
+# caller stays at the final value (30s) forever.
+REGISTRATION_BACKOFF_SCHEDULE = (1, 2, 4, 8, 16, 30)
+
+# HTTP timeout (seconds) applied to every backend client call.
+REQUEST_TIMEOUT = 10
+
+# Fallback runner config values used when the register response omits a key.
+DEFAULT_HEARTBEAT_INTERVAL_SEC = 15
+DEFAULT_POLL_INTERVAL_SEC = 3
+DEFAULT_MAX_CONCURRENT_JOBS = 8

runner/heartbeat.py

@@ -0,0 +1,71 @@
+import logging
+import threading
+import time
+
+import requests
+
+from .client import BackendAPIError, BackendAuthError
+
+logger = logging.getLogger(__name__)
+
+# Consecutive 401s that trigger fail-fast (spec §10).
+_FATAL_AUTH_FAILURES = 2
+
+
+class HeartbeatThread(threading.Thread):
+    """Periodically POST /runners/<rn>/heartbeat to renew leases.
+
+    Sends the first beat immediately, then every ``interval_sec``. Two
+    consecutive 401s trigger ``on_fatal`` (the future main.py wires this to
+    process exit) and stop the loop. Any non-401 outcome resets the counter.
+    """
+
+    def __init__(self, client, identity, tracker, on_fatal, interval_sec=None):
+        super().__init__(daemon=True)
+        self._client = client
+        self._identity = identity
+        self._tracker = tracker
+        self._on_fatal = on_fatal
+        self._interval_sec = (interval_sec if interval_sec is not None else
+                              identity.config.heartbeat_interval_sec)
+        self._stop_event = threading.Event()
+        self._consecutive_auth_failures = 0
+
+    def run(self):
+        while not self._stop_event.is_set():
+            started = time.monotonic()
+            fatal = self._beat()
+            if fatal:
+                break
+            # Fixed cadence: subtract the time the beat took so a slow or
+            # timed-out request cannot push the next beat past the lease
+            # budget (15s interval vs 30s TTL allows exactly one miss).
+            # Interruptible wait so stop() takes effect immediately.
+            elapsed = time.monotonic() - started
+            self._stop_event.wait(max(0.0, self._interval_sec - elapsed))
+
+    def _beat(self):
+        """Send one heartbeat. Returns True iff fail-fast was triggered."""
+        try:
+            self._client.heartbeat(self._identity, self._tracker.snapshot())
+        except BackendAuthError:
+            self._consecutive_auth_failures += 1
+            if self._consecutive_auth_failures >= _FATAL_AUTH_FAILURES:
+                logger.error(
+                    'heartbeat got %d consecutive 401s; failing fast',
+                    self._consecutive_auth_failures,
+                )
+                self._on_fatal()
+                return True
+            logger.warning('heartbeat got 401 (%d consecutive)',
+                           self._consecutive_auth_failures)
+            return False
+        except (BackendAPIError, requests.RequestException) as err:
+            self._consecutive_auth_failures = 0
+            logger.warning('heartbeat failed: %s', err)
+            return False
+        self._consecutive_auth_failures = 0
+        return False
+
+    def stop(self):
+        self._stop_event.set()

runner/registration.py

@@ -0,0 +1,43 @@
+import logging
+import time
+
+import requests
+
+from .client import BackendAPIError, BackendAuthError
+from .config import REGISTRATION_BACKOFF_SCHEDULE
+
+logger = logging.getLogger(__name__)
+
+
+def register_with_backoff(
+    client,
+    registration_token,
+    name,
+    *,
+    schedule=REGISTRATION_BACKOFF_SCHEDULE,
+    sleep=time.sleep,
+):
+    """Register with the backend, retrying transient failures indefinitely.
+
+    Backoff follows ``schedule`` (1->2->4->8->16->30s) and then stays at the
+    final value forever. A 401 (BackendAuthError) is fatal per ADR-0004: it is
+    re-raised immediately so the caller can fail-fast. ``sleep`` is injectable
+    for tests.
+    """
+    attempt = 0
+    while True:
+        try:
+            return client.register(registration_token, name)
+        except BackendAuthError:
+            # Bad registration token: no amount of retrying will help.
+            raise
+        except (BackendAPIError, requests.RequestException) as err:
+            delay = schedule[min(attempt, len(schedule) - 1)]
+            logger.warning(
+                'runner registration failed (attempt %d): %s; retrying in %ds',
+                attempt + 1,
+                err,
+                delay,
+            )
+            attempt += 1
+            sleep(delay)

tests/test_active_jobs.py

@@ -0,0 +1,71 @@
+import threading
+
+from runner.active_jobs import ActiveJobTracker
+
+
+def test_add_and_snapshot():
+    tracker = ActiveJobTracker()
+    tracker.add('jb_1')
+    tracker.add('jb_2')
+
+    assert sorted(tracker.snapshot()) == ['jb_1', 'jb_2']
+    assert len(tracker) == 2
+
+
+def test_add_is_idempotent():
+    tracker = ActiveJobTracker()
+    tracker.add('jb_1')
+    tracker.add('jb_1')
+
+    assert tracker.snapshot() == ['jb_1']
+    assert len(tracker) == 1
+
+
+def test_remove():
+    tracker = ActiveJobTracker()
+    tracker.add('jb_1')
+    tracker.add('jb_2')
+    tracker.remove('jb_1')
+
+    assert tracker.snapshot() == ['jb_2']
+    assert len(tracker) == 1
+
+
+def test_remove_absent_is_noop():
+    tracker = ActiveJobTracker()
+    tracker.add('jb_1')
+    tracker.remove('jb_does_not_exist')
+
+    assert tracker.snapshot() == ['jb_1']
+    assert len(tracker) == 1
+
+
+def test_snapshot_is_a_copy():
+    tracker = ActiveJobTracker()
+    tracker.add('jb_1')
+    snap = tracker.snapshot()
+    snap.append('jb_mutated')
+
+    assert tracker.snapshot() == ['jb_1']
+
+
+def test_thread_safety_smoke():
+    tracker = ActiveJobTracker()
+    n = 200
+
+    def worker(base):
+        for i in range(n):
+            job_id = f'{base}_{i}'
+            tracker.add(job_id)
+            tracker.remove(job_id)
+
+    threads = [threading.Thread(target=worker, args=[b]) for b in 'abcd']
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    # Every add was paired with a remove, so the tracker ends empty and did
+    # not raise (e.g. "set changed size during iteration") under contention.
+    assert len(tracker) == 0
+    assert tracker.snapshot() == []