feat(runner): add job poller and result sender

[as535364]

closes Normal-OJ/Normal-OJ#69

Slice 3/4 of the pull-based runner. Dark PR: purely additive, nothing is wired into the process yet (main.py entrypoint is slice 4). The legacy push path is untouched.

What

• runner/client.py: BackendClient gains next_job (200 → JobPayload, 204 → None), complete (expects 204) and abort (expects 202), per spec §7.3–§7.5. JobPayload normalizes problem_id to int at the wire boundary.
• runner/poller.py: PollerThread polls only when there is spare capacity (len(tracker) < max_concurrent_jobs), adds the claimed job to the tracker before prep so the heartbeat renews the lease during downloads, and retries prep locally (3 attempts, backoff 1s/2s) before queueing abort(prep_failed). Dispatch (Dispatcher.handle) gets exactly one shot — see design notes. prepare_job reuses the existing ensure_testdata + file_manager.extract helpers and downloads code from the presigned code_url. Unexpected errors in the poll loop are logged and never kill the thread.
• runner/result_sender.py: single reporting channel. All complete/abort reports retry with exponential backoff (1→2→4→8→16s, max 5 retries); 409/404 are terminal drops; a 400 on complete becomes abort(rejected) (spec §7.5, INV5); complete exhaustion keeps a local backup (file_manager.backup_data) and gives up — lease-expiry reclaim is the safety net. stop() drains whatever is already queued before exiting, which is the building block for slice 4's SIGTERM drain.

Design notes

• The poller's prep_failed abort goes through the result queue instead of calling the backend directly, so there is exactly one reporting channel and the drain semantics of slice 4 cover it for free.
• Aborts finalize local state (job dir + tracker entry) before the request is sent. A 202 means the backend has already requeued the job, so this same runner may re-claim the same job_id immediately; finalizing late would let the old claim's cleanup delete the new claim's dir and tracking (ABA race, found in review). rejected aborts move the dir aside via backup_data (evidence of what the backend refused); prep_failed aborts clean it. Consequence: abort-retry exhaustion no longer backs up — the dir was finalized before the first attempt. The complete path keeps send-then-finalize: success deletes the job backend-side (no requeue), and holding the tracker entry during retries keeps the lease renewed so a delivered-late result needs no re-execution.
• Dispatch is never retried. Dispatcher.handle() is not atomic: on queue.Full it releases its bookkeeping but already-enqueued task entries stay in the queue; calling handle() again for the same job_id would revive them and duplicate execution (found in review). A dispatch failure aborts straight away (prep_failed, counts toward attempts per spec §7.5). The remaining exposure — a later re-claim of the same job by the same runner while orphaned entries are still queued — closes in slice 4, which makes handle() atomic. Slice 4 should also make the capacity gate account for task-queue space (max_concurrent_jobs counts jobs, QUEUE_SIZE counts tasks), which is what makes queue.Full reachable under load in the first place.
• 401 is deliberately not terminal in poller/sender retries: heartbeat owns 401 fail-fast (spec §10).

Testing

• 55 unit tests across the client/poller/sender suites (no docker, no network): wire contract, poller outcome branches (capacity gate, 204, errors, prep retry/exhaustion, single-shot dispatch, tracker ordering, thread survival), sender retry matrix (terminal/retry/exhaustion × abort reasons), abort finalize-before-send ordering, drain-on-stop behavior.
• Full local suite: 91 passed; only the two docker-daemon-dependent executor tests fail locally (no daemon), they run in CI.

[as535364]

Codex:

Review findings

[P1] abort / reclaim 有 ABA race

abort() 收到 202 時，backend 已把同一個 job_id 放回 pending；但 _process() 是之後才執行 cleanup 與 tracker.remove()。此時 poller 只要還有 capacity，就可能由同一 runner 立刻 reclaim 同一 job。由於 tracker 是 set，新的 add(job_id) 不會建立新的 generation；舊 sender 接著會刪掉新 job dir，並移除新 lease 的 tracking，導致執行失敗或 lease 過期。

建議讓 next_job claim 與 abort finalization 對同一 runner 串行化，或讓 tracker／cleanup 帶 lease generation，並補一個可重現此 interleaving 的 concurrency test。

相關位置：runner/result_sender.py:104-114、runner/poller.py:93-106。

[P1] dispatch retry 不是 atomic，可能重複執行 task

Dispatcher.handle() 會逐筆 put_nowait()；若中途 queue.Full，只會 release(job_id)，先前已 enqueue 的 entries 並未 rollback。poller 目前把 _prepare() 與 _dispatch() 放在同一個 retry scope，接著會刪除／重建 job dir 並再次 dispatch。舊 entries 之後看到重新建立的 job bookkeeping，就會與新 entries 一起執行，造成 duplicate execution 或在目錄重建期間失敗。

建議分開 prep retry 與 dispatch failure，並在 retry dispatch 前提供 atomic capacity reservation 或完整 rollback。

相關位置：runner/poller.py:108-124、dispatcher/dispatcher.py:101-125。

[as535364]

Claude: 兩條 P1 已在 b7ab2a4 處理。

P1-1（abort/reclaim ABA race）：成立。修法是把 abort 的本地收尾（目錄＋tracker）搬到送出之前。202 回來時 backend 已經把 job 推回 pending，同一台 runner 隨時可能重撈；收尾先做完，重撈的瞬間本地就沒有舊狀態可以誤刪。幾個連帶調整：rejected 改走 backup_data，backend 拒收的 payload 目錄留著查 serializer 問題，move 走一樣不擋新 claim；abort 重送耗盡不再 backup，因為目錄在第一次送出前就處理掉了；收尾失敗照樣送 abort，不然 job 會卡到 lease 過期。complete 路徑不動：成功是刪 job 不會 requeue，retry 期間留著 tracker 還能維持續租，晚到的結果不用重算。測試補的是順序斷言（cleanup → tracker.remove → send），沒有做雙 thread 的 interleaving 重現，那種測試非決定性，守住順序不變量就夠了。

P1-2（dispatch 非 atomic）：也成立，而且觸發條件比 review 描述的更寬：capacity gate 數的是 job（max_concurrent_jobs=8），queue 裝的是 task（QUEUE_SIZE=16），一題多測資時滿載就會 queue.Full。這個 PR 先止血：dispatch 只呼叫一次，失敗直接 abort(prep_failed)，不再重跑 prepare＋handle，因為每呼叫一次 handle 就多一次讓部分入列的孤兒 entries 復活的機會。殘餘窗口（abort 後同機重撈、孤兒還在 queue）要等 handle() 原子化才關得掉，那是 dispatcher 的修改，歸 keystone。已經連同 capacity gate 應改以 task 容量為準記到 Normal-OJ/Normal-OJ#70。