Farpost · Dispatch Architecture
$ A Generic Dispatch Loop, Built Before It Needed to Be
Farpost dispatches work to professionals — claims first, with inspections and other request types planned to follow. Rather than write a claim-specific dispatch function and duplicate it later for inspections, the dispatch loop was built once, against a generic "dispatchable request" shape, before a second consumer of it existed.
The subtle part is how it decides who's already been tried. Reputation-based ranking is live — a professional's position in the list can shift between attempts — so re-dispatch has to exclude candidates who already tried by who they are, not by where they landed in the last ranking. Get that backwards, and a re-rank between attempts can silently skip the wrong person.
async def dispatch_work_request(
request: DispatchableRequest,
notify: NotifyCallable,
) -> None:
"""Core dispatch loop. Called via asyncio.create_task — never awaited directly.
1. Find eligible candidates
2. Rank by reputation
3. Exclude already-tried candidates by identity (not ranking position — ranking
can shift between attempts since it's reputation-based and live)
4. Create WorkRequestAttempt
5. Increment attempt counter on request
6. Call notify(candidate, attempt)
7. Schedule an instant in-process timeout check (supplements, doesn't replace,
the periodic check_request_timeouts sweep)
"""
repo = WorkRequestAttemptRepository()
candidates = await find_candidates(
request.postal_prefix, request.target_role, request.required_capabilities
)
if not candidates:
await request.on_no_candidates()
return
ranked = await rank_candidates(candidates)
existing_attempts = await repo.list_for_request(request.request_id)
already_tried = {a.candidate_slug for a in existing_attempts}
eligible = [c for c in ranked if c.slug not in already_tried]
if not eligible:
await request.on_all_exhausted()
return
candidate = eligible[0]
# ... create WorkRequestAttempt, notify(candidate, attempt) ...
asyncio.get_event_loop().call_later(
request.timeout_minutes * 60,
lambda: asyncio.create_task(_check_attempt_timeout(str(attempt.id))),
)The fix
already_tried is built from each attempt's candidate_slug — a stable identity — and checked against ranked candidates by that same slug, not by list index. Two independent completion checks both converge on the same WorkRequestAttemptRepository: an in-process timer scheduled the moment a candidate is notified, and a separate periodic sweep. If both fire for the same attempt, the second one finds nothing left to do and becomes a no-op rather than dispatching twice.
Why this matters
This is what generalizing infrastructure looks like when it's done deliberately rather than forced: a structural Protocol type let inspections and future request types plug into the same dispatch loop with no shared base class, and the exclusion-by-identity detail — easy to get wrong on a live, reordering ranking — was correct on the first pass. That's judgment under ambiguity: building for a second consumer that doesn't exist yet, without guessing wrong about the one subtle correctness detail that would matter once it did.