From 297749b27cda31ff41d59ca459abb73486765e8f Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Tue, 14 Jul 2026 01:09:43 +0800
Subject: [PATCH 1/9] feat(dispatch): add runner registry foundation (register,
 tokens, GC)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Slice 1/6 of the pull-based sandbox runner rework (Normal-OJ/Normal-OJ#61).
Dark PR: no callers yet.

- dispatch/redis_keys.py: centralized spec-§8 key namespace
- dispatch/config.py: spec-§13 parameters + fail-closed registration token accessor
- dispatch/runner.py: register (rn_/rk_, SHA-256 hash only), constant-time
  verify_token / verify_registration_token, lazy 7d identity GC, list_runners
- tests: 19 fakeredis unit tests incl. revocation and non-ASCII auth inputs
---
 dispatch/__init__.py          |   6 +
 dispatch/config.py            |  30 +++++
 dispatch/redis_keys.py        |  44 +++++++
 dispatch/runner.py            | 183 ++++++++++++++++++++++++++
 tests/test_dispatch_runner.py | 233 ++++++++++++++++++++++++++++++++++
 5 files changed, 496 insertions(+)
 create mode 100644 dispatch/__init__.py
 create mode 100644 dispatch/config.py
 create mode 100644 dispatch/redis_keys.py
 create mode 100644 dispatch/runner.py
 create mode 100644 tests/test_dispatch_runner.py

diff --git a/dispatch/__init__.py b/dispatch/__init__.py
new file mode 100644
index 0000000..845d45a
--- /dev/null
+++ b/dispatch/__init__.py
@@ -0,0 +1,6 @@
+"""Redis-based pull dispatch module (spec §11).
+
+This slice ships only the runner-identity foundation (redis_keys, config,
+runner registration / token verification / GC). It has no callers yet —
+the HTTP layer and job lifecycle land in later slices.
+"""
diff --git a/dispatch/config.py b/dispatch/config.py
new file mode 100644
index 0000000..53434de
--- /dev/null
+++ b/dispatch/config.py
@@ -0,0 +1,30 @@
+"""Pull-dispatch parameters (spec §13) and the registration-token accessor."""
+
+import os
+from typing import Optional
+
+# --- §13 parameters -----------------------------------------------------
+HEARTBEAT_INTERVAL_SEC = 15
+LEASE_TTL_SEC = 30
+POLL_INTERVAL_SEC = 3
+ORPHAN_SCAN_INTERVAL_SEC = 15
+MAX_ATTEMPTS = 3
+IDENTITY_TTL_SEC = 7 * 24 * 60 * 60  # 7 days
+PRESIGNED_URL_TTL_SEC = 60 * 60  # 1 hour
+MAX_CONCURRENT_JOBS = 8  # advertised to runners in the register response (§7.1)
+
+_REGISTRATION_TOKEN_ENV = 'RUNNER_REGISTRATION_TOKEN'
+
+
+def registration_token() -> Optional[str]:
+    """The shared runner registration secret, read live from the environment.
+
+    Read at call time (not cached at import) so the value stays consistent
+    with the deployed env and so verification fails closed the moment the
+    secret is removed. Returns ``None`` when unset or empty; callers must
+    treat that as "reject everything" rather than crashing.
+    """
+    token = os.getenv(_REGISTRATION_TOKEN_ENV)
+    if not token:
+        return None
+    return token
diff --git a/dispatch/redis_keys.py b/dispatch/redis_keys.py
new file mode 100644
index 0000000..9cb992b
--- /dev/null
+++ b/dispatch/redis_keys.py
@@ -0,0 +1,44 @@
+"""Centralized Redis key naming for the pull-dispatch namespace (spec §8).
+
+Pure constants and functions — no I/O. Job-related keys are defined now even
+though this slice does not use them, so the whole §8 schema lives in one place.
+"""
+
+# --- identity (soft state; ADR-0004) ------------------------------------
+# ZSET: member=rn_id, score=last heartbeat epoch
+RUNNERS_REGISTERED = 'runners:registered'
+
+# --- job queue ----------------------------------------------------------
+JOBS_PENDING = 'jobs:pending'  # LIST of pending jb_id
+JOBS_LEASED = 'jobs:leased'  # SET of leased jb_id
+DISPATCH_LAST_RECOVERY = 'dispatch:last_recovery'  # STRING time gate (SET NX EX)
+
+
+def runner_meta(runner_id: str) -> str:
+    """HASH {name, registered_at, registration_ip}, TTL 7d."""
+    return f'runner:{runner_id}:meta'
+
+
+def runner_token_hash(runner_id: str) -> str:
+    """STRING SHA-256(rk_token), TTL 7d."""
+    return f'runner:{runner_id}:token_hash'
+
+
+def runner_alive(runner_id: str) -> str:
+    """STRING "1", TTL 30s — monitoring only, never used for decisions."""
+    return f'runner:{runner_id}:alive'
+
+
+def job(job_id: str) -> str:
+    """HASH holding a job's full state."""
+    return f'job:{job_id}'
+
+
+def submission_current_job(submission_id: str) -> str:
+    """STRING currency pointer (INV4)."""
+    return f'submission:{submission_id}:current_job'
+
+
+def submission_job_lock(submission_id: str) -> str:
+    """Per-submission serialization lock (INV3)."""
+    return f'submission:{submission_id}:job_lock'
diff --git a/dispatch/runner.py b/dispatch/runner.py
new file mode 100644
index 0000000..b9a8af7
--- /dev/null
+++ b/dispatch/runner.py
@@ -0,0 +1,183 @@
+"""Runner identity: registration, token verification, and lazy GC (spec §7.1, §8).
+
+Identity is soft state that can be rebuilt from scratch (ADR-0004): the
+``runners:registered`` ZSET tracks last-seen, while meta/token_hash carry a 7d
+TTL. Dead identities evaporate via TTL; the ZSET members (which have no TTL) are
+swept lazily on register / list_runners.
+
+Security notes:
+- The runner token is returned exactly once. Only its SHA-256 hex is stored, so
+  a Redis dump never reveals a usable credential.
+- All token comparisons are constant-time (``hmac.compare_digest``).
+- A missing ``token_hash`` key means the identity was revoked (or expired) →
+  verification fails. This is the single revocation mechanism (ADR-0004).
+- Registration verification fails closed: if the shared secret is unset/empty in
+  the environment, every candidate is rejected rather than accepted or crashing.
+"""
+
+import hashlib
+import hmac
+import secrets
+import time
+from dataclasses import dataclass
+from typing import Dict, List, Optional
+
+from ulid import ULID
+
+from mongo.utils import RedisCache
+from . import config
+from . import redis_keys
+
+RUNNER_ID_PREFIX = 'rn_'
+RUNNER_TOKEN_PREFIX = 'rk_'
+
+# One shared RedisCache instance. In production every RedisCache() shares the
+# pooled real Redis, but under fakeredis each instance gets an isolated dataset;
+# caching one instance keeps register/verify/list coherent in both worlds.
+_cache: Optional[RedisCache] = None
+
+
+@dataclass(frozen=True)
+class Registration:
+    runner_id: str
+    token: str
+
+
+def _redis():
+    global _cache
+    if _cache is None:
+        _cache = RedisCache()
+    return _cache.client
+
+
+def _now() -> float:
+    return time.time()
+
+
+def _token_hash(token: str) -> str:
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+def verify_registration_token(candidate: Optional[str]) -> bool:
+    """Constant-time check of a register request's shared secret. Fails closed."""
+    expected = config.registration_token()
+    # Fail closed: no configured secret ⇒ registration is disabled, not open.
+    if not expected or not candidate:
+        return False
+    # Compare UTF-8 bytes: compare_digest raises TypeError on non-ASCII str, and
+    # `candidate` is attacker-controlled, so str comparison could crash (500)
+    # instead of failing closed (401).
+    return hmac.compare_digest(expected.encode(), candidate.encode())
+
+
+def register(name: str, ip: str) -> Registration:
+    """Mint a fresh runner identity and persist its soft state (spec §7.1).
+
+    Returns the runner_id and the plaintext token (shown once). Only the token's
+    SHA-256 hex is stored. Also sweeps expired identities before registering.
+    """
+    now = _now()
+    _gc(now)
+
+    runner_id = RUNNER_ID_PREFIX + str(ULID())
+    token = RUNNER_TOKEN_PREFIX + secrets.token_urlsafe(32)
+
+    client = _redis()
+    ttl = config.IDENTITY_TTL_SEC
+    meta_key = redis_keys.runner_meta(runner_id)
+
+    pipe = client.pipeline()
+    pipe.zadd(redis_keys.RUNNERS_REGISTERED, {runner_id: now})
+    pipe.hset(
+        meta_key,
+        mapping={
+            'name': name,
+            'registered_at': repr(now),
+            'registration_ip': ip,
+        },
+    )
+    pipe.expire(meta_key, ttl)
+    pipe.set(redis_keys.runner_token_hash(runner_id),
+             _token_hash(token),
+             ex=ttl)
+    pipe.execute()
+
+    return Registration(runner_id=runner_id, token=token)
+
+
+def verify_token(runner_id: Optional[str], token: Optional[str]) -> bool:
+    """Constant-time check that ``token`` matches the stored hash for ``runner_id``.
+
+    A missing token_hash key (revoked or expired) yields ``False`` (→ 401).
+    """
+    if not runner_id or not token:
+        return False
+    stored = _redis().get(redis_keys.runner_token_hash(runner_id))
+    if stored is None:
+        return False
+    stored_hex = stored.decode()
+    return hmac.compare_digest(stored_hex, _token_hash(token))
+
+
+def list_runners() -> List[Dict]:
+    """Return identity-layer facts for all live runners (spec §7.6 subset).
+
+    Sweeps expired identities first. Fields: runner_id, name, last_seen,
+    registered_at. Liveness/held-jobs are added by the admin-API slice.
+    """
+    now = _now()
+    _gc(now)
+
+    client = _redis()
+    members = client.zrange(redis_keys.RUNNERS_REGISTERED,
+                            0,
+                            -1,
+                            withscores=True)
+
+    runners: List[Dict] = []
+    for member, score in members:
+        runner_id = member.decode() if isinstance(member, bytes) else member
+        raw_meta = client.hgetall(redis_keys.runner_meta(runner_id))
+        meta = {
+            (k.decode() if isinstance(k, bytes) else k):
+            (v.decode() if isinstance(v, bytes) else v)
+            for k, v in raw_meta.items()
+        }
+        runners.append({
+            'runner_id': runner_id,
+            'name': meta.get('name'),
+            'last_seen': score,
+            'registered_at': meta.get('registered_at'),
+        })
+    return runners
+
+
+def _gc(now: Optional[float] = None) -> None:
+    """Evict identities whose last-seen is older than the 7d TTL.
+
+    Removes the ZSET member (which carries no TTL of its own) plus the meta,
+    token_hash and alive keys. Meta/token_hash usually expire on their own; the
+    explicit deletes keep the namespace clean even if timing drifts.
+    """
+    if now is None:
+        now = _now()
+    cutoff = now - config.IDENTITY_TTL_SEC
+
+    client = _redis()
+    # Strictly older than the cutoff: '(' makes the max bound exclusive.
+    expired = client.zrangebyscore(
+        redis_keys.RUNNERS_REGISTERED,
+        '-inf',
+        f'({cutoff!r}',
+    )
+    if not expired:
+        return
+
+    pipe = client.pipeline()
+    for member in expired:
+        runner_id = member.decode() if isinstance(member, bytes) else member
+        pipe.zrem(redis_keys.RUNNERS_REGISTERED, runner_id)
+        pipe.delete(redis_keys.runner_meta(runner_id))
+        pipe.delete(redis_keys.runner_token_hash(runner_id))
+        pipe.delete(redis_keys.runner_alive(runner_id))
+    pipe.execute()
diff --git a/tests/test_dispatch_runner.py b/tests/test_dispatch_runner.py
new file mode 100644
index 0000000..0960114
--- /dev/null
+++ b/tests/test_dispatch_runner.py
@@ -0,0 +1,233 @@
+import hashlib
+import os
+
+import pytest
+
+from mongo.utils import RedisCache
+from dispatch import config, redis_keys, runner
+
+REG_TOKEN_ENV = 'RUNNER_REGISTRATION_TOKEN'
+
+
+@pytest.fixture(autouse=True, scope='session')
+def setup_minio():
+    # Shadow conftest's Docker/MinIO session fixture: these identity-layer unit
+    # tests touch only fakeredis, so they must not require a container engine.
+    yield
+
+
+def _reset():
+    # Force each test onto a fresh FakeStrictRedis: RedisCache caches its
+    # connection pool on the class and dispatch caches one RedisCache instance.
+    RedisCache.POOL = None
+    runner._cache = None
+
+
+def setup_function(_):
+    # REDIS_PORT must be unset so RedisCache falls back to fakeredis.
+    os.environ.pop('REDIS_PORT', None)
+    _reset()
+
+
+def teardown_function(_):
+    _reset()
+    os.environ.pop(REG_TOKEN_ENV, None)
+
+
+def _sha256_hex(text: str) -> str:
+    return hashlib.sha256(text.encode()).hexdigest()
+
+
+# --- registration token -------------------------------------------------
+
+
+def test_verify_registration_token_accepts_correct(monkeypatch):
+    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    assert runner.verify_registration_token('super-secret') is True
+
+
+def test_verify_registration_token_rejects_wrong(monkeypatch):
+    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    assert runner.verify_registration_token('nope') is False
+    assert runner.verify_registration_token('') is False
+    assert runner.verify_registration_token(None) is False
+
+
+def test_verify_registration_token_unset_env_always_rejected(monkeypatch):
+    monkeypatch.delenv(REG_TOKEN_ENV, raising=False)
+    assert runner.verify_registration_token('anything') is False
+    assert runner.verify_registration_token('') is False
+
+
+def test_verify_registration_token_empty_env_always_rejected(monkeypatch):
+    monkeypatch.setenv(REG_TOKEN_ENV, '')
+    assert runner.verify_registration_token('') is False
+    assert runner.verify_registration_token('anything') is False
+
+
+def test_verify_registration_token_non_ascii_candidate_rejected(monkeypatch):
+    # candidate is attacker-controlled (JSON body); non-ASCII must fail closed,
+    # not raise TypeError from hmac.compare_digest on str inputs.
+    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    assert runner.verify_registration_token('sécret-ü') is False
+
+
+def test_verify_registration_token_non_ascii_secret(monkeypatch):
+    monkeypatch.setenv(REG_TOKEN_ENV, 'sécret-ü')
+    assert runner.verify_registration_token('sécret-ü') is True
+    assert runner.verify_registration_token('super-secret') is False
+
+
+# --- register -----------------------------------------------------------
+
+
+def test_register_creates_zset_meta_and_token_hash(monkeypatch):
+    now = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: now)
+
+    reg = runner.register('runner-ec2-1', '10.0.0.7')
+
+    assert reg.runner_id.startswith('rn_')
+    assert reg.token.startswith('rk_')
+
+    client = runner._redis()
+    # ZSET member with score == registration time
+    assert client.zscore(redis_keys.RUNNERS_REGISTERED, reg.runner_id) == now
+
+    # meta hash
+    meta = client.hgetall(redis_keys.runner_meta(reg.runner_id))
+    meta = {k.decode(): v.decode() for k, v in meta.items()}
+    assert meta['name'] == 'runner-ec2-1'
+    assert meta['registration_ip'] == '10.0.0.7'
+    assert float(meta['registered_at']) == now
+
+    # token_hash stores only the SHA-256 of the token, never the token itself
+    stored = client.get(redis_keys.runner_token_hash(reg.runner_id))
+    assert stored.decode() == _sha256_hex(reg.token)
+    assert reg.token not in stored.decode()
+
+    # 7d TTLs on meta and token_hash
+    ttl = config.IDENTITY_TTL_SEC
+    assert ttl - 5 <= client.ttl(redis_keys.runner_meta(reg.runner_id)) <= ttl
+    assert ttl - 5 <= client.ttl(redis_keys.runner_token_hash(
+        reg.runner_id)) <= ttl
+
+
+def test_register_generates_unique_ids():
+    a = runner.register('a', '1.1.1.1')
+    b = runner.register('b', '2.2.2.2')
+    assert a.runner_id != b.runner_id
+    assert a.token != b.token
+
+
+# --- verify_token -------------------------------------------------------
+
+
+def test_issued_token_verifies():
+    reg = runner.register('r', '1.1.1.1')
+    assert runner.verify_token(reg.runner_id, reg.token) is True
+
+
+def test_wrong_token_fails():
+    reg = runner.register('r', '1.1.1.1')
+    assert runner.verify_token(reg.runner_id, 'rk_wrong') is False
+
+
+def test_token_for_different_runner_fails():
+    a = runner.register('a', '1.1.1.1')
+    b = runner.register('b', '2.2.2.2')
+    assert runner.verify_token(a.runner_id, b.token) is False
+    assert runner.verify_token(b.runner_id, a.token) is False
+
+
+def test_revocation_deletes_token_hash_then_401():
+    reg = runner.register('r', '1.1.1.1')
+    assert runner.verify_token(reg.runner_id, reg.token) is True
+    # revoke: delete the token key
+    runner._redis().delete(redis_keys.runner_token_hash(reg.runner_id))
+    assert runner.verify_token(reg.runner_id, reg.token) is False
+
+
+def test_verify_token_unknown_runner_fails():
+    assert runner.verify_token('rn_does_not_exist', 'rk_whatever') is False
+
+
+def test_verify_token_missing_args_fails():
+    reg = runner.register('r', '1.1.1.1')
+    assert runner.verify_token('', '') is False
+    assert runner.verify_token(reg.runner_id, '') is False
+    assert runner.verify_token(None, None) is False
+
+
+# --- lazy GC ------------------------------------------------------------
+
+
+def test_gc_on_register_evicts_expired_identity(monkeypatch):
+    t0 = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: t0)
+    old = runner.register('old', '1.1.1.1')
+
+    # 8 days later a new registration triggers GC of the stale identity
+    monkeypatch.setattr(runner, '_now', lambda: t0 + 8 * 86400)
+    fresh = runner.register('fresh', '2.2.2.2')
+
+    client = runner._redis()
+    assert client.zscore(redis_keys.RUNNERS_REGISTERED, old.runner_id) is None
+    assert client.exists(redis_keys.runner_meta(old.runner_id)) == 0
+    assert client.exists(redis_keys.runner_token_hash(old.runner_id)) == 0
+    # fresh identity untouched
+    assert client.zscore(redis_keys.RUNNERS_REGISTERED,
+                         fresh.runner_id) is not None
+
+
+def test_gc_on_list_runners_evicts_expired_identity(monkeypatch):
+    t0 = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: t0)
+    old = runner.register('old', '1.1.1.1')
+
+    monkeypatch.setattr(runner, '_now', lambda: t0 + 8 * 86400)
+    listed = runner.list_runners()
+
+    assert listed == []
+    client = runner._redis()
+    assert client.zscore(redis_keys.RUNNERS_REGISTERED, old.runner_id) is None
+    assert client.exists(redis_keys.runner_meta(old.runner_id)) == 0
+    assert client.exists(redis_keys.runner_token_hash(old.runner_id)) == 0
+
+
+def test_fresh_identity_survives_gc(monkeypatch):
+    t0 = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: t0)
+    a = runner.register('a', '1.1.1.1')
+
+    # one day later — well within the 7d TTL
+    monkeypatch.setattr(runner, '_now', lambda: t0 + 86400)
+    b = runner.register('b', '2.2.2.2')
+
+    ids = {r['runner_id'] for r in runner.list_runners()}
+    assert a.runner_id in ids
+    assert b.runner_id in ids
+
+
+# --- list_runners -------------------------------------------------------
+
+
+def test_list_runners_returns_identity_fields(monkeypatch):
+    now = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: now)
+    reg = runner.register('runner-ec2-1', '10.0.0.7')
+
+    listed = runner.list_runners()
+    assert len(listed) == 1
+    entry = listed[0]
+    assert entry['runner_id'] == reg.runner_id
+    assert entry['name'] == 'runner-ec2-1'
+    assert entry['last_seen'] == now
+    assert float(entry['registered_at']) == now
+    # identity layer must not leak secrets
+    assert 'token' not in entry
+    assert 'token_hash' not in entry
+
+
+def test_list_runners_empty():
+    assert runner.list_runners() == []

