diff --git a/dispatch/job.py b/dispatch/job.py
new file mode 100644
index 0000000..f02312b
--- /dev/null
+++ b/dispatch/job.py
@@ -0,0 +1,182 @@
+"""Job lifecycle: enqueue / claim / renew / reclaim (spec §7.2, §7.3, §8, §9).
+
+This is the Python skin over the Lua state transitions in ``scripts.py``. Every
+transition is atomic in Redis; this layer only picks the ``now`` clock (Python
+``time.time()``, passed into every script via ARGV — never a server-side TIME),
+walks ``jobs:leased`` for the time-gated orphan scan, and shapes payloads.
+
+Slice 2 of the delivery plan (§15.2) — dark: nothing calls these yet. The
+``complete``/``abort``/JE-landing transitions are slice 3; presigned ``code_url``
+signing on the payload is slice 4. Both are called out at their seams below.
+"""
+
+import time
+from typing import Dict, Optional
+
+from ulid import ULID
+
+from mongo.utils import RedisCache
+from . import params
+from . import redis_keys
+from . import scripts
+
+JOB_ID_PREFIX = 'jb_'
+
+# One shared RedisCache instance, mirroring runner.py: under fakeredis each
+# RedisCache owns an isolated dataset, so enqueue/claim/renew must share one to
+# stay coherent; in production they all share the pooled real Redis anyway.
+_cache: Optional[RedisCache] = None
+
+
+def _redis():
+    global _cache
+    if _cache is None:
+        _cache = RedisCache()
+    return _cache.client
+
+
+def _now() -> float:
+    return time.time()
+
+
+def _decode(value) -> Optional[str]:
+    if value is None:
+        return None
+    return value.decode() if isinstance(value, bytes) else value
+
+
+def enqueue_job(
+    submission_id: str,
+    problem_id: str,
+    language: int,
+    code_minio_path: str,
+    checker: str,
+    tasks_meta_json: str,
+) -> str:
+    """Enqueue a fresh judging job for ``submission_id`` (spec §8, §9).
+
+    Writes the job hash, points ``submission:<sid>:current_job`` at it (last
+    enqueue wins — that is exactly how a rejudge supersedes an in-flight job; the
+    stale one is destroyed lazily on the dispatch side by ``claim_pending``, INV4,
+    never proactively here), then LPUSHes onto the FIFO pending queue.
+    """
+    now = _now()
+    job_id = JOB_ID_PREFIX + str(ULID())
+
+    client = _redis()
+    job_key = redis_keys.job(job_id)
+
+    pipe = client.pipeline()
+    pipe.hset(
+        job_key,
+        mapping={
+            'submission_id': submission_id,
+            'problem_id': problem_id,
+            'language': language,
+            'code_minio_path': code_minio_path,
+            'checker': checker,
+            'tasks_meta_json': tasks_meta_json,
+            'leased_by': '',
+            'lease_deadline': '0',
+            'state': 'pending',
+            'attempts': 0,
+            'created_at': repr(now),
+            'last_error': '',
+        },
+    )
+    pipe.set(redis_keys.submission_current_job(submission_id), job_id)
+    pipe.lpush(redis_keys.JOBS_PENDING, job_id)
+    pipe.execute()
+
+    return job_id
+
+
+def claim_next_job(runner_id: str) -> Optional[Dict]:
+    """Hand ``runner_id`` its next job, or None when there is nothing to do.
+
+    Order matches spec §7.3: first a time-gated orphan scan (at most one runner
+    per ``ORPHAN_SCAN_INTERVAL_SEC`` window), whose first successful reclaim goes
+    straight to this caller; otherwise fall through to the pending queue.
+    """
+    now = _now()
+    client = _redis()
+
+    reclaimed = _orphan_scan(client, runner_id, now)
+    if reclaimed is not None:
+        return _payload(client, reclaimed)
+
+    job_id = scripts.load(client).claim_pending(
+        keys=[redis_keys.JOBS_PENDING, redis_keys.JOBS_LEASED],
+        args=[now, runner_id, params.LEASE_TTL_SEC],
+    )
+    if job_id is None:
+        return None
+    return _payload(client, _decode(job_id))
+
+
+def renew_lease(runner_id: str, job_id: str) -> bool:
+    """Extend the lease on ``job_id`` iff ``runner_id`` still owns it (spec §7.2).
+
+    Thin wrapper over ``renew_lease`` Lua; heartbeat wiring is slice 4.
+    """
+    now = _now()
+    result = scripts.load(_redis()).renew_lease(
+        keys=[redis_keys.JOBS_LEASED,
+              redis_keys.job(job_id)],
+        args=[job_id, runner_id, now, params.LEASE_TTL_SEC],
+    )
+    return result == 1
+
+
+def _orphan_scan(client, runner_id: str, now: float) -> Optional[str]:
+    """Time-gated sweep of ``jobs:leased``; return an id reclaimed for the caller.
+
+    The ``SET NX EX`` gate on ``dispatch:last_recovery`` lets at most one scan run
+    per window (spec §7.3 step 1). Within a winning scan, each expired candidate
+    is offered to ``reclaim_expired``: the first ``1`` (reclaimed to this caller)
+    is returned immediately; ``-1`` (attempts exhausted) candidates are skipped
+    and left in place — the JE landing sweep that removes them is slice 3.
+    """
+    acquired = client.set(
+        redis_keys.DISPATCH_LAST_RECOVERY,
+        '1',
+        nx=True,
+        ex=params.ORPHAN_SCAN_INTERVAL_SEC,
+    )
+    if not acquired:
+        return None
+
+    reclaim = scripts.load(client).reclaim_expired
+    for member in client.smembers(redis_keys.JOBS_LEASED):
+        job_id = _decode(member)
+        result = reclaim(
+            keys=[redis_keys.JOBS_LEASED,
+                  redis_keys.job(job_id)],
+            args=[
+                job_id,
+                runner_id,
+                now,
+                params.LEASE_TTL_SEC,
+                params.MAX_ATTEMPTS,
+            ],
+        )
+        if result == 1:
+            return job_id
+    return None
+
+
+def _payload(client, job_id: str) -> Optional[Dict]:
+    """Shape a job hash into a claim payload (spec §7.3), or None if it vanished.
+
+    Returns ``job_id`` plus the raw hash fields the HTTP layer will need. Typed
+    coercion and the presigned ``code_url`` are slice 4's concern, not this one's.
+    """
+    raw = client.hgetall(redis_keys.job(job_id))
+    if not raw:
+        return None
+    fields: Dict[str, Optional[str]] = {
+        _decode(k): _decode(v)
+        for k, v in raw.items()
+    }
+    fields['job_id'] = job_id
+    return fields
diff --git a/dispatch/scripts.py b/dispatch/scripts.py
new file mode 100644
index 0000000..a6a5c04
--- /dev/null
+++ b/dispatch/scripts.py
@@ -0,0 +1,170 @@
+"""Lua scripts — one per job state transition (spec §8, §9).
+
+Each script is atomic on the Redis side, so a state transition can never be
+observed half-applied even under concurrent runners (INV2/INV3/INV4/INV5).
+Every script takes ``now`` via ARGV — never ``redis.call('TIME')`` — so tests
+are deterministic and fakeredis (which has no server clock for scripts) works.
+
+The scripts here cover slice 2 of the delivery plan (§15.2): ``claim_pending``,
+``renew_lease``, ``reclaim_expired``. ``abort_requeue`` belongs to slice 3 and
+is deliberately absent.
+
+Key handling: static keys (``jobs:pending``, ``jobs:leased``, a specific
+``job:<jb>`` hash) are passed in as KEYS by the Python layer via redis_keys.
+``claim_pending`` is the sole exception — it discovers a job id by RPOP and so
+must build that job's ``job:<jb>`` hash key and its
+``submission:<sid>:current_job`` pointer key from inside Lua; those two literal
+templates mirror ``redis_keys.job`` / ``redis_keys.submission_current_job`` and
+are the only place key names are spelled outside redis_keys.
+
+Corrupted state (partial Redis data loss — never produced by normal operation)
+is reaped lazily, in passing: the job hash is the single source of truth and
+the pending list / leased set are only indexes, so a script that stumbles on a
+dangling entry destroys or unindexes it and moves on instead of erroring. The
+affected submission stays Pending and is rescued by rejudge (spec §12).
+"""
+
+from dataclasses import dataclass
+from typing import Any
+
+# claim_pending — pending → leased, with dispatch-side currency destroy (INV4).
+#
+# KEYS[1] = jobs:pending (LIST)   KEYS[2] = jobs:leased (SET)
+# ARGV[1] = now   ARGV[2] = runner_id   ARGV[3] = lease_ttl_sec
+#
+# Loop popping the FIFO tail: a job whose hash has evaporated or lost its
+# submission_id (currency would be unverifiable — corrupted beyond judging) is
+# destroyed and skipped; a job that is no longer its submission's current_job is
+# destroyed on the spot (the dispatch-side half of INV4 — a superseded rejudge
+# job never runs); the first live current job is leased (attempts+1,
+# state=leased) and its id returned. A missing attempts field falls back to 0
+# instead of erroring — the id is already popped at that point, so a script
+# error would silently lose the job. Empty queue → nil.
+CLAIM_PENDING = '''
+local now = tonumber(ARGV[1])
+local runner = ARGV[2]
+local ttl = tonumber(ARGV[3])
+while true do
+  local jb = redis.call('RPOP', KEYS[1])
+  if not jb then
+    return nil
+  end
+  local job_key = 'job:' .. jb
+  local sid = redis.call('HGET', job_key, 'submission_id')
+  if not sid then
+    -- Hash evaporated (DEL is a no-op then) or corrupted beyond judging (no
+    -- submission_id means currency cannot be verified): destroy, move on.
+    redis.call('DEL', job_key)
+  else
+    local current = redis.call('GET', 'submission:' .. sid .. ':current_job')
+    if current == jb then
+      local attempts = (tonumber(redis.call('HGET', job_key, 'attempts')) or 0) + 1
+      redis.call('HSET', job_key,
+        'attempts', attempts,
+        'leased_by', runner,
+        'lease_deadline', now + ttl,
+        'state', 'leased')
+      redis.call('SADD', KEYS[2], jb)
+      return jb
+    else
+      redis.call('DEL', job_key)
+    end
+  end
+end
+'''
+
+# renew_lease — heartbeat extends an owned lease (spec §7.2).
+#
+# KEYS[1] = jobs:leased (SET)   KEYS[2] = job:<jb> (HASH)
+# ARGV[1] = job_id   ARGV[2] = runner_id   ARGV[3] = now   ARGV[4] = lease_ttl_sec
+#
+# Only when the job is still leased AND owned by the caller: push the deadline to
+# now+ttl, return 1. Anything else (not leased, wrong owner, evaporated hash) →
+# 0. An evaporated hash additionally reaps the ghost member from jobs:leased —
+# the hash is the source of truth, the set only an index. Never creates or
+# resurrects a key (SREM only removes) — HSET runs only on the confirmed-owner
+# path, where the hash provably exists.
+RENEW_LEASE = '''
+if redis.call('SISMEMBER', KEYS[1], ARGV[1]) == 0 then
+  return 0
+end
+if redis.call('EXISTS', KEYS[2]) == 0 then
+  -- Hash is the source of truth; a member whose hash evaporated is a ghost --
+  -- reap it so the orphan scan stops rescanning it forever.
+  redis.call('SREM', KEYS[1], ARGV[1])
+  return 0
+end
+local leased_by = redis.call('HGET', KEYS[2], 'leased_by')
+if leased_by == ARGV[2] then
+  redis.call('HSET', KEYS[2], 'lease_deadline', tonumber(ARGV[3]) + tonumber(ARGV[4]))
+  return 1
+end
+return 0
+'''
+
+# reclaim_expired — expired lease → new leaseholder, or converge (spec §7.3, §9).
+#
+# KEYS[1] = jobs:leased (SET)   KEYS[2] = job:<jb> (HASH)
+# ARGV[1] = job_id   ARGV[2] = new_runner   ARGV[3] = now
+# ARGV[4] = lease_ttl_sec   ARGV[5] = max_attempts
+#
+# Eligibility is decided ONLY by lease_deadline vs now (INV2) — runner liveness
+# never enters. If expired and attempts < max: attempts+1, hand to new_runner,
+# fresh deadline, return 1. If expired and attempts already >= max: return -1 and
+# leave the job UNTOUCHED (INV5 boundary; the JE landing sweep is slice 3, not
+# this script's job). Not leased / not expired → 0. An evaporated hash → 0 and
+# the ghost member is reaped from jobs:leased (hash is the source of truth); a
+# hash that still exists but lost its lease_deadline field stays tracked — a
+# runner may well be judging it, and tearing down live state risks losing the
+# job entirely.
+RECLAIM_EXPIRED = '''
+if redis.call('SISMEMBER', KEYS[1], ARGV[1]) == 0 then
+  return 0
+end
+if redis.call('EXISTS', KEYS[2]) == 0 then
+  -- Hash is the source of truth; a member whose hash evaporated is a ghost --
+  -- reap it so the orphan scan stops rescanning it forever.
+  redis.call('SREM', KEYS[1], ARGV[1])
+  return 0
+end
+local deadline = redis.call('HGET', KEYS[2], 'lease_deadline')
+if not deadline then
+  -- Hash alive but the field is gone: leave it tracked (see header note).
+  return 0
+end
+local now = tonumber(ARGV[3])
+if tonumber(deadline) >= now then
+  return 0
+end
+local attempts = tonumber(redis.call('HGET', KEYS[2], 'attempts'))
+if attempts >= tonumber(ARGV[5]) then
+  return -1
+end
+redis.call('HSET', KEYS[2],
+  'attempts', attempts + 1,
+  'leased_by', ARGV[2],
+  'lease_deadline', now + tonumber(ARGV[4]),
+  'state', 'leased')
+return 1
+'''
+
+
+@dataclass(frozen=True)
+class Scripts:
+    claim_pending: Any
+    renew_lease: Any
+    reclaim_expired: Any
+
+
+def load(client) -> Scripts:
+    """Register the per-transition scripts on ``client`` and return them.
+
+    ``register_script`` is cheap (it only stores the body and computes the SHA
+    lazily on first EVALSHA), so binding to the currently active shared client on
+    each use keeps scripts coherent with the ``_cache`` reset done per test.
+    """
+    return Scripts(
+        claim_pending=client.register_script(CLAIM_PENDING),
+        renew_lease=client.register_script(RENEW_LEASE),
+        reclaim_expired=client.register_script(RECLAIM_EXPIRED),
+    )
diff --git a/poetry.lock b/poetry.lock
index 4b72b43..a39efae 100644
--- a/poetry.lock
+++ b/poetry.lock
@@ -725,6 +725,83 @@ files = [
     {file = "iniconfig-2.1.0.tar.gz", hash = "sha256:3abbd2e30b36733fee78f9c7f7308f2d0050e88f0087fd25c2645f63c773e1c7"},
 ]
 
+[[package]]
+name = "lupa"
+version = "2.8"
+description = "Python wrapper around Lua and LuaJIT"
+optional = false
+python-versions = ">=3.8"
+groups = ["dev"]
+files = [
+    {file = "lupa-2.8-cp310-abi3-win32.whl", hash = "sha256:c2a5fd15dc62374e1661a55f01744c9ec1c56f291ba4a0749d3af2174556e78f"},
+    {file = "lupa-2.8-cp310-abi3-win_arm64.whl", hash = "sha256:9e304fb1c50cf23fd8882afbe1aa87525ef8a72667bcab3b37b2bbb2bc542269"},
+    {file = "lupa-2.8-cp310-cp310-macosx_11_0_arm64.whl", hash = "sha256:97bd01e90b8031e56a5fd5bb70605aea09f1dba675c1140308a52780f93d06f1"},
+    {file = "lupa-2.8-cp310-cp310-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:0b5ebe1a13c45767919c86750b84fe2da9f6288b6f3cea4ce7660bb2abc9d921"},
+    {file = "lupa-2.8-cp310-cp310-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:097e7d0f1719a88020b67c82e05d53d7973c166952393afcecfd8434c7e19a15"},
+    {file = "lupa-2.8-cp310-cp310-win_amd64.whl", hash = "sha256:7bb223ee8f72d0dc076b0d65296ee72f1c69450f9d2fed5315f7707d98c4a03d"},
+    {file = "lupa-2.8-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:b12e43c1fb787189dfc28cd604aef0baa2cb95e27da19498d520361d0ace070a"},
+    {file = "lupa-2.8-cp311-cp311-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:f6f603391dffb256e36a79fd2044084d5f4b8a0a4c0e5ad291cd3ab3aaf1fd0a"},
+    {file = "lupa-2.8-cp311-cp311-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9f6f41c91366e7d0d474f87d81c1274af861f40812bf729c9f97ab4c8f3c7ac8"},
+    {file = "lupa-2.8-cp311-cp311-win_amd64.whl", hash = "sha256:f5a6af145b0ea818f01d27bfe2583a4b538570bef61d22c8773e0eccf011234c"},
+    {file = "lupa-2.8-cp312-abi3-macosx_10_13_x86_64.whl", hash = "sha256:f4342f4de76ae7ce2ab0672d36003bdb7e1a33252f293b569298ddd792e70e33"},
+    {file = "lupa-2.8-cp312-abi3-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:4203fa1659315e939a5304e75001b8cc14234fb3cbb3ed86c049b0cc5d90fcee"},
+    {file = "lupa-2.8-cp312-abi3-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:81f2d843ce668b653146c007467570210ae44be51dac6926666c51d49536f307"},
+    {file = "lupa-2.8-cp312-abi3-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:d3d0cde2c77588d1c60875a4f34f059513476c6e1775351897195b51e0f3df08"},
+    {file = "lupa-2.8-cp312-abi3-manylinux_2_34_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:9e0d11b8f3a8dac6413f704fef7161d048bb10c58bdac6cbffa5e60efa56e9a3"},
+    {file = "lupa-2.8-cp312-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:54cff414f21f8cd8c6be4aae52541f3b9cd39602b59e3a3db9b5c9f9f674ff18"},
+    {file = "lupa-2.8-cp312-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:24b4d8af5558e549b70daf1547f5c1c1d664ecea9fc790f83efe5d75e9a93797"},
+    {file = "lupa-2.8-cp312-abi3-musllinux_1_2_i686.whl", hash = "sha256:ce86dff1ee7f7cf45f5622065ae991949dd7bb1703581cbc58a630137bb7ccf9"},
+    {file = "lupa-2.8-cp312-abi3-musllinux_1_2_ppc64le.whl", hash = "sha256:f4d01b2a08c70bbb883a9e082b6b36b89121ed5910b710f1ba11c73295ff4fba"},
+    {file = "lupa-2.8-cp312-abi3-musllinux_1_2_riscv64.whl", hash = "sha256:7f210d5a8353e510ea1199c42cf3cbdd630553bf2bc8fb4c00fea06fdec7c798"},
+    {file = "lupa-2.8-cp312-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:4f81a02806e7c7ad26d8c6fa222c8bef1b0c1b124347c879be880b41339d41e4"},
+    {file = "lupa-2.8-cp312-abi3-win32.whl", hash = "sha256:360056453a7a4eaa4ac5a204c31a5a014b1eb2ee5490603234d2ba831684f1f2"},
+    {file = "lupa-2.8-cp312-abi3-win_arm64.whl", hash = "sha256:1628371c6592a6d5650497a9e31fb2bb3a7e9883c1f301d1111265e484045af9"},
+    {file = "lupa-2.8-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:450650f91c48c2415b0d59ab3abfcfda3b6efb5b858205f4d4bda8ad141fa529"},
+    {file = "lupa-2.8-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:27044f3363047f946b3d3aab9157cbd172b3538ada9ec1baef43432bf7d03a78"},
+    {file = "lupa-2.8-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:8cf4f064a0e5531afce2d7d750120c10c10f9529139af6ca6150d13151034398"},
+    {file = "lupa-2.8-cp312-cp312-win_amd64.whl", hash = "sha256:281bedc5deb92d31e649a3552edd662449365a635904fa4d5cb4509c7245e34e"},
+    {file = "lupa-2.8-cp313-cp313-macosx_11_0_arm64.whl", hash = "sha256:45fc9da0145ecb0083ef5ff9975116cc784bd0258bdc2bd131ba15483ce18398"},
+    {file = "lupa-2.8-cp313-cp313-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:58e18afed57955b41130e269c78f53d4123ab86e236b53816f4cbffa25cb5d30"},
+    {file = "lupa-2.8-cp313-cp313-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:fc47f536ac13a79cef47d29a2b205576a22841f042a2bcec1676b95806e7706a"},
+    {file = "lupa-2.8-cp313-cp313-win_amd64.whl", hash = "sha256:ce9404c661dbac65cc9bed351ad45e797af93d30d70be309a3fa8209ac86d93b"},
+    {file = "lupa-2.8-cp314-cp314-macosx_11_0_arm64.whl", hash = "sha256:348c3f8ecabb6324dcbc05c2740d762ef8fcec7b06c79e45262ab97a217684e3"},
+    {file = "lupa-2.8-cp314-cp314-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:951496471056061598a7d1729a6cdf48d662fec777a9f2d8aa5a1e62fd30e5a5"},
+    {file = "lupa-2.8-cp314-cp314-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:a591b9947ca347b41a63370e121d6e2b1458fe6dde9ae065029ec10a37f25ff4"},
+    {file = "lupa-2.8-cp314-cp314-win_amd64.whl", hash = "sha256:3903c9cf628dae2f56405503247b77a61a3a61bd2dda470e336950c74776d55d"},
+    {file = "lupa-2.8-cp314-cp314t-macosx_11_0_arm64.whl", hash = "sha256:f711a8ab0486b9ac6fdda94a22ddcfbc9f0d4a27e3a8cf1bf79c6e48b33017c1"},
+    {file = "lupa-2.8-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:dc51250e76367a3e27fcd01dc769b9bfcbbc34f48df48dde53d6af6e75b7eaa5"},
+    {file = "lupa-2.8-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:f8a22088a552828958603323f0a5c4b3e11e03b75d0bf4c965ef879de9b60a8d"},
+    {file = "lupa-2.8-cp314-cp314t-win32.whl", hash = "sha256:4f7c553c1d8cfffbe85d81daef730d12cae4b6002d457542914da0ac8a1145b3"},
+    {file = "lupa-2.8-cp314-cp314t-win_amd64.whl", hash = "sha256:d8766aff03a78c80ad2d188a8bdb216de5ec838359cd87e05bbdfa56394a6105"},
+    {file = "lupa-2.8-cp314-cp314t-win_arm64.whl", hash = "sha256:91d622777febda3ab1bed1d45295f2f32a4680c7b3d7caf8c669998ed5c44118"},
+    {file = "lupa-2.8-cp38-cp38-macosx_11_0_arm64.whl", hash = "sha256:81b283bfb13cc43fa4910fc98ec110ab861bcb39680f48b266f99d6e3be1049e"},
+    {file = "lupa-2.8-cp38-cp38-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:5caf45d15d424cee52fd67341e96e2b1dde0658ae90eb156ac56aa0d8330bc38"},
+    {file = "lupa-2.8-cp38-cp38-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:33e7e5aebca64b154b0a1679caf79e19254ff37bba51e87abab6848f97cb2de1"},
+    {file = "lupa-2.8-cp38-cp38-win32.whl", hash = "sha256:e8d4f4dd4acf4a0e42adc6b1ad220e1c86fe3028402c2f78bd0728a6d241bbe9"},
+    {file = "lupa-2.8-cp38-cp38-win_amd64.whl", hash = "sha256:1ac2b1ec7504e6148cba1bc35ac36c74d18a0ca6d367ffe7e78a3773c2694c0e"},
+    {file = "lupa-2.8-cp39-abi3-macosx_10_9_x86_64.whl", hash = "sha256:b036738282a5acd2e71fdddb317c9df8b87c1673aa57f403d05fcc2be8abc4ba"},
+    {file = "lupa-2.8-cp39-abi3-manylinux2010_i686.manylinux_2_12_i686.manylinux_2_28_i686.whl", hash = "sha256:ac6b6e8d0e617e26a98cbb44880bcd75de5d32b3ad7b3b3793583909292b47ed"},
+    {file = "lupa-2.8-cp39-abi3-manylinux2014_armv7l.manylinux_2_17_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:ba3a7dd839f90c3d2e53bebe3c192b1f3f9fd720a6781256405123211fd0dce6"},
+    {file = "lupa-2.8-cp39-abi3-manylinux2014_ppc64le.manylinux_2_17_ppc64le.manylinux_2_28_ppc64le.whl", hash = "sha256:d7edb13a7a5250b5c6c22d1495d9e842b5c9fc5081c8fe6b5efe2112fe3e41f9"},
+    {file = "lupa-2.8-cp39-abi3-manylinux_2_34_riscv64.manylinux_2_39_riscv64.whl", hash = "sha256:891f72e0bffbed1e4175f975aeb2a083956586a100066525e1be485f617f7b25"},
+    {file = "lupa-2.8-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:a295f87b5b7ebbfd5191932e8cb0e51df3c7769101ac6b6c7d7c9fb27bfd1307"},
+    {file = "lupa-2.8-cp39-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:4fe5d7a810b64ea8511eb885fc8cdde042ee5ff7b7d08ae78f32449756acb177"},
+    {file = "lupa-2.8-cp39-abi3-musllinux_1_2_i686.whl", hash = "sha256:bfc470012ef66ad064c7bd77416af03a3452ef630b04b9012595ea13f2e54518"},
+    {file = "lupa-2.8-cp39-abi3-musllinux_1_2_ppc64le.whl", hash = "sha256:250e035fdaffe8c87093e3ebc206ac29a26131b1568ea711d780c26001ce96e7"},
+    {file = "lupa-2.8-cp39-abi3-musllinux_1_2_riscv64.whl", hash = "sha256:b9bddb09acfffb4f828f790f444b11dc0cca591afea1a244d9329eea2d20c003"},
+    {file = "lupa-2.8-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:2e64acbbd47e9b82a64405a39e0d2b36a5a7dad8ab41c0f3437f572f7d282ba3"},
+    {file = "lupa-2.8-cp39-cp39-macosx_11_0_arm64.whl", hash = "sha256:f6ddca4774d5ca451768a95e378a3aa041076e29f4613b8562f8e98efb6690fd"},
+    {file = "lupa-2.8-cp39-cp39-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:3ffcfd8e19f943ad459136b3f60f085ae4948f024192a93ca4b4ac3023ec88d8"},
+    {file = "lupa-2.8-cp39-cp39-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:9f3f3955f65f9fde2dc6eda3041ccd394cf54d4bf083f0cdf6feb3d58e5f38d3"},
+    {file = "lupa-2.8-cp39-cp39-win32.whl", hash = "sha256:9e76e45057cfcaa20ee3422c2289a91f9d51783d020da3570ee226de8f6e71cd"},
+    {file = "lupa-2.8-cp39-cp39-win_amd64.whl", hash = "sha256:6fbcc9911f05c67affbd225fc024268e61e98a18ad1b1c2aed6c8796e4056554"},
+    {file = "lupa-2.8-cp39-cp39-win_arm64.whl", hash = "sha256:6c817d5421094507662e5f8feb8cd1e154c10879921c06079b6063be9d8f33c5"},
+    {file = "lupa-2.8-pp311-pypy311_pp73-manylinux2014_aarch64.manylinux_2_17_aarch64.manylinux_2_28_aarch64.whl", hash = "sha256:32e4e5103bbddcdd2458fb2ccae6c8ba11c9997c711d7e379e0d45551d109c76"},
+    {file = "lupa-2.8-pp311-pypy311_pp73-manylinux2014_x86_64.manylinux_2_17_x86_64.manylinux_2_28_x86_64.whl", hash = "sha256:7667001804657496dee9feced2daae5000b4604a3218dd8e6b7b754982ba88b8"},
+    {file = "lupa-2.8-pp311-pypy311_pp73-win_amd64.whl", hash = "sha256:86f6f668966965b15247dc32d064cfe7be67b71e584ccfacbe2f637575296878"},
+    {file = "lupa-2.8.tar.gz", hash = "sha256:d8022641b9ec8ecf2c5ecbe9f47e5a70e0b87c4b5ae921b92cb02a638e0acd08"},
+]
+
 [[package]]
 name = "lxml"
 version = "6.1.0"