From 8e78c34013a4d61ae880d414ff8659d20427f21c Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Tue, 14 Jul 2026 01:18:35 +0800
Subject: [PATCH 2/9] fix(dispatch): harden runner auth type handling at trust
 boundary

verify_token and verify_registration_token now reject non-str inputs
(bytes/int/list/dict) with False instead of raising AttributeError on
.encode(). JSON request bodies can legally carry non-str values, so the
auth boundary must fail closed (401) rather than crash (500).

Add regression tests covering bytes/int/list/dict for both functions.
---
 dispatch/runner.py            |  8 +++++++-
 tests/test_dispatch_runner.py | 16 ++++++++++++++++
 2 files changed, 23 insertions(+), 1 deletion(-)

diff --git a/dispatch/runner.py b/dispatch/runner.py
index b9a8af7..028e7b3 100644
--- a/dispatch/runner.py
+++ b/dispatch/runner.py
@@ -62,7 +62,9 @@ def verify_registration_token(candidate: Optional[str]) -> bool:
     """Constant-time check of a register request's shared secret. Fails closed."""
     expected = config.registration_token()
     # Fail closed: no configured secret ⇒ registration is disabled, not open.
-    if not expected or not candidate:
+    # Reject non-str candidates too — a JSON body can carry ints/lists/bytes,
+    # and .encode() below would otherwise raise instead of returning False.
+    if not expected or not isinstance(candidate, str) or not candidate:
         return False
     # Compare UTF-8 bytes: compare_digest raises TypeError on non-ASCII str, and
     # `candidate` is attacker-controlled, so str comparison could crash (500)