@@ -2206,4 +2283,4 @@ platformdirs = ">=3.5.1"
 [metadata]
 lock-version = "2.1"
 python-versions = "^3.11"
-content-hash = "24f89be08124671130842e4f8254ce485f5942d56908d131a09ff42ffadf59c9"
+content-hash = "3fafd26539a79930a010ebb4b7c6d996231e89dfda84edb4ea673ff152d9764e"
diff --git a/pyproject.toml b/pyproject.toml
index d94a335..f979b0f 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -43,6 +43,8 @@ testcontainers = {extras = ["minio"], version = "^4.10.0"}
 coverage-badge = "^1.1.2"
 # coverage-badge imports pkg_resources, removed in setuptools 82
 setuptools = "<82"
+# Lua runtime for fakeredis EVAL — dispatch Lua script tests (spec §16)
+lupa = "^2.8"
 
 [build-system]
 requires = ["poetry-core"]
diff --git a/tests/test_dispatch_job.py b/tests/test_dispatch_job.py
new file mode 100644
index 0000000..3af8665
--- /dev/null
+++ b/tests/test_dispatch_job.py
@@ -0,0 +1,389 @@
+import threading
+
+import pytest
+
+from config import settings
+from mongo.utils import RedisCache
+from dispatch import params, redis_keys, scripts
+from dispatch import job as job_mod
+
+
+@pytest.fixture(autouse=True, scope='session')
+def setup_minio():
+    # Shadow conftest's Docker/MinIO session fixture: these job-lifecycle unit
+    # tests touch only fakeredis, so they must not require a container engine.
+    yield
+
+
+@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
+    job_mod._cache = None
+    yield
+    RedisCache.POOL = None
+    job_mod._cache = None
+
+
+def _enqueue(submission_id='sn_1', problem_id='pb_1', language=2):
+    return job_mod.enqueue_job(
+        submission_id=submission_id,
+        problem_id=problem_id,
+        language=language,
+        code_minio_path=f'{submission_id}/main.py',
+        checker='diff',
+        tasks_meta_json='[]',
+    )
+
+
+def _client():
+    return job_mod._redis()
+
+
+def _hash(job_id):
+    raw = _client().hgetall(redis_keys.job(job_id))
+    return {k.decode(): v.decode() for k, v in raw.items()}
+
+
+def _reclaim(job_id, runner, now):
+    return scripts.load(_client()).reclaim_expired(
+        keys=[redis_keys.JOBS_LEASED,
+              redis_keys.job(job_id)],
+        args=[job_id, runner, now, params.LEASE_TTL_SEC, params.MAX_ATTEMPTS],
+    )
+
+
+def _expire(job_id, deadline='0'):
+    _client().hset(redis_keys.job(job_id), 'lease_deadline', deadline)
+
+
+# --- enqueue / claim happy path -----------------------------------------
+
+
+def test_enqueue_writes_hash_pointer_and_pending(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 1_000_000.0)
+    job_id = _enqueue(submission_id='sn_7', problem_id='pb_3', language=1)
+
+    assert job_id.startswith('jb_')
+    h = _hash(job_id)
+    assert h['submission_id'] == 'sn_7'
+    assert h['problem_id'] == 'pb_3'
+    assert h['language'] == '1'
+    assert h['code_minio_path'] == 'sn_7/main.py'
+    assert h['checker'] == 'diff'
+    assert h['tasks_meta_json'] == '[]'
+    assert h['state'] == 'pending'
+    assert h['attempts'] == '0'
+    assert h['leased_by'] == ''
+    assert h['lease_deadline'] == '0'
+    assert h['last_error'] == ''
+    assert float(h['created_at']) == 1_000_000.0
+
+    client = _client()
+    assert client.get(
+        redis_keys.submission_current_job('sn_7')).decode() == job_id
+    assert client.lrange(redis_keys.JOBS_PENDING, 0, -1) == [job_id.encode()]
+
+
+def test_claim_happy_path_moves_pending_to_leased(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 2_000_000.0)
+    job_id = _enqueue(submission_id='sn_1')
+
+    payload = job_mod.claim_next_job('rn_a')
+
+    assert payload['job_id'] == job_id
+    assert payload['submission_id'] == 'sn_1'
+    assert payload['state'] == 'leased'
+    assert payload['attempts'] == '1'
+    assert payload['leased_by'] == 'rn_a'
+    assert float(
+        payload['lease_deadline']) == 2_000_000.0 + params.LEASE_TTL_SEC
+
+    client = _client()
+    # membership moved pending -> leased
+    assert client.lrange(redis_keys.JOBS_PENDING, 0, -1) == []
+    assert client.sismember(redis_keys.JOBS_LEASED, job_id)
+
+
+def test_claim_empty_queue_returns_none():
+    assert job_mod.claim_next_job('rn_a') is None
+
+
+def test_claim_is_fifo(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 10.0)
+    first = _enqueue(submission_id='sn_1')
+    second = _enqueue(submission_id='sn_2')
+
+    assert job_mod.claim_next_job('rn_a')['job_id'] == first
+    assert job_mod.claim_next_job('rn_b')['job_id'] == second
+
+
+# --- INV4 currency (dispatch-side destroy) ------------------------------
+
+
+def test_rejudge_supersedes_stale_job_is_destroyed(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 100.0)
+    stale = _enqueue(submission_id='sn_9')
+    # rejudge: a newer job for the same submission moves the current_job pointer
+    current = _enqueue(submission_id='sn_9')
+    assert stale != current
+
+    # claim pops the stale one first (FIFO tail), destroys it, returns current
+    payload = job_mod.claim_next_job('rn_a')
+    assert payload['job_id'] == current
+
+    client = _client()
+    # stale hash destroyed; not leased
+    assert client.exists(redis_keys.job(stale)) == 0
+    assert not client.sismember(redis_keys.JOBS_LEASED, stale)
+    # current is the one leased
+    assert client.sismember(redis_keys.JOBS_LEASED, current)
+    assert client.get(
+        redis_keys.submission_current_job('sn_9')).decode() == current
+
+
+# --- renew --------------------------------------------------------------
+
+
+def test_renew_extends_deadline_for_owner(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 500.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_owner')  # deadline = 530
+
+    monkeypatch.setattr(job_mod, '_now', lambda: 800.0)
+    assert job_mod.renew_lease('rn_owner', job_id) is True
+    assert float(
+        _hash(job_id)['lease_deadline']) == 800.0 + params.LEASE_TTL_SEC
+
+
+def test_renew_wrong_runner_refused_deadline_untouched(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 500.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_owner')
+    before = _hash(job_id)['lease_deadline']
+
+    monkeypatch.setattr(job_mod, '_now', lambda: 800.0)
+    assert job_mod.renew_lease('rn_intruder', job_id) is False
+    assert _hash(job_id)['lease_deadline'] == before
+
+
+def test_renew_unknown_job_returns_false():
+    assert job_mod.renew_lease('rn_owner', 'jb_missing') is False
+    # must not resurrect anything
+    assert _client().exists(redis_keys.job('jb_missing')) == 0
+    assert not _client().sismember(redis_keys.JOBS_LEASED, 'jb_missing')
+
+
+# --- INV2 reclaim eligibility (lease_deadline only) ---------------------
+
+
+def test_reclaim_uses_only_lease_deadline_not_runner_liveness(monkeypatch):
+    # A job whose runner is very much "alive" (its alive key is present) but whose
+    # lease has expired MUST be reclaimable — runner liveness never participates.
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_alive')  # leased, deadline = 1030
+
+    client = _client()
+    client.set(redis_keys.runner_alive('rn_alive'), '1', ex=30)  # "alive"
+    assert client.exists(redis_keys.runner_alive('rn_alive')) == 1
+
+    # lease expired at now=2000 (> 1030) → reclaimable despite the alive key
+    assert _reclaim(job_id, 'rn_new', now=2000.0) == 1
+    h = _hash(job_id)
+    assert h['leased_by'] == 'rn_new'
+    assert h['attempts'] == '2'
+    assert float(h['lease_deadline']) == 2000.0 + params.LEASE_TTL_SEC
+
+
+def test_reclaim_not_expired_returns_zero(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')  # deadline = 1030
+
+    # now still before the deadline → not an orphan
+    assert _reclaim(job_id, 'rn_new', now=1020.0) == 0
+    assert _hash(job_id)['leased_by'] == 'rn_a'
+
+
+def test_reclaim_missing_or_unleased_returns_zero():
+    # never leased
+    assert _reclaim('jb_ghost', 'rn_new', now=9999.0) == 0
+
+
+# --- double-reclaim race ------------------------------------------------
+
+
+def test_reclaim_renewed_deadline_blocks_late_reclaimer(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')  # deadline = 1030
+
+    # both runners see the same expired lease at now=2000
+    assert _reclaim(job_id, 'rn_x', now=2000.0) == 1
+    # the winner renewed the deadline to 2030, so the loser now sees a live lease
+    assert _reclaim(job_id, 'rn_y', now=2000.0) == 0
+    assert _hash(job_id)['leased_by'] == 'rn_x'
+
+
+def test_concurrent_double_reclaim_exactly_one_winner(monkeypatch):
+    # Two runners race to reclaim the same expired job. The whole transition
+    # is one Lua script (atomic on the server), so no interleaving can produce
+    # two winners or a double attempts bump — assert the outcome is exactly
+    # one 1 and one 0, order-independent.
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')  # attempts = 1, deadline = 1030
+
+    reclaim = scripts.load(_client()).reclaim_expired
+    barrier = threading.Barrier(2)
+    results = {}
+
+    def racer(runner_id):
+        barrier.wait()  # line up both reclaims to fire together
+        results[runner_id] = reclaim(
+            keys=[redis_keys.JOBS_LEASED,
+                  redis_keys.job(job_id)],
+            args=[
+                job_id, runner_id, 2000.0, params.LEASE_TTL_SEC,
+                params.MAX_ATTEMPTS
+            ],
+        )
+
+    threads = [
+        threading.Thread(target=racer, args=(rn, )) for rn in ('rn_x', 'rn_y')
+    ]
+    for t in threads:
+        t.start()
+    for t in threads:
+        t.join()
+
+    assert sorted(results.values()) == [0, 1]  # exactly one winner
+    h = _hash(job_id)
+    winner = next(rn for rn, r in results.items() if r == 1)
+    assert h['leased_by'] == winner
+    assert h['attempts'] == '2'  # bumped exactly once (was 1 after claim)
+    assert float(h['lease_deadline']) == 2000.0 + params.LEASE_TTL_SEC
+
+
+# --- INV5 attempts boundary ---------------------------------------------
+
+
+def test_attempts_accumulate_and_converge_at_max(monkeypatch):
+    assert params.MAX_ATTEMPTS == 3
+    monkeypatch.setattr(job_mod, '_now', lambda: 0.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')  # attempts = 1, deadline = 30
+    assert _hash(job_id)['attempts'] == '1'
+
+    # reclaim twice, each after the prior lease expires
+    assert _reclaim(job_id, 'rn_b', now=100.0) == 1  # attempts = 2
+    assert _hash(job_id)['attempts'] == '2'
+    assert _reclaim(job_id, 'rn_c', now=200.0) == 1  # attempts = 3
+    assert _hash(job_id)['attempts'] == '3'
+
+    # at MAX_ATTEMPTS the next reclaim returns -1 and touches nothing
+    before = _hash(job_id)
+    assert _reclaim(job_id, 'rn_d', now=300.0) == -1
+    assert _hash(job_id) == before
+    # job hash + jobs:leased membership remain in place (JE landing is slice 3)
+    assert _client().sismember(redis_keys.JOBS_LEASED, job_id)
+
+
+# --- time gate ----------------------------------------------------------
+
+
+def test_orphan_scan_is_time_gated(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')  # leased, deadline = 1030
+    _expire(job_id, deadline='0')  # force the lease expired
+
+    client = _client()
+    # Hold the gate manually so the next claim's scan is skipped.
+    client.set(redis_keys.DISPATCH_LAST_RECOVERY,
+               '1',
+               ex=params.ORPHAN_SCAN_INTERVAL_SEC)
+
+    monkeypatch.setattr(job_mod, '_now', lambda: 5000.0)
+    # gate held → scan skipped → expired job stays with rn_a, nothing to hand out
+    assert job_mod.claim_next_job('rn_b') is None
+    assert _hash(job_id)['leased_by'] == 'rn_a'
+
+    # release the gate → next claim scans, reclaims the expired job to the caller
+    client.delete(redis_keys.DISPATCH_LAST_RECOVERY)
+    payload = job_mod.claim_next_job('rn_b')
+    assert payload['job_id'] == job_id
+    assert payload['leased_by'] == 'rn_b'
+
+
+def test_orphan_scan_reclaim_handed_to_caller(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')  # leased, deadline = 1030
+    _expire(job_id, deadline='0')
+
+    # The first claim already took the gate (its EX runs on fakeredis wall-clock,
+    # not our monkeypatched _now); release it so the next claim scans.
+    _client().delete(redis_keys.DISPATCH_LAST_RECOVERY)
+    monkeypatch.setattr(job_mod, '_now', lambda: 5000.0)
+    payload = job_mod.claim_next_job('rn_b')
+    assert payload['job_id'] == job_id
+    assert payload['leased_by'] == 'rn_b'
+    assert payload['attempts'] == '2'
+
+
+# --- corrupted / ghost state (data-loss defensive paths) -----------------
+
+
+def test_claim_destroys_job_missing_submission_id_and_continues(monkeypatch):
+    # A hash without submission_id cannot have its currency verified: the claim
+    # loop must destroy it and keep popping — not error out after the RPOP
+    # (which would silently lose the job id from the queue).
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    corrupted = _enqueue(submission_id='sn_bad')
+    healthy = _enqueue(submission_id='sn_good')
+    _client().hdel(redis_keys.job(corrupted), 'submission_id')
+
+    # FIFO: the corrupted job pops first, gets destroyed, loop continues
+    payload = job_mod.claim_next_job('rn_a')
+    assert payload['job_id'] == healthy
+    assert _client().exists(redis_keys.job(corrupted)) == 0
+    assert _client().llen(redis_keys.JOBS_PENDING) == 0
+
+
+def test_claim_missing_attempts_field_falls_back_to_zero(monkeypatch):
+    # attempts is a counter, not identity: if the field is lost the job is
+    # still judgeable — claim proceeds as if attempts were 0.
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    _client().hdel(redis_keys.job(job_id), 'attempts')
+
+    payload = job_mod.claim_next_job('rn_a')
+    assert payload['job_id'] == job_id
+    assert _hash(job_id)['attempts'] == '1'
+
+
+def test_reclaim_reaps_ghost_member_when_hash_evaporated(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')
+    _client().delete(redis_keys.job(job_id))  # simulate partial data loss
+
+    assert _reclaim(job_id, 'rn_new', now=9999.0) == 0
+    # the dangling jobs:leased member was reaped — no eternal rescanning
+    assert not _client().sismember(redis_keys.JOBS_LEASED, job_id)
+
+
+def test_renew_reaps_ghost_member_when_hash_evaporated(monkeypatch):
+    monkeypatch.setattr(job_mod, '_now', lambda: 1000.0)
+    job_id = _enqueue()
+    job_mod.claim_next_job('rn_a')
+    _client().delete(redis_keys.job(job_id))  # simulate partial data loss
+
+    assert job_mod.renew_lease('rn_a', job_id) is False
+    assert not _client().sismember(redis_keys.JOBS_LEASED, job_id)