@@ -110,6 +112,10 @@ def verify_token(runner_id: Optional[str], token: Optional[str]) -> bool:
 
     A missing token_hash key (revoked or expired) yields ``False`` (→ 401).
     """
+    # Reject non-str inputs from the trust boundary: runner_id feeds a Redis key
+    # and token feeds _token_hash().encode() — both would otherwise raise.
+    if not isinstance(runner_id, str) or not isinstance(token, str):
+        return False
     if not runner_id or not token:
         return False
     stored = _redis().get(redis_keys.runner_token_hash(runner_id))
diff --git a/tests/test_dispatch_runner.py b/tests/test_dispatch_runner.py
index 0960114..ce546a1 100644
--- a/tests/test_dispatch_runner.py
+++ b/tests/test_dispatch_runner.py
@@ -78,6 +78,14 @@ def test_verify_registration_token_non_ascii_secret(monkeypatch):
     assert runner.verify_registration_token('super-secret') is False
 
 
+@pytest.mark.parametrize('candidate', [b'rk_bytes', 12345, ['x'], {'a': 1}])
+def test_verify_registration_token_non_str_candidate_rejected(
+        monkeypatch, candidate):
+    # JSON bodies can legally carry non-str values; must fail closed, not raise.
+    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    assert runner.verify_registration_token(candidate) is False
+
+
 # --- register -----------------------------------------------------------
 
 
@@ -159,6 +167,14 @@ def test_verify_token_missing_args_fails():
     assert runner.verify_token(None, None) is False
 
 
+@pytest.mark.parametrize('bad', [b'rk_bytes', 12345, ['x'], {'a': 1}])
+def test_verify_token_non_str_inputs_rejected(bad):
+    # Non-str token or runner_id from a JSON body must fail closed, not raise.
+    reg = runner.register('r', '1.1.1.1')
+    assert runner.verify_token(reg.runner_id, bad) is False
+    assert runner.verify_token(bad, reg.token) is False
+
+
 # --- lazy GC ------------------------------------------------------------
 
 

From 66cfb11e2c932e94cb92442537dacd0c2e1b4023 Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Wed, 15 Jul 2026 00:13:21 +0800
Subject: [PATCH 3/9] fix(dispatch): make TTL the sole identity invalidator; GC
 only sweeps evaporated members

External review on PR #341 found a TOCTOU in _gc(): it did zrangebyscore
then unconditionally deleted meta/token_hash/alive + zrem. A heartbeat (or
clock skew) landing between the scan and the delete could get its
just-renewed token_hash deleted, turning a live runner into a 401.

Close the race structurally instead of with atomicity machinery: TTL is
now the ONLY thing that invalidates a live identity. _gc() keeps the >7d
score prefilter, then EXISTS-checks each candidate's token_hash and skips
any that still exists (its TTL has not fired). Only members whose
token_hash already evaporated are swept (zrem + delete meta/alive). Safe
without locks because a token_hash can never reappear for the same rn_id:
register always mints a fresh ULID and heartbeat requires token auth.

Adapt GC tests to simulate the token_hash TTL firing by deleting the key
(fakeredis TTLs use real wall-clock, not the monkeypatched _now), and add
a regression test asserting a stale-by-score member with a surviving
token_hash is spared and still verifies.
---
 dispatch/runner.py            | 50 ++++++++++++++++++++++++++---------
 tests/test_dispatch_runner.py | 31 ++++++++++++++++++++--
 2 files changed, 67 insertions(+), 14 deletions(-)

diff --git a/dispatch/runner.py b/dispatch/runner.py
index 028e7b3..68c451e 100644
--- a/dispatch/runner.py
+++ b/dispatch/runner.py
@@ -2,8 +2,11 @@
 
 Identity is soft state that can be rebuilt from scratch (ADR-0004): the
 ``runners:registered`` ZSET tracks last-seen, while meta/token_hash carry a 7d
-TTL. Dead identities evaporate via TTL; the ZSET members (which have no TTL) are
-swept lazily on register / list_runners.
+TTL. TTL is the *only* thing that invalidates a live identity. GC never kills an
+identity — it lazily sweeps the corpses (ZSET members have no TTL of their own)
+left behind once ``token_hash`` has already evaporated, so a heartbeat that
+renews an identity between GC's scan and sweep can never lose its just-renewed
+token (the TOCTOU is structurally impossible, no atomicity machinery needed).
 
 Security notes:
 - The runner token is returned exactly once. Only its SHA-256 hex is stored, so
@@ -159,11 +162,18 @@ def list_runners() -> List[Dict]:
 
 
 def _gc(now: Optional[float] = None) -> None:
-    """Evict identities whose last-seen is older than the 7d TTL.
-
-    Removes the ZSET member (which carries no TTL of its own) plus the meta,
-    token_hash and alive keys. Meta/token_hash usually expire on their own; the
-    explicit deletes keep the namespace clean even if timing drifts.
+    """Sweep ZSET corpses whose identity keys have already evaporated via TTL.
+
+    TTL is the sole invalidator: GC only removes a member once its
+    ``token_hash`` is already gone. The score prefilter (>7d stale) just narrows
+    the candidate set; any candidate whose ``token_hash`` still exists is skipped
+    untouched — its TTL has not fired, so touching it is exactly what caused the
+    TOCTOU (a heartbeat renewing the key between scan and delete).
+
+    Race-freedom without atomicity: once a ``token_hash`` is gone it can never
+    reappear for the same ``rn_id`` — register always mints a fresh ULID, and
+    heartbeat (future) requires token auth that fails without the key. So an
+    ``EXISTS == 0`` observation stays true, making the sweep safe.
     """
     if now is None:
         now = _now()
@@ -171,19 +181,35 @@ def _gc(now: Optional[float] = None) -> None:
 
     client = _redis()
     # Strictly older than the cutoff: '(' makes the max bound exclusive.
-    expired = client.zrangebyscore(
+    candidates = client.zrangebyscore(
         redis_keys.RUNNERS_REGISTERED,
         '-inf',
         f'({cutoff!r}',
     )
-    if not expired:
+    if not candidates:
+        return
+
+    runner_ids = [
+        member.decode() if isinstance(member, bytes) else member
+        for member in candidates
+    ]
+
+    # Only sweep members whose token_hash has already expired (TTL fired).
+    exists_pipe = client.pipeline()
+    for runner_id in runner_ids:
+        exists_pipe.exists(redis_keys.runner_token_hash(runner_id))
+    still_alive = exists_pipe.execute()
+
+    sweep = [
+        runner_id for runner_id, alive in zip(runner_ids, still_alive)
+        if not alive
+    ]
+    if not sweep:
         return
 
     pipe = client.pipeline()
-    for member in expired:
-        runner_id = member.decode() if isinstance(member, bytes) else member
+    for runner_id in sweep:
         pipe.zrem(redis_keys.RUNNERS_REGISTERED, runner_id)
         pipe.delete(redis_keys.runner_meta(runner_id))
-        pipe.delete(redis_keys.runner_token_hash(runner_id))
         pipe.delete(redis_keys.runner_alive(runner_id))
     pipe.execute()
diff --git a/tests/test_dispatch_runner.py b/tests/test_dispatch_runner.py
index ce546a1..e372770 100644
--- a/tests/test_dispatch_runner.py
+++ b/tests/test_dispatch_runner.py
@@ -183,11 +183,15 @@ def test_gc_on_register_evicts_expired_identity(monkeypatch):
     monkeypatch.setattr(runner, '_now', lambda: t0)
     old = runner.register('old', '1.1.1.1')
 
+    client = runner._redis()
+    # fakeredis TTLs run on real wall-clock, not our monkeypatched _now, so
+    # simulate the 7d token_hash TTL having fired by deleting the key.
+    client.delete(redis_keys.runner_token_hash(old.runner_id))
+
     # 8 days later a new registration triggers GC of the stale identity
     monkeypatch.setattr(runner, '_now', lambda: t0 + 8 * 86400)
     fresh = runner.register('fresh', '2.2.2.2')
 
-    client = runner._redis()
     assert client.zscore(redis_keys.RUNNERS_REGISTERED, old.runner_id) is None
     assert client.exists(redis_keys.runner_meta(old.runner_id)) == 0
     assert client.exists(redis_keys.runner_token_hash(old.runner_id)) == 0
@@ -201,16 +205,39 @@ def test_gc_on_list_runners_evicts_expired_identity(monkeypatch):
     monkeypatch.setattr(runner, '_now', lambda: t0)
     old = runner.register('old', '1.1.1.1')
 
+    client = runner._redis()
+    # Simulate the token_hash TTL having fired (see note above).
+    client.delete(redis_keys.runner_token_hash(old.runner_id))
+
     monkeypatch.setattr(runner, '_now', lambda: t0 + 8 * 86400)
     listed = runner.list_runners()
 
     assert listed == []
-    client = runner._redis()
     assert client.zscore(redis_keys.RUNNERS_REGISTERED, old.runner_id) is None
     assert client.exists(redis_keys.runner_meta(old.runner_id)) == 0
     assert client.exists(redis_keys.runner_token_hash(old.runner_id)) == 0
 
 
+def test_gc_spares_stale_member_whose_token_hash_survives(monkeypatch):
+    # TOCTOU regression (PR #341): a member older than 7d by score but whose
+    # token_hash still exists (clock skew, or a heartbeat that just renewed it)
+    # must NOT be swept — TTL is the sole invalidator.
+    t0 = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: t0)
+    old = runner.register('old', '1.1.1.1')
+
+    # advance past the cutoff WITHOUT deleting token_hash
+    monkeypatch.setattr(runner, '_now', lambda: t0 + 8 * 86400)
+    runner.register('fresh', '2.2.2.2')  # GC via register
+    runner.list_runners()  # GC via list_runners
+
+    client = runner._redis()
+    assert client.zscore(redis_keys.RUNNERS_REGISTERED,
+                         old.runner_id) is not None
+    assert client.exists(redis_keys.runner_meta(old.runner_id)) == 1
+    assert runner.verify_token(old.runner_id, old.token) is True
+
+
 def test_fresh_identity_survives_gc(monkeypatch):
     t0 = 1_000_000.0
     monkeypatch.setattr(runner, '_now', lambda: t0)

From e5dc757630125dfa3908bc2efa464ca539bad24c Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Wed, 15 Jul 2026 00:44:19 +0800
Subject: [PATCH 4/9] refactor(dispatch): rename GC exists-flags to avoid clash
 with runner_alive

---
 dispatch/runner.py | 7 ++++---
 1 file changed, 4 insertions(+), 3 deletions(-)

diff --git a/dispatch/runner.py b/dispatch/runner.py
index 68c451e..b6614dd 100644
--- a/dispatch/runner.py
+++ b/dispatch/runner.py
@@ -198,11 +198,12 @@ def _gc(now: Optional[float] = None) -> None:
     exists_pipe = client.pipeline()
     for runner_id in runner_ids:
         exists_pipe.exists(redis_keys.runner_token_hash(runner_id))
-    still_alive = exists_pipe.execute()
+    token_hash_exists = exists_pipe.execute()
 
     sweep = [
-        runner_id for runner_id, alive in zip(runner_ids, still_alive)
-        if not alive
+        runner_id
+        for runner_id, has_token in zip(runner_ids, token_hash_exists)
+        if not has_token
     ]
     if not sweep:
         return

From 7bff8d4e911613f517cd404bfbd223098e2bee58 Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Thu, 16 Jul 2026 15:03:53 +0800
Subject: [PATCH 5/9] docs(dispatch): honest listing semantics + precise
 compare_digest notes
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Address Copilot review on #341:
- list_runners docstring no longer claims a liveness view; revoked/dead
  identities stay listed until GC as a deliberate observability window
  (revocation only guarantees immediate auth failure, ADR-0004) — pinned
  by test_revoked_runner_stays_listed_until_gc
- compare_digest comments now cite the hmac docs' 'str (ASCII only)'
  contract (same-type non-ASCII str raises TypeError; empirically checked)
---
 dispatch/runner.py            |  8 ++++++--
 tests/test_dispatch_runner.py | 24 ++++++++++++++++++++++--
 2 files changed, 28 insertions(+), 4 deletions(-)

diff --git a/dispatch/runner.py b/dispatch/runner.py
index b6614dd..33eba42 100644
--- a/dispatch/runner.py
+++ b/dispatch/runner.py
@@ -69,7 +69,8 @@ def verify_registration_token(candidate: Optional[str]) -> bool:
     # and .encode() below would otherwise raise instead of returning False.
     if not expected or not isinstance(candidate, str) or not candidate:
         return False
-    # Compare UTF-8 bytes: compare_digest raises TypeError on non-ASCII str, and
+    # Compare UTF-8 bytes: compare_digest accepts str only when both sides are
+    # ASCII ("str (ASCII only)", hmac docs) and raises TypeError otherwise;
     # `candidate` is attacker-controlled, so str comparison could crash (500)
     # instead of failing closed (401).
     return hmac.compare_digest(expected.encode(), candidate.encode())
@@ -129,8 +130,11 @@ def verify_token(runner_id: Optional[str], token: Optional[str]) -> bool:
 
 
 def list_runners() -> List[Dict]:
-    """Return identity-layer facts for all live runners (spec §7.6 subset).
+    """Return identity-layer facts for all registered identities (spec §7.6 subset).
 
+    Not a liveness view: a revoked or dead runner stays listed (frozen
+    last_seen) until identity GC sweeps it — a deliberate observability
+    window; revocation only guarantees immediate auth failure (ADR-0004).
     Sweeps expired identities first. Fields: runner_id, name, last_seen,
     registered_at. Liveness/held-jobs are added by the admin-API slice.
     """
diff --git a/tests/test_dispatch_runner.py b/tests/test_dispatch_runner.py
index e372770..4440103 100644
--- a/tests/test_dispatch_runner.py
+++ b/tests/test_dispatch_runner.py
@@ -66,8 +66,10 @@ def test_verify_registration_token_empty_env_always_rejected(monkeypatch):
 
 
 def test_verify_registration_token_non_ascii_candidate_rejected(monkeypatch):
-    # candidate is attacker-controlled (JSON body); non-ASCII must fail closed,
-    # not raise TypeError from hmac.compare_digest on str inputs.
+    # candidate is attacker-controlled (JSON body); non-ASCII must fail closed.
+    # compare_digest accepts str only when both sides are ASCII ("str (ASCII
+    # only)", hmac docs) and raises TypeError otherwise — hence the UTF-8
+    # bytes comparison in verify_registration_token.
     monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
     assert runner.verify_registration_token('sécret-ü') is False
 
@@ -156,6 +158,24 @@ def test_revocation_deletes_token_hash_then_401():
     assert runner.verify_token(reg.runner_id, reg.token) is False
 
 
+def test_revoked_runner_stays_listed_until_gc(monkeypatch):
+    # Revocation guarantees immediate auth failure only; the identity stays
+    # visible in list_runners (deliberate observability window) until GC
+    # sweeps it past the 7d last-seen cutoff.
+    t0 = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: t0)
+    reg = runner.register('r', '1.1.1.1')
+    runner._redis().delete(redis_keys.runner_token_hash(reg.runner_id))
+
+    assert runner.verify_token(reg.runner_id, reg.token) is False
+    listed = {entry['runner_id'] for entry in runner.list_runners()}
+    assert reg.runner_id in listed
+
+    # past the GC cutoff the corpse is swept
+    monkeypatch.setattr(runner, '_now', lambda: t0 + 8 * 86400)
+    assert runner.list_runners() == []
+
+
 def test_verify_token_unknown_runner_fails():
     assert runner.verify_token('rn_does_not_exist', 'rk_whatever') is False
 

From 887c99f94709899f1582ff141abac0de50dd42f1 Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Mon, 20 Jul 2026 22:00:58 +0800
Subject: [PATCH 6/9] refactor(dispatch): registration token becomes a
 startup-snapshot setting (ADR-0005)
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

RUNNER_REGISTRATION_TOKEN moves from a live env accessor in dispatch/config.py
to a field on the centralized Settings (top-level config.py, #345), loaded and
validated once at startup. Unset/empty still means registration disabled (fail
closed); rotating the shared secret now takes a restart, per-runner revocation
stays immediate.

dispatch/config.py is renamed to dispatch/params.py and now holds only the
spec §13 protocol parameters — the dispatch module no longer reads settings
of its own.
---
 config.py                     |  5 ++++
 dispatch/__init__.py          |  2 +-
 dispatch/config.py            | 30 ----------------------
 dispatch/params.py            | 14 +++++++++++
 dispatch/runner.py            | 18 +++++++++-----
 tests/test_config.py          |  1 +
 tests/test_dispatch_runner.py | 47 ++++++++++++++++-------------------
 7 files changed, 54 insertions(+), 63 deletions(-)
 delete mode 100644 dispatch/config.py
 create mode 100644 dispatch/params.py

diff --git a/config.py b/config.py
index 0223078..9aafc2d 100644
--- a/config.py
+++ b/config.py
@@ -38,6 +38,11 @@ class Settings(BaseSettings):
     SMTP_NOREPLY: Optional[str] = None
     SMTP_NOREPLY_PASSWORD: Optional[str] = None
 
+    # Shared secret runners present when registering (spec §7.1). Unset/empty
+    # ⇒ registration is disabled, fail closed. Startup snapshot: rotating it
+    # takes a restart; per-runner revocation stays immediate (ADR-0005).
+    RUNNER_REGISTRATION_TOKEN: Optional[str] = None
+
     SUBMISSION_TMP_DIR: str = Field(
         default_factory=lambda: tempfile.mkdtemp(suffix='noj-submissions'))
 
diff --git a/dispatch/__init__.py b/dispatch/__init__.py
index 845d45a..8c98654 100644
--- a/dispatch/__init__.py
+++ b/dispatch/__init__.py
@@ -1,6 +1,6 @@
 """Redis-based pull dispatch module (spec §11).
 
-This slice ships only the runner-identity foundation (redis_keys, config,
+This slice ships only the runner-identity foundation (redis_keys, params,
 runner registration / token verification / GC). It has no callers yet —
 the HTTP layer and job lifecycle land in later slices.
 """
diff --git a/dispatch/config.py b/dispatch/config.py
deleted file mode 100644
index 53434de..0000000
--- a/dispatch/config.py
+++ /dev/null
@@ -1,30 +0,0 @@
-"""Pull-dispatch parameters (spec §13) and the registration-token accessor."""
-
-import os
-from typing import Optional
-
-# --- §13 parameters -----------------------------------------------------
-HEARTBEAT_INTERVAL_SEC = 15
-LEASE_TTL_SEC = 30
-POLL_INTERVAL_SEC = 3
-ORPHAN_SCAN_INTERVAL_SEC = 15
-MAX_ATTEMPTS = 3
-IDENTITY_TTL_SEC = 7 * 24 * 60 * 60  # 7 days
-PRESIGNED_URL_TTL_SEC = 60 * 60  # 1 hour
-MAX_CONCURRENT_JOBS = 8  # advertised to runners in the register response (§7.1)
-
-_REGISTRATION_TOKEN_ENV = 'RUNNER_REGISTRATION_TOKEN'
-
-
-def registration_token() -> Optional[str]:
-    """The shared runner registration secret, read live from the environment.
-
-    Read at call time (not cached at import) so the value stays consistent
-    with the deployed env and so verification fails closed the moment the
-    secret is removed. Returns ``None`` when unset or empty; callers must
-    treat that as "reject everything" rather than crashing.
-    """
-    token = os.getenv(_REGISTRATION_TOKEN_ENV)
-    if not token:
-        return None
-    return token
diff --git a/dispatch/params.py b/dispatch/params.py
new file mode 100644
index 0000000..a61f66b
--- /dev/null
+++ b/dispatch/params.py
@@ -0,0 +1,14 @@
+"""Pull-dispatch protocol parameters (spec §13).
+
+Deployment settings never live here — they live in the top-level
+``config.py`` Settings (ADR-0005).
+"""
+
+HEARTBEAT_INTERVAL_SEC = 15
+LEASE_TTL_SEC = 30
+POLL_INTERVAL_SEC = 3
+ORPHAN_SCAN_INTERVAL_SEC = 15
+MAX_ATTEMPTS = 3
+IDENTITY_TTL_SEC = 7 * 24 * 60 * 60  # 7 days
+PRESIGNED_URL_TTL_SEC = 60 * 60  # 1 hour
+MAX_CONCURRENT_JOBS = 8  # advertised to runners in the register response (§7.1)
diff --git a/dispatch/runner.py b/dispatch/runner.py
index 33eba42..8afda4c 100644
--- a/dispatch/runner.py
+++ b/dispatch/runner.py
@@ -15,7 +15,8 @@
 - A missing ``token_hash`` key means the identity was revoked (or expired) →
   verification fails. This is the single revocation mechanism (ADR-0004).
 - Registration verification fails closed: if the shared secret is unset/empty in
-  the environment, every candidate is rejected rather than accepted or crashing.
+  the deployment settings (startup snapshot, ADR-0005), every candidate is
+  rejected rather than accepted or crashing.
 """
 
 import hashlib
@@ -27,8 +28,9 @@
 
 from ulid import ULID
 
+from config import settings
 from mongo.utils import RedisCache
-from . import config
+from . import params
 from . import redis_keys
 
 RUNNER_ID_PREFIX = 'rn_'
@@ -62,8 +64,12 @@ def _token_hash(token: str) -> str:
 
 
 def verify_registration_token(candidate: Optional[str]) -> bool:
-    """Constant-time check of a register request's shared secret. Fails closed."""
-    expected = config.registration_token()
+    """Constant-time check of a register request's shared secret. Fails closed.
+
+    The secret is a startup-snapshot deployment setting (ADR-0005): rotating
+    it takes a Back-End restart; unset/empty ⇒ registration is disabled.
+    """
+    expected = settings.RUNNER_REGISTRATION_TOKEN
     # Fail closed: no configured secret ⇒ registration is disabled, not open.
     # Reject non-str candidates too — a JSON body can carry ints/lists/bytes,
     # and .encode() below would otherwise raise instead of returning False.
@@ -89,7 +95,7 @@ def register(name: str, ip: str) -> Registration:
     token = RUNNER_TOKEN_PREFIX + secrets.token_urlsafe(32)
 
     client = _redis()
-    ttl = config.IDENTITY_TTL_SEC
+    ttl = params.IDENTITY_TTL_SEC
     meta_key = redis_keys.runner_meta(runner_id)
 
     pipe = client.pipeline()
@@ -181,7 +187,7 @@ def _gc(now: Optional[float] = None) -> None:
     """
     if now is None:
         now = _now()
-    cutoff = now - config.IDENTITY_TTL_SEC
+    cutoff = now - params.IDENTITY_TTL_SEC
 
     client = _redis()
     # Strictly older than the cutoff: '(' makes the max bound exclusive.
diff --git a/tests/test_config.py b/tests/test_config.py
index 4e79d20..fabcb87 100644
--- a/tests/test_config.py
+++ b/tests/test_config.py
@@ -34,6 +34,7 @@ def test_defaults(clean_env):
     assert s.SMTP_SERVER is None
     assert s.SMTP_NOREPLY is None
     assert s.SMTP_NOREPLY_PASSWORD is None
+    assert s.RUNNER_REGISTRATION_TOKEN is None
 
 
 def test_debug_default_false(clean_env):
diff --git a/tests/test_dispatch_runner.py b/tests/test_dispatch_runner.py
index 4440103..936b242 100644
--- a/tests/test_dispatch_runner.py
+++ b/tests/test_dispatch_runner.py
@@ -1,12 +1,10 @@
 import hashlib
-import os
 
 import pytest
 
+from config import settings
 from mongo.utils import RedisCache
-from dispatch import config, redis_keys, runner
-
-REG_TOKEN_ENV = 'RUNNER_REGISTRATION_TOKEN'
+from dispatch import params, redis_keys, runner
 
 
 @pytest.fixture(autouse=True, scope='session')
@@ -16,22 +14,18 @@ def setup_minio():
     yield
 
 
-def _reset():
+@pytest.fixture(autouse=True)
+def fresh_fakeredis(monkeypatch):
     # Force each test onto a fresh FakeStrictRedis: RedisCache caches its
     # connection pool on the class and dispatch caches one RedisCache instance.
+    # REDIS_PORT must be None in settings so RedisCache falls back to fakeredis.
+    monkeypatch.setattr(settings, 'REDIS_HOST', None)
+    monkeypatch.setattr(settings, 'REDIS_PORT', None)
+    RedisCache.POOL = None
+    runner._cache = None
+    yield
     RedisCache.POOL = None
     runner._cache = None
-
-
-def setup_function(_):
-    # REDIS_PORT must be unset so RedisCache falls back to fakeredis.
-    os.environ.pop('REDIS_PORT', None)
-    _reset()
-
-
-def teardown_function(_):
-    _reset()
-    os.environ.pop(REG_TOKEN_ENV, None)
 
 
 def _sha256_hex(text: str) -> str:
@@ -42,25 +36,26 @@ def _sha256_hex(text: str) -> str:
 
 
 def test_verify_registration_token_accepts_correct(monkeypatch):
-    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', 'super-secret')
     assert runner.verify_registration_token('super-secret') is True
 
 
 def test_verify_registration_token_rejects_wrong(monkeypatch):
-    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', 'super-secret')
     assert runner.verify_registration_token('nope') is False
     assert runner.verify_registration_token('') is False
     assert runner.verify_registration_token(None) is False
 
 
-def test_verify_registration_token_unset_env_always_rejected(monkeypatch):
-    monkeypatch.delenv(REG_TOKEN_ENV, raising=False)
+def test_verify_registration_token_unset_always_rejected(monkeypatch):
+    # Unset setting ⇒ registration disabled (ADR-0005), never open.
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', None)
     assert runner.verify_registration_token('anything') is False
     assert runner.verify_registration_token('') is False
 
 
-def test_verify_registration_token_empty_env_always_rejected(monkeypatch):
-    monkeypatch.setenv(REG_TOKEN_ENV, '')
+def test_verify_registration_token_empty_always_rejected(monkeypatch):
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', '')
     assert runner.verify_registration_token('') is False
     assert runner.verify_registration_token('anything') is False
 
@@ -70,12 +65,12 @@ def test_verify_registration_token_non_ascii_candidate_rejected(monkeypatch):
     # compare_digest accepts str only when both sides are ASCII ("str (ASCII
     # only)", hmac docs) and raises TypeError otherwise — hence the UTF-8
     # bytes comparison in verify_registration_token.
-    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', 'super-secret')
     assert runner.verify_registration_token('sécret-ü') is False
 
 
 def test_verify_registration_token_non_ascii_secret(monkeypatch):
-    monkeypatch.setenv(REG_TOKEN_ENV, 'sécret-ü')
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', 'sécret-ü')
     assert runner.verify_registration_token('sécret-ü') is True
     assert runner.verify_registration_token('super-secret') is False
 
@@ -84,7 +79,7 @@ def test_verify_registration_token_non_ascii_secret(monkeypatch):
 def test_verify_registration_token_non_str_candidate_rejected(
         monkeypatch, candidate):
     # JSON bodies can legally carry non-str values; must fail closed, not raise.
-    monkeypatch.setenv(REG_TOKEN_ENV, 'super-secret')
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', 'super-secret')
     assert runner.verify_registration_token(candidate) is False
 
 
@@ -117,7 +112,7 @@ def test_register_creates_zset_meta_and_token_hash(monkeypatch):
     assert reg.token not in stored.decode()
 
     # 7d TTLs on meta and token_hash
-    ttl = config.IDENTITY_TTL_SEC
+    ttl = params.IDENTITY_TTL_SEC
     assert ttl - 5 <= client.ttl(redis_keys.runner_meta(reg.runner_id)) <= ttl
     assert ttl - 5 <= client.ttl(redis_keys.runner_token_hash(
         reg.runner_id)) <= ttl

From f4317708c320c525eff46d1dce21eaaa78e410c2 Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Tue, 21 Jul 2026 00:28:16 +0800
Subject: [PATCH 7/9] fix(dispatch): fail closed on lone-surrogate auth inputs
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

json.loads accepts JSON strings holding lone UTF-16 surrogates, and .encode()
on such a str raises UnicodeEncodeError — a hostile register/auth body could
turn into a 500 instead of a 401. Encode both comparison sides through a
fail-closed _utf8 helper, reject unencodable runner_id/token before touching
Redis (redis-py UTF-8 encodes keys), and compare the stored token hash as raw
bytes instead of assuming it decodes as UTF-8.
---
 dispatch/runner.py            | 40 +++++++++++++++++++++++++++++++----
 tests/test_dispatch_runner.py | 19 +++++++++++++++++
 2 files changed, 55 insertions(+), 4 deletions(-)

diff --git a/dispatch/runner.py b/dispatch/runner.py
index 8afda4c..1ba4e0f 100644
--- a/dispatch/runner.py
+++ b/dispatch/runner.py
@@ -17,6 +17,9 @@
 - Registration verification fails closed: if the shared secret is unset/empty in
   the deployment settings (startup snapshot, ADR-0005), every candidate is
   rejected rather than accepted or crashing.
+- Verification never raises on hostile input: non-str values and lone UTF-16
+  surrogate strings (which json.loads accepts but .encode() rejects) fail closed
+  to False → 401, never a 500.
 """
 
 import hashlib
@@ -63,6 +66,20 @@ def _token_hash(token: str) -> str:
     return hashlib.sha256(token.encode()).hexdigest()
 
 
+def _utf8(s: str) -> Optional[bytes]:
+    """UTF-8 encode ``s``, or None if it cannot be encoded (fail closed).
+
+    json.loads happily yields str values holding a lone UTF-16 surrogate (the
+    JSON literal "\\ud800" decodes to such a str), and .encode() on those raises
+    UnicodeEncodeError. Callers at the trust boundary use None to fail closed
+    instead of letting a hostile body turn into a 500.
+    """
+    try:
+        return s.encode()
+    except UnicodeEncodeError:
+        return None
+
+
 def verify_registration_token(candidate: Optional[str]) -> bool:
     """Constant-time check of a register request's shared secret. Fails closed.
 
@@ -78,8 +95,13 @@ def verify_registration_token(candidate: Optional[str]) -> bool:
     # Compare UTF-8 bytes: compare_digest accepts str only when both sides are
     # ASCII ("str (ASCII only)", hmac docs) and raises TypeError otherwise;
     # `candidate` is attacker-controlled, so str comparison could crash (500)
-    # instead of failing closed (401).
-    return hmac.compare_digest(expected.encode(), candidate.encode())
+    # instead of failing closed (401). _utf8 also fails closed on a lone-
+    # surrogate candidate whose .encode() would raise UnicodeEncodeError.
+    expected_bytes = _utf8(expected)
+    candidate_bytes = _utf8(candidate)
+    if expected_bytes is None or candidate_bytes is None:
+        return False
+    return hmac.compare_digest(expected_bytes, candidate_bytes)
 
 
 def register(name: str, ip: str) -> Registration:
@@ -128,11 +150,21 @@ def verify_token(runner_id: Optional[str], token: Optional[str]) -> bool:
         return False
     if not runner_id or not token:
         return False
+    # Fail closed before touching Redis on a lone-surrogate runner_id/token whose
+    # UTF-8 encode raises (redis-py encodes keys to UTF-8; _token_hash encodes
+    # the token) — a hostile body must 401, not 500.
+    if _utf8(runner_id) is None:
+        return False
+    token_bytes = _utf8(token)
+    if token_bytes is None:
+        return False
     stored = _redis().get(redis_keys.runner_token_hash(runner_id))
     if stored is None:
         return False
-    stored_hex = stored.decode()
-    return hmac.compare_digest(stored_hex, _token_hash(token))
+    # Compare raw bytes: no assumption the stored value decodes as UTF-8.
+    return hmac.compare_digest(
+        stored,
+        hashlib.sha256(token_bytes).hexdigest().encode())
 
 
 def list_runners() -> List[Dict]:
diff --git a/tests/test_dispatch_runner.py b/tests/test_dispatch_runner.py
index 936b242..8e64480 100644
--- a/tests/test_dispatch_runner.py
+++ b/tests/test_dispatch_runner.py
@@ -83,6 +83,15 @@ def test_verify_registration_token_non_str_candidate_rejected(
     assert runner.verify_registration_token(candidate) is False
 
 
+def test_verify_registration_token_lone_surrogate_candidate_rejected(
+        monkeypatch):
+    # json.loads happily yields str values holding a lone UTF-16 surrogate
+    # (e.g. json.loads('"\\ud800"')), whose .encode() raises UnicodeEncodeError.
+    # An attacker-controlled candidate must fail closed (False), never a 500.
+    monkeypatch.setattr(settings, 'RUNNER_REGISTRATION_TOKEN', 'super-secret')
+    assert runner.verify_registration_token('\ud800') is False
+
+
 # --- register -----------------------------------------------------------
 
 
@@ -190,6 +199,16 @@ def test_verify_token_non_str_inputs_rejected(bad):
     assert runner.verify_token(bad, reg.token) is False
 
 
+def test_verify_token_lone_surrogate_inputs_rejected():
+    # json.loads can yield str holding a lone UTF-16 surrogate (e.g.
+    # json.loads('"\\ud800"')) whose .encode() raises UnicodeEncodeError;
+    # runner_id also feeds a Redis key (redis-py UTF-8 encodes it, raising too).
+    # Both token and runner_id must fail closed (False), never a 500.
+    reg = runner.register('r', '1.1.1.1')
+    assert runner.verify_token(reg.runner_id, '\ud800') is False
+    assert runner.verify_token('\ud800', reg.token) is False
+
+
 # --- lazy GC ------------------------------------------------------------
 
 

From 05d1dadb733624eee589bceef4bb9addaddc3edf Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Tue, 21 Jul 2026 00:47:33 +0800
Subject: [PATCH 8/9] perf(dispatch): batch list_runners meta fetches, drop
 MULTI/EXEC from GC pipelines

list_runners fetched each runner's meta hash in its own round trip (N+1);
batch them in one non-transactional pipeline. GC's exists probe and sweep
need no atomicity (TTL monotonicity, see module docstring), so skip
MULTI/EXEC there too. register keeps its transaction: identity creation is
all-or-nothing on purpose.
---
 dispatch/runner.py | 20 +++++++++++++++-----
 1 file changed, 15 insertions(+), 5 deletions(-)

diff --git a/dispatch/runner.py b/dispatch/runner.py
index 1ba4e0f..7e482ef 100644
--- a/dispatch/runner.py
+++ b/dispatch/runner.py
@@ -120,6 +120,8 @@ def register(name: str, ip: str) -> Registration:
     ttl = params.IDENTITY_TTL_SEC
     meta_key = redis_keys.runner_meta(runner_id)
 
+    # Identity creation is deliberately all-or-nothing (MULTI/EXEC), unlike the
+    # non-transactional pipelines in _gc/list_runners.
     pipe = client.pipeline()
     pipe.zadd(redis_keys.RUNNERS_REGISTERED, {runner_id: now})
     pipe.hset(
@@ -185,10 +187,18 @@ def list_runners() -> List[Dict]:
                             -1,
                             withscores=True)
 
+    runner_ids = [
+        member.decode() if isinstance(member, bytes) else member
+        for member, _ in members
+    ]
+
+    meta_pipe = client.pipeline(transaction=False)
+    for runner_id in runner_ids:
+        meta_pipe.hgetall(redis_keys.runner_meta(runner_id))
+    raw_metas = meta_pipe.execute()
+
     runners: List[Dict] = []
-    for member, score in members:
-        runner_id = member.decode() if isinstance(member, bytes) else member
-        raw_meta = client.hgetall(redis_keys.runner_meta(runner_id))
+    for (_, score), runner_id, raw_meta in zip(members, runner_ids, raw_metas):
         meta = {
             (k.decode() if isinstance(k, bytes) else k):
             (v.decode() if isinstance(v, bytes) else v)
@@ -237,7 +247,7 @@ def _gc(now: Optional[float] = None) -> None:
     ]
 
     # Only sweep members whose token_hash has already expired (TTL fired).
-    exists_pipe = client.pipeline()
+    exists_pipe = client.pipeline(transaction=False)
     for runner_id in runner_ids:
         exists_pipe.exists(redis_keys.runner_token_hash(runner_id))
     token_hash_exists = exists_pipe.execute()
@@ -250,7 +260,7 @@ def _gc(now: Optional[float] = None) -> None:
     if not sweep:
         return
 
-    pipe = client.pipeline()
+    pipe = client.pipeline(transaction=False)
     for runner_id in sweep:
         pipe.zrem(redis_keys.RUNNERS_REGISTERED, runner_id)
         pipe.delete(redis_keys.runner_meta(runner_id))

From cab1efea13b2524e551007a538e7a26005bc8ede Mon Sep 17 00:00:00 2001
From: as535364 <me@ujoj.cc>
Date: Tue, 21 Jul 2026 00:50:43 +0800
Subject: [PATCH 9/9] test(dispatch): pin meta pairing in batched list_runners

---
 tests/test_dispatch_runner.py | 16 ++++++++++++++++
 1 file changed, 16 insertions(+)

diff --git a/tests/test_dispatch_runner.py b/tests/test_dispatch_runner.py
index 8e64480..408d220 100644
--- a/tests/test_dispatch_runner.py
+++ b/tests/test_dispatch_runner.py
@@ -306,5 +306,21 @@ def test_list_runners_returns_identity_fields(monkeypatch):
     assert 'token_hash' not in entry
 
 
+def test_list_runners_pairs_meta_with_correct_runner(monkeypatch):
+    # The batched meta fetch pairs ZSET members with pipeline results by
+    # position; each entry must keep its own name/last_seen.
+    t0 = 1_000_000.0
+    monkeypatch.setattr(runner, '_now', lambda: t0)
+    a = runner.register('runner-a', '1.1.1.1')
+    monkeypatch.setattr(runner, '_now', lambda: t0 + 60)
+    b = runner.register('runner-b', '2.2.2.2')
+
+    by_id = {entry['runner_id']: entry for entry in runner.list_runners()}
+    assert by_id[a.runner_id]['name'] == 'runner-a'
+    assert by_id[a.runner_id]['last_seen'] == t0
+    assert by_id[b.runner_id]['name'] == 'runner-b'
+    assert by_id[b.runner_id]['last_seen'] == t0 + 60
+
+
 def test_list_runners_empty():
     assert runner.list_runners() == []