erd_search.py

#!/usr/bin/env python3.13
"""erd_search.py — Parallel ERD_ALL precache CLI.

Subcommands
-----------
start           Start the supervisor and the report web server via systemd
                (systemctl --user start), unless each is already running.
                --swarm-only starts the supervisor alone; --web-only starts
                the report web server alone.
stop            Stop the supervisor and the report web server via systemd
                (systemctl --user stop).  --swarm-only stops the supervisor
                alone; --web-only stops the report web server alone.
restart         Restart the supervisor and the report web server via systemd
                (systemctl --user restart): a stop followed by a start in one
                step, per service.  --swarm-only restarts the supervisor
                alone; --web-only restarts the report web server alone.

view            Shared swarm reports in text, JSON, or watched JSON Lines.

run             Start the supervisor directly (without systemd), for
                development or one-shot use.  All output goes to
                runtime/erd_search.log.

queue           Queue mutation operations.
queue add       Add branches for one or more words to the work queue.
                Idempotent: existing branches are never duplicated; priority
                is upgraded if the new request is higher.  Words are queued on
                a descending priority ladder in the order given, --priority-step
                apart, so the swarm works them one at a time, and the batch is
                appended below every word the queue still owes work unless
                --priority names a rung outright.  If there is not enough room
                below the queue's current floor to seat the batch on distinct
                rungs, every unfinished request is shifted upward first to
                reclaim it, rather than clamping the new batch onto the
                priority floor; reports the headroom left below the batch it
                seats.  With --words-file,
                --priority-words restricts the ladder to a subset of the file's
                words.  --delete-erd-cache forces a recompute of branches that
                are already cached, discarding both the cache entry and the
                queue-side work state of any branch already completed (its
                claims, bundle stats and active-branch row) so it is claimable
                again; a branch a worker is still solving is left alone.
                Otherwise, branches with reusable cached results are not
                queued.  Reports how many branches are new versus already
                queued, and how many were already solved.

queue clear     Wipe all queue state (pending branches, active state, candidate
                claims, heartbeats).  Does not touch the ERD cache.

queue remove    Remove a pending branch from the queue.  Use --force to also
                cancel an in-progress branch (workers move on after their
                current candidate evaluation completes).

queue priority  Change the priority of a queued branch.  Higher numbers are
                worked sooner; 0 is the default.

queue opener-priority
                Change the requested priority of an opener-work request by
                word.  Takes effect immediately for both its pending roots
                and its active/promoted descendants.

epoch           Show or change the telemetry epoch used to compare swarm
                telemetry from one claiming regime.

queue set-disk-stop
                Keep the swarm down across reboots and systemd restarts.

queue reconcile-orphaned-ownership
                Demote open owned branches whose opener-work membership was
                lost while they were still open (its promoting candidate was
                retracted, or its owning request completed moments before it
                attached), making them claimable again.  The run loop
                self-heals this on every membership resolution; this command
                is for branches stranded before that (e.g. accumulated while
                the swarm was down).

For exporting a trimmed cache snapshot to sync to the iPhone, or importing
one from another machine, see export_cache.py and import_cache.py — the
cache is shared with interactive play (wordle.py), not swarm-specific.
"""

from __future__ import annotations

import argparse
import json
import logging
import multiprocessing
import os
import sqlite3
import signal
import sys
import time
from datetime import datetime

from cache_sqlite import ScoreCache
from hint_cache import HintCacheError, open_hint_cache
from report_model import (
    BRANCH_STATUSES,
    BRANCH_WORKER_STATUSES,
    OVERVIEW_BRANCH_STATUSES,
    OVERVIEW_BRANCH_WORKER_STATUSES,
    applied_branch_filters,
    is_overview_request,
    OPENER_SORT_FIELDS,
    ROOT_PROGRESS_SORT_FIELDS,
    OPENER_STATES,
    ReportFilters,
    ReportRequest,
    WORKER_LIVENESS_SECONDS,
    parse_branch_filter,
    parse_report_branch_target,
    parse_rich_spine as _parse_spine,
    validate_report_request,
)
from runtime_paths import (
    DEFAULT_ANSWER_LIST_PATH,
    DEFAULT_CACHE_PATH,
    DEFAULT_CANDIDATE_LIST_PATH,
    DEFAULT_QUEUE_PATH,
    DEFAULT_SEARCH_LOG_PATH,
    ensure_runtime_dir,
)
from wordle_engine import ERD_ALL, GAME_GUESSES, ResponseCache, load_word_list
from erd_queue import (
    DISK_STOP_FRACTION,
    ERDQueue,
    QUEUE_WAL_HARD_CEILING_BYTES,
    OPENER_PRIORITY_MAX,
    OPENER_PRIORITY_MIN,
    check_opener_priority_range,
    disk_stats,
    encode_subset,
)
import erd_swarm

ANSWER_FILE = DEFAULT_ANSWER_LIST_PATH
WORDS_FILE = DEFAULT_CANDIDATE_LIST_PATH
DEFAULT_CACHE = DEFAULT_CACHE_PATH
DEFAULT_QUEUE = DEFAULT_QUEUE_PATH

logger = logging.getLogger('wordle')


def _view_watch_interval(value):
    interval = float(value)
    if interval < 0.2:
        raise argparse.ArgumentTypeError("watch interval must be at least 0.2 seconds")
    return interval


def _comma_separated_filter(value, option_name, choices):
    try:
        return parse_branch_filter(value, option_name, choices)
    except ValueError as error:
        raise argparse.ArgumentTypeError(str(error)) from error


def _branch_status_filter(value):
    return _comma_separated_filter(value, "branch status", BRANCH_STATUSES)


def _branch_worker_status_filter(value):
    return _comma_separated_filter(value, "branch worker status", BRANCH_WORKER_STATUSES)


def _opener_state_filter(value):
    return _comma_separated_filter(value, "opener state", OPENER_STATES)


def cmd_view(args):
    from report_terminal import run_view

    run_view(args)


# ---------------------------------------------------------------------------
# queue add
# ---------------------------------------------------------------------------

DEFAULT_PRIORITY_STEP = 5


def priority_ladder(words, top_priority, step):
    """Map words to descending opener priorities, first word highest.

    Words tied at one priority are all eligible to start at once, so a
    blocked worker widens the batch by starting whichever of them has no
    branch open yet rather than opening another branch of the word already
    running.  A ladder breaks every tie, which keeps the swarm on one word
    until that word runs out of claimable work.

    Adjacent words differ by `step`, leaving unused values between rungs so
    a word can be reordered later with `queue opener-priority` without
    disturbing its neighbours.  A step of 0 puts every word on
    `top_priority`.

    Rungs below OPENER_PRIORITY_MIN clamp onto it, so a list too long to seat
    on distinct rungs gives them to the leading words and ties the remainder
    on the minimum.  The tail is undifferentiated but still ranks below every
    seated word.
    """
    if step <= 0:
        return {word: top_priority for word in words}
    return {word: max(OPENER_PRIORITY_MIN, top_priority - step * index)
            for index, word in enumerate(words)}


def _make_room_for_append(queue, lowest_queued, step, n_words):
    """Reclaim room for an append whose ladder would clamp onto
    OPENER_PRIORITY_MIN, by shifting every unfinished opener-work request
    upward instead of ratcheting the append's own floor below the minimum.

    Called only when `queue add` has no explicit --priority and the queue
    already owes work.  Without this, `lowest_queued - 1` is the append's
    only ceiling and it only ever falls as batches accumulate, making
    OPENER_PRIORITY_MAX practically unreachable on a queue that never fully
    drains (issue #276).  Shifting the unfinished set up by the shortfall
    instead preserves every request's relative order and every tie, and
    frees exactly the room this batch needs.

    Returns (lowest_queued, shift): the (possibly raised) floor to seat the
    new batch below, and how far unfinished work was shifted (0 if it
    already fit).  Raises ValueError, before touching anything, if shifting
    every unfinished request up to OPENER_PRIORITY_MAX still cannot make
    room — the range is already fully occupied.
    """
    required_span = step * max(0, n_words - 1)
    naive_floor = (lowest_queued - 1) - required_span
    if naive_floor >= OPENER_PRIORITY_MIN:
        return lowest_queued, 0
    max_unfinished = queue.max_unfinished_opener_priority()
    headroom_above = OPENER_PRIORITY_MAX - max_unfinished
    shortfall = OPENER_PRIORITY_MIN - naive_floor
    shift = min(shortfall, headroom_above)
    if naive_floor + shift < OPENER_PRIORITY_MIN:
        needed_ceiling = lowest_queued - 1 + (shortfall - headroom_above)
        raise ValueError(
            f'{n_words:,} words at step {step:,} need a ladder reaching '
            f'{shortfall:,} below the current queue floor of '
            f'{lowest_queued:,}; shifting every unfinished opener-work '
            f'request up to the maximum {OPENER_PRIORITY_MAX:,} only frees '
            f'{headroom_above:,}.  The range would need a ceiling of at '
            f'least {needed_ceiling:,} to hold both.  Use queue '
            f'opener-priority to make room by hand, or pass --priority to '
            f'place this batch deliberately.')
    queue.shift_unfinished_opener_priorities(shift)
    return lowest_queued + shift, shift


def ladder_top_priority(lowest_queued, explicit_priority, step, n_words):
    """The rung a batch's first word takes.

    `queue add` appends: with no --priority the batch descends from just
    below `lowest_queued`, the lowest priority the queue still owes work, so
    it preempts nothing already queued and leaves the space beneath it for
    the next batch.  Ladders therefore run downward from the top of the range
    rather than upward from its floor, which is what keeps repeated appends
    from exhausting the space.  `lowest_queued` is None when nothing is owed,
    and the batch takes the top of the range.

    An explicit --priority names the *last* word's rung instead, and the
    batch is seated high enough for the ladder to land on it — that is how a
    batch is deliberately placed ahead of queued work.  Raises ValueError if
    the range cannot hold a ladder whose last rung is the one named: silently
    seating the batch lower would hand back something other than the priority
    the caller asked for.
    """
    if explicit_priority is not None:
        top_priority = explicit_priority + step * max(0, n_words - 1)
        if top_priority > OPENER_PRIORITY_MAX:
            raise ValueError(
                f'--priority {explicit_priority:,} needs a ladder reaching '
                f'{top_priority:,} for {n_words:,} words at step {step:,}, '
                f'above the maximum {OPENER_PRIORITY_MAX:,}.  Lower '
                f'--priority, use a smaller --priority-step, or add fewer '
                f'words at a time.')
        return top_priority
    if lowest_queued is None:
        return OPENER_PRIORITY_MAX
    return max(OPENER_PRIORITY_MIN, lowest_queued - 1)


def invalidate_branches_for_recompute(queue, score_cache, branch_keys):
    """Drop each branch's cached result and the work state that would keep it
    unclaimable, so it is genuinely recomputed.

    Nothing above these branches needs invalidating alongside them.  A word's
    own ERD is folded from its response groups' cached results on every read,
    so a group deleted here reports as unresolved immediately, and the word
    reads as pending until it is solved again.

    A branch the queue finished keeps a `done` pending row, and can still
    hold candidate claims and an `active_branches` row.  Re-adding it leaves
    all three in place — `add_pending_many` carries priority forward but not
    status, `create_branch` ignores an existing `active_branches` row, and
    claims already marked done finalize the branch again without evaluating
    anything.

    A branch carrying an *open* active_branches row is left entirely alone,
    cache entry included.  A worker is solving it, and the two halves have to
    agree: deleting the cached result while refusing to requeue would leave
    exactly the branch this function exists to prevent — no cached answer and
    a `done` row no worker will claim.

    Returns (reset_count, busy_count).
    """
    reset_count = 0
    busy_count = 0
    for branch_key in branch_keys:
        active = queue.get_active_branch(branch_key)
        if active is not None and active['status'] == 'open':
            busy_count += 1
            continue
        score_cache.delete(branch_key, ERD_ALL)
        if queue.requeue_completed_branch(branch_key):
            reset_count += 1
    return reset_count, busy_count


def cmd_queue_add(args):
    """Add branches for one or more words (or a words-file) to the queue.

    With --word: adds all response branches for each given word with at
    least 2 answer words (and at most --max-branch-size, if given).  With
    --pattern as well: adds only that single branch per word.

    With --words-file: walks every word in the file, same as --word with
    that file's contents.  --priority-words marks a subset of those words as
    higher priority: they are queued at --priority while the rest are queued
    at 0.

    Words are queued on a descending priority ladder in the order given, so
    the swarm works them one at a time rather than starting all of them at
    once; --priority-step sets the gap between rungs.  Without --priority the
    whole ladder sits below every request the queue still owes work, so a new
    batch never preempts one already queued; --priority names the last rung
    outright and is how a batch is put ahead of queued work.  With
    --priority-words, the ladder covers that subset and the rest stay flat at
    the minimum.  A word repeated in the input takes its first position and is
    queued once.

    An append (no --priority) whose ladder would run past the priority
    minimum first shifts every unfinished request upward by the shortfall,
    reclaiming exactly the room the new batch needs rather than ratcheting
    its own floor down or clamping onto the minimum — a queue fed
    continuously (never fully draining) therefore does not lose rungs to
    successive appends.  Reports the headroom left below the batch it seats,
    and raises before queuing anything if even that shift cannot make room.
    A batch that ends up with nothing to queue (every branch already cached,
    or none large enough) never reladders: already-queued priorities are
    left untouched, and the request cannot be refused at the ceiling for
    what is ultimately a no-op.

    Already-queued branches are never duplicated; their priority is upgraded
    if the new request is higher.  Branches with reusable cached results are
    already solved and are not queued, unless --delete-erd-cache is given.
    For each word, reports how many unresolved branches are new versus already
    queued, and how many response groups were already solved.
    --delete-erd-cache deletes each queued branch's existing ERD cache entry
    first, so it gets recomputed instead of being claimed and immediately
    marked done as already-cached.  For branches the queue already completed
    it also discards the work state that would otherwise keep them
    unclaimable — candidate claims, republish and hole rows, bundle stats,
    and the active-branch row — and returns them to 'pending'.  A branch a
    worker is still solving is skipped rather than pulled out from under it.
    """
    from wordle_ui import parse_pattern, fmt_pattern

    all_answers = load_word_list(ANSWER_FILE)
    if args.word:
        words_to_process = [word.strip().lower() for word in args.word]
    else:
        words_to_process = [word.strip().lower()
                            for word in load_word_list(args.words_file)]
    words_to_process = list(dict.fromkeys(words_to_process))

    candidate_words = set(load_word_list(WORDS_FILE))
    invalid_words = [word for word in words_to_process
                     if len(word) != 5 or word not in candidate_words]
    if invalid_words:
        invalid_display = ', '.join(sorted(set(invalid_words)))
        raise ValueError(
            f'invalid candidate word(s): {invalid_display}; expected '
            f'five-letter words from {WORDS_FILE}')

    priority_words = {w.strip().lower() for w in (args.priority_words or [])}
    if priority_words and not args.words_file:
        print('Warning: --priority-words only applies with --words-file; '
              'ignoring it.  Use --priority directly with --word.')
        priority_words = set()

    base_priority = (OPENER_PRIORITY_MIN if args.priority is None
                     else args.priority)
    check_opener_priority_range(base_priority)
    if args.priority_step < 0:
        raise ValueError('--priority-step must not be negative')

    score_cache = ScoreCache(args.cache, all_answers)
    rcache = ResponseCache(all_answers, score_cache)
    queue = ERDQueue(args.queue)

    unknown = priority_words - set(words_to_process)
    if unknown:
        print(f'Warning: priority words not in the word list: '
              f'{", ".join(sorted(unknown))}')

    # A branch reached by --word has guess_depth 1 (one guess played), so it
    # is solved at ROOT_BUDGET - 1 == GAME_GUESSES - 1.
    branch_budget = GAME_GUESSES - 1

    n_new = 0
    n_already_queued = 0
    n_already_solved = 0
    n_reset = 0
    n_busy = 0
    try:
        # First pass: work out what each word actually needs queuing --
        # cache state, and (under --delete-erd-cache) invalidation -- none of
        # which depends on priority.  This decides whether the batch has any
        # real work *before* the ladder is seated, so a batch that is fully
        # cached (nothing to queue) never shifts already-queued opener-work
        # priorities, and can't be refused at the ceiling for a no-op.
        word_plans = {}
        skip_message_by_word = {}
        for word in words_to_process:
            if args.pattern:
                code = parse_pattern(args.pattern)
                groups = rcache.group_words(word, all_answers)
                branch = groups.get(code, [])
                if len(branch) < 2:
                    skip_message_by_word[word] = (
                        f'{word.upper()} {fmt_pattern(code)}: '
                        f'{len(branch)} answer word(s) — nothing to queue.')
                    rows = []
                elif (args.max_branch_size is not None
                      and len(branch) > args.max_branch_size):
                    skip_message_by_word[word] = (
                        f'{word.upper()} {fmt_pattern(code)}: '
                        f'{len(branch)} words exceeds --max-branch-size '
                        f'{args.max_branch_size}, skipping.')
                    rows = []
                else:
                    rows = [(encode_subset(branch), len(branch), code)]
            else:
                groups = rcache.group_words(word, all_answers)
                rows = [
                    (encode_subset(branch), len(branch), code)
                    for code, branch in groups.items()
                    if len(branch) >= 2
                    and (args.max_branch_size is None
                         or len(branch) <= args.max_branch_size)
                ]

            if not rows:
                word_plans[word] = None
                continue

            branch_keys = [branch_key for branch_key, _count, _code in rows]
            cache_states = score_cache.report_branch_states(
                branch_keys, ERD_ALL, budget=branch_budget)
            already_cached_keys = {
                key for key, state in cache_states.items()
                if state['cache_state'] in ('exact', 'loss')}

            if args.delete_erd_cache:
                word_reset, word_busy = invalidate_branches_for_recompute(
                    queue, score_cache, branch_keys)
                n_reset += word_reset
                n_busy += word_busy
                rows_to_queue = rows
                already_solved_keys = set()
            else:
                rows_to_queue = [
                    row for row in rows if row[0] not in already_cached_keys]
                already_solved_keys = already_cached_keys

            already_queued_keys = set(
                queue.status_by_branch_keys(
                    [branch_key for branch_key, _count, _code
                     in rows_to_queue]))
            word_plans[word] = (rows_to_queue, already_queued_keys,
                               already_solved_keys)

        any_rows_to_queue = any(
            plan is not None and plan[0] for plan in word_plans.values())

        laddered_words = [word for word in words_to_process
                          if not priority_words or word in priority_words]
        lowest_queued = queue.lowest_unfinished_opener_priority()
        shift = 0
        if (args.priority is None and lowest_queued is not None
                and laddered_words and any_rows_to_queue):
            lowest_queued, shift = _make_room_for_append(
                queue, lowest_queued, args.priority_step, len(laddered_words))
        top_priority = ladder_top_priority(
            lowest_queued, args.priority, args.priority_step,
            len(laddered_words))
        ladder = priority_ladder(laddered_words, top_priority,
                                 args.priority_step)
        if laddered_words:
            if shift:
                print(f'Raised every unfinished opener-work request by '
                      f'{shift:,} to make room for this batch below priority '
                      f'{lowest_queued:,}.')
            # _make_room_for_append guarantees room below lowest_queued (or
            # raises before we get here), so an append never ties the incumbent.
            appended = args.priority is None and lowest_queued is not None
            placement = (f', behind queued work down to priority {lowest_queued:,}'
                         if appended else '')
            print(f'{laddered_words[0].upper()} first at priority '
                  f'{ladder[laddered_words[0]]:,}, '
                  f'{laddered_words[-1].upper()} last at '
                  f'{ladder[laddered_words[-1]]:,}{placement}.')
            floor = min(ladder.values())
            headroom = floor - OPENER_PRIORITY_MIN
            print(f'{headroom:,} priority value(s) of headroom remain below it, '
                  f'down to {OPENER_PRIORITY_MIN:,}.')
            on_floor = sum(1 for word in laddered_words if ladder[word] == floor)
            if args.priority_step and on_floor > 1:
                # Distinct rungs only collide once the ladder clamps, so a tie here
                # always sits on OPENER_PRIORITY_MIN with nothing beneath it.  A
                # smaller step cannot divide headroom that does not exist; raising
                # the incumbent or naming a priority is what actually works.
                print(f'Warning: {len(laddered_words):,} words do not fit on a '
                      f'ladder of step {args.priority_step:,} below priority '
                      f'{top_priority:,}; the last {on_floor:,} share priority '
                      f'{floor:,} and will start together.  Raise the queued work '
                      f'with queue opener-priority, or pass --priority to place '
                      f'this batch deliberately.')

        # Second pass: assign each word's rung and write it, now that the
        # ladder (and any reladdering) has been decided.
        for word in words_to_process:
            plan = word_plans[word]
            if plan is None:
                message = skip_message_by_word.get(word)
                if message is not None:
                    print(message)
                continue

            rows_to_queue, already_queued_keys, already_solved_keys = plan
            priority = ladder.get(word, OPENER_PRIORITY_MIN)
            rows_with_priority = [
                (branch_key, count, priority, word, code)
                for branch_key, count, code in rows_to_queue]
            if rows_with_priority:
                queue.add_pending_many(rows_with_priority)

            word_already_queued = len(already_queued_keys)
            word_new = len(rows_to_queue) - word_already_queued
            word_already_solved = len(already_solved_keys)
            n_new += word_new
            n_already_queued += word_already_queued
            n_already_solved += word_already_solved
            if not rows_to_queue:
                response_group_label = (
                    'response group' if word_already_solved == 1
                    else 'response groups')
                print(f'{word.upper()}: already solved — '
                      f'{word_already_solved:,} {response_group_label} already '
                      f'cached; nothing queued.')
            else:
                print(f'{word.upper()}: {len(rows_to_queue):,} branch(es) '
                      f'— {word_new:,} new, '
                      f'{word_already_queued:,} already queued, '
                      f'{word_already_solved:,} already solved.')

        total = queue.total_branches()
        n_added = n_new + n_already_queued
        if n_reset:
            print(f'\n{n_reset:,} completed branch(es) cleared for recompute '
                  f'(claims and active state discarded with the cache entry).')
        if n_busy:
            print(f'{n_busy:,} branch(es) are being solved right now and were '
                  f'left untouched, cache entry included.  Re-run once they '
                  f'finish to recompute them.')
        print(f'\n{n_added:,} branch(es) queued across '
              f'{len(words_to_process):,} word(s): {n_new:,} new, '
              f'{n_already_queued:,} already queued, '
              f'{n_already_solved:,} already solved.  '
              f'Queue total: {total:,}.')

    except KeyboardInterrupt:
        print('\nInterrupted.')
    finally:
        score_cache.checkpoint()
        score_cache.close()
        queue.close()


# ---------------------------------------------------------------------------
# queue clear
# ---------------------------------------------------------------------------

def cmd_queue_clear_disk_stop(args):
    """Release the disk-stop latch so `run` will start again."""
    queue = ERDQueue(args.queue)
    try:
        latch = queue.disk_stop()
        if latch is None:
            print('No disk-stop latch is set.')
            return
        used_fraction = disk_stats(args.queue)['used_fraction']
        queue.clear_disk_stop()
        print(f'Disk-stop latch cleared (was: {latch["reason"]}).  '
              f'Disk is now {100 * used_fraction:.1f}% full.')
        if used_fraction >= DISK_STOP_FRACTION:
            print(f'Warning: still at or above the '
                  f'{100 * DISK_STOP_FRACTION:.0f}% stop threshold — '
                  f'run will refuse to start until space is freed.',
                  file=sys.stderr)
    finally:
        queue.close()


def cmd_queue_set_disk_stop(args):
    """Latch the swarm down without replacing an existing latch reason."""
    queue = ERDQueue(args.queue)
    try:
        if queue.set_disk_stop_if_unset(args.reason):
            print(f'Disk-stop latch set: {args.reason}.')
            return
        latch = queue.disk_stop()
        print(f'Disk-stop latch is already set ({latch["reason"]}); '
              'it remains unchanged.')
    finally:
        queue.close()


def cmd_queue_clear(args):
    """Wipe all queue state (pending branches, active branches, candidate claims,
    heartbeats, and run metadata).  The ERD cache is not touched.

    Requires confirmation unless --yes is passed.
    """
    queue = ERDQueue(args.queue)
    try:
        counts = queue.counts_by_status()
        pending = counts.get('pending', 0)
        done = counts.get('done', 0)
        in_prog = len(queue.branches_in_progress())

        print(f'Queue: {pending:,} pending   {done:,} done   {in_prog} in progress')
        if not args.yes:
            ans = input('Clear all queue state? [y/N] ').strip().lower()
            if ans != 'y':
                print('Aborted.')
                return

        queue.clear()
        print('Queue cleared.')
    finally:
        queue.close()


# ---------------------------------------------------------------------------
# queue remove
# ---------------------------------------------------------------------------

def cmd_queue_remove(args):
    """Remove a branch from the pending queue.

    Only removes branches with status='pending'.  If the branch is currently
    in-progress (being worked by a worker), use --force to also cancel it by
    clearing its active_branches and candidate_claims rows so the worker's next heartbeat
    yields no further claims.  The worker will eventually notice the branch
    is gone and move on.
    """
    from wordle_ui import parse_pattern, fmt_pattern

    all_answers = load_word_list(ANSWER_FILE)
    word = args.word.strip().lower()
    code = parse_pattern(args.pattern)
    pat = fmt_pattern(code)

    sc = ScoreCache(args.cache, all_answers)
    rcache = ResponseCache(all_answers, sc)
    groups = rcache.group_words(word, all_answers)
    branch = groups.get(code, [])
    sc.close()

    if not branch:
        print(f'{word.upper()} {pat}: no branch.')
        return

    branch_key = encode_subset(branch)
    queue = ERDQueue(args.queue)

    active = queue.get_active_branch(branch_key)
    if active and not args.force:
        print(f'{word.upper()} {pat}: branch is in-progress.  '
              f'Use --force to also cancel the active work.')
        queue.close()
        return

    if active and args.force:
        # Atomically clear candidate claims, the active_branches row, and the
        # pending_branches row.  All three DELETEs run in one transaction so a
        # crash partway through cannot leave orphaned rows.  (remove_pending()
        # alone would silently no-op here because the pending row still has
        # status='in_progress' after the active state is cleared.)
        queue.cancel_active_branch(branch_key, remove_from_queue=True)
        queue.close()
        print(f'Cancelled in-progress work and removed {word.upper()} {pat} from queue.')
        return

    removed = queue.remove_pending(branch_key)
    queue.close()

    if removed:
        print(f'Removed {word.upper()} {pat} from queue.')
    else:
        print(f'{word.upper()} {pat}: not found in pending queue '
              f'(may already be done or not queued).')


# ---------------------------------------------------------------------------
# queue priority
# ---------------------------------------------------------------------------

def cmd_queue_priority(args):
    """Set the priority of a queued branch.

    Priority is an integer; higher numbers are worked sooner.  User-settable
    values in the range 0–999,999 are reserved for normal use: 0 = default,
    higher = sooner.
    """
    if getattr(args, 'opener_word', None):
        queue = ERDQueue(args.queue)
        try:
            updated = queue.set_ownerless_active_priority(
                args.opener_word.strip().lower(), args.priority)
        except ValueError as error:
            print(error)
            return
        finally:
            queue.close()
        print(f'{updated:,} ownerless open branch(es) for '
              f'{args.opener_word.strip().upper()}: priority set to '
              f'{args.priority}.')
        return

    from wordle_ui import parse_pattern, fmt_pattern

    all_answers = load_word_list(ANSWER_FILE)
    word = args.word.strip().lower()
    code = parse_pattern(args.pattern)
    pat = fmt_pattern(code)

    sc = ScoreCache(args.cache, all_answers)
    rcache = ResponseCache(all_answers, sc)
    groups = rcache.group_words(word, all_answers)
    branch = groups.get(code, [])
    sc.close()

    if not branch:
        print(f'{word.upper()} {pat}: no branch.')
        return

    branch_key = encode_subset(branch)
    queue = ERDQueue(args.queue)
    updated = queue.set_priority(branch_key, args.priority)
    queue.close()

    if updated:
        print(f'{word.upper()} {pat}: priority set to {args.priority}.')
    else:
        print(f'{word.upper()} {pat}: not found in pending queue.')


# ---------------------------------------------------------------------------
# queue opener-priority
# ---------------------------------------------------------------------------

def cmd_queue_opener_priority(args):
    """Set the requested priority of an opener-work request, by word.

    Resolves the word to an open (non-complete) opener_work_id via
    ERDQueue.opener_work_candidates() and defers to
    ERDQueue.set_opener_work_priority(), which applies the change to both the
    request's pending roots and its active/promoted descendants in one
    transaction.  A branch owned by more than one live request keeps the
    higher of their requested priorities (MAX(owner_priority) at the branch
    level), so lowering one request's priority does not necessarily lower a
    branch it shares with a higher-priority request.

    A word with more than one open request is ambiguous; --opener-work-id
    picks one.  --opener-work-id may also name a completed request directly,
    which is reported as such rather than as "not found".  A word whose
    requests are all complete is reported distinctly from a word with none.
    """
    word = args.word.strip().lower()

    try:
        check_opener_priority_range(args.priority)
    except ValueError as error:
        print(error)
        return

    queue = ERDQueue(args.queue)
    try:
        all_rows = {row['opener_work_id']: row
                    for row in queue.opener_work_rows()
                    if row['opener'] == word}

        if args.opener_work_id is not None:
            if args.opener_work_id not in all_rows:
                print(f'{word.upper()}: no opener-work request with id '
                      f'{args.opener_work_id}.')
                return
            opener_work_id = args.opener_work_id
        else:
            open_ids = [row['opener_work_id']
                        for row in queue.opener_work_candidates()
                        if row['opener'] == word]
            if not open_ids:
                if all_rows:
                    print(f'{word.upper()}: all {len(all_rows)} '
                          f'opener-work request(s) are complete.')
                else:
                    print(f'{word.upper()}: no opener-work request found.')
                return
            if len(open_ids) > 1:
                print(f'{word.upper()}: ambiguous, {len(open_ids)} open '
                      f'opener-work requests match.  '
                      f'Use --opener-work-id to disambiguate.')
                for candidate_id in sorted(open_ids):
                    row = all_rows[candidate_id]
                    requested_at = datetime.fromtimestamp(
                        row['requested_at']).strftime('%Y-%m-%d %H:%M')
                    print(f'  id {candidate_id}  '
                          f'priority {row["requested_priority"]}  '
                          f'{row["state"]}  {row["root_count"]} direct, '
                          f'{row["branch_count"]} branch(es)  '
                          f'requested {requested_at}')
                return
            opener_work_id = open_ids[0]

        updated = queue.set_opener_work_priority(opener_work_id, args.priority)
    finally:
        queue.close()

    if updated:
        print(f'{word.upper()} (id {opener_work_id}): '
              f'requested priority set to {args.priority}.')
    else:
        print(f'{word.upper()} (id {opener_work_id}): '
              f'request is complete, cannot reprioritize.')


# ---------------------------------------------------------------------------
# start / stop  (systemd delegation)
# ---------------------------------------------------------------------------

_SYSTEMD_SERVICE = 'wordle-erd'
_REPORT_SERVER_SYSTEMD_SERVICE = 'wordle-report-server'


def _run_systemctl(service: str, action: str, *extra: str) -> int:
    """Run `systemctl --user <action> <service> [extra...]` and return the
    exit code."""
    import subprocess
    result = subprocess.run(
        ['systemctl', '--user', action, service, *extra],
        capture_output=False)
    return result.returncode


def _run_journalctl(service: str, since: float) -> int:
    """Print journal entries for ``service`` written since ``since``."""
    import subprocess
    result = subprocess.run(
        ['journalctl', '--user', '--unit', service, '--since', f'@{since}',
         '--no-pager', '--full'],
        capture_output=False)
    return result.returncode


def _service_scope_noun(args) -> str:
    """The subject describing which services a scoped command acted on."""
    if args.swarm_only:
        return 'Supervisor'
    if args.web_only:
        return 'Report server'
    return 'Supervisor and report server'


def _add_service_scope_flags(parser, verb: str) -> None:
    """Add the mutually exclusive --swarm-only / --web-only scope flags shared
    by start, stop, and restart.  Neither flag means act on both services."""
    scope = parser.add_mutually_exclusive_group()
    scope.add_argument(
        '--swarm-only', action='store_true',
        help=f'{verb} the supervisor only; leave the report web server alone')
    scope.add_argument(
        '--web-only', action='store_true',
        help=f'{verb} the report web server only; leave the supervisor alone')


def _start_or_restart_services(args, action: str) -> None:
    """Shared body for `start` and `restart`: run `systemctl <action>` on the
    supervisor (unless --web-only) and the report server (unless --swarm-only).

    A supervisor failure aborts before touching the report server: a broken
    supervisor is the primary problem, and the report server has nothing new
    to add while it's down.  Once the supervisor action succeeds, the report
    server is attempted independently and its failure is reported without
    undoing the supervisor action -- these are two separately-managed
    services, not a transaction."""
    diagnostics_since = time.time()
    if not args.web_only:
        rc = _run_systemctl(_SYSTEMD_SERVICE, action)
        if rc != 0:
            print(f'systemctl {action} failed (exit {rc}).  '
                  f'Is the service installed?  '
                  f'Check: systemctl --user status {_SYSTEMD_SERVICE}',
                  file=sys.stderr)
            sys.exit(rc)

    server_rc = 0
    if not args.swarm_only:
        server_rc = _run_systemctl(_REPORT_SERVER_SYSTEMD_SERVICE, action)
        if server_rc != 0:
            print(f'systemctl {action} failed (exit {server_rc}).  '
                  f'Is the service installed?  '
                  f'Check: systemctl --user status '
                  f'{_REPORT_SERVER_SYSTEMD_SERVICE}',
                  file=sys.stderr)

    if not args.web_only:
        _run_systemctl(_SYSTEMD_SERVICE, 'status', '--no-pager', '--lines=0')
        _run_journalctl(_SYSTEMD_SERVICE, diagnostics_since)
    if not args.swarm_only:
        _run_systemctl(
            _REPORT_SERVER_SYSTEMD_SERVICE, 'status', '--no-pager',
            '--lines=0')
        _run_journalctl(_REPORT_SERVER_SYSTEMD_SERVICE, diagnostics_since)
    if server_rc != 0:
        sys.exit(server_rc)


def cmd_start(args):
    """Start the supervisor and the report web server via systemd, scoped by
    --swarm-only / --web-only.  Starting a service that is already running is a
    no-op."""
    _start_or_restart_services(args, 'start')


def cmd_stop(args):
    """Stop the supervisor and the report web server via systemd, scoped by
    --swarm-only / --web-only.

    Both stops are attempted even if the first fails: stopping is best-effort
    cleanup, not a pipeline, so a failure on one service must never skip the
    other."""
    rc = 0
    if not args.web_only:
        rc = _run_systemctl(_SYSTEMD_SERVICE, 'stop')
        if rc != 0:
            print(f'systemctl stop failed (exit {rc}).', file=sys.stderr)

    server_rc = 0
    if not args.swarm_only:
        server_rc = _run_systemctl(_REPORT_SERVER_SYSTEMD_SERVICE, 'stop')
        if server_rc != 0:
            print(f'systemctl stop failed for report server (exit '
                  f'{server_rc}).', file=sys.stderr)

    if rc == 0 and server_rc == 0:
        print(f'{_service_scope_noun(args)} stopped.')
    else:
        sys.exit(rc or server_rc)


def cmd_restart(args):
    """Restart the supervisor and the report web server via systemd, scoped by
    --swarm-only / --web-only (stop + start in one step, per service)."""
    _start_or_restart_services(args, 'restart')


# ---------------------------------------------------------------------------
# run (supervisor)
# ---------------------------------------------------------------------------

def _checkpoint_cache_on_start(cache_path):
    """Flush any leftover WAL into the main DB through SQLite's own recovery.

    A worker killed mid-write leaves a -wal holding committed transactions.
    NEVER delete that file: SQLite replays it on the next open, and removing
    it discards committed data and corrupts the main DB.  Instead open the
    DB single-threaded and TRUNCATE-checkpoint, which is the blessed way to
    drain the WAL cleanly before the worker swarm starts hammering it.
    """
    import sqlite3
    try:
        conn = sqlite3.connect(cache_path, timeout=60)
        conn.execute('PRAGMA busy_timeout = 60000')
        conn.execute('PRAGMA wal_checkpoint(TRUNCATE)')
        conn.close()
        print('Startup: WAL checkpointed clean.')
    except sqlite3.Error as e:
        print(f'Startup: WAL checkpoint failed: {e}', file=sys.stderr)


DISK_SAMPLE_SECONDS = 30
# Cadence for the supervisor's check_opener_work_invariants() sweep.  The
# reconciliation in _resolve_branch_memberships already demotes routine
# pruning residue on every membership resolution, so a violation surviving
# to this sweep is a genuine anomaly worth logging rather than the expected
# output of alpha-beta pruning.
INVARIANT_CHECK_SECONDS = 300
QUEUE_WAL_QUIESCE_BYTES = 2 * 1024 ** 3
TRUNCATE_RETRY_SECONDS = 15
# Hard ceiling on the queue WAL (QUEUE_WAL_HARD_CEILING_BYTES, shared with
# the workers' backstop via erd_queue).  In healthy operation the
# quiesce/TRUNCATE protocol keeps the WAL near QUEUE_WAL_QUIESCE_BYTES; a WAL
# an order of magnitude larger means TRUNCATE has been losing for many cycles
# and the file is on course to fill the disk — the failure mode that corrupts
# the database.  At this size the supervisor stops trying to recover in place:
# it captures a diagnostic snapshot, signals every worker to dump its stacks,
# latches the swarm down (a manual `queue clear-disk-stop` is then required),
# and exits.
# Fill/drain rates below this magnitude read as "steady": smaller than this
# and the trend is within statvfs sampling noise (page cache, unrelated
# processes on the same filesystem).  Anything at or above it is shown via
# _fmt_size, whose adaptive K/M/G unit keeps a reportable rate from ever
# rounding to a bare "0".
DISK_RATE_FLOOR_BYTES = 10_000   # 10 kB/s


def _disk_guard(queue, queue_path) -> bool:
    """Stop and latch the swarm when disk use reaches its reserved margin."""
    used_fraction = disk_stats(queue_path)['used_fraction']
    if used_fraction < DISK_STOP_FRACTION:
        return False
    logger.critical('Disk %.1f%% full (>= %.0f%% stop threshold) — stopping '
                    'swarm and latching down.  Clear with: '
                    'erd_search.py queue clear-disk-stop',
                    100 * used_fraction, 100 * DISK_STOP_FRACTION)
    try:
        queue.set_disk_stop(f'supervisor: disk {100 * used_fraction:.1f}% full')
    except sqlite3.OperationalError as exc:
        # The startup fullness check keeps the swarm down without a latch row.
        logger.critical('Could not write disk_stop latch: %s', exc)
    return True


def _supervisor_checkpoint(queue):
    """Backfill the queue WAL without taking the writer lock."""
    queue.checkpoint('PASSIVE')


def _check_opener_work_invariants(queue):
    """Log any check_opener_work_invariants() violations found right now."""
    violations = queue.check_opener_work_invariants()
    if violations:
        logger.warning('Opener-work invariant check found %d violation(s):',
                       len(violations))
        for violation in violations:
            logger.warning('  %s', violation)


def _maybe_quiesce_truncate(queue):
    """Quiesce queue readers and truncate an oversized WAL."""
    wal_bytes = queue.wal_size_bytes()
    if wal_bytes < QUEUE_WAL_QUIESCE_BYTES:
        return
    logger.info('Queue WAL at %.2f GB — quiescing workers for TRUNCATE.',
                wal_bytes / 1e9)
    queue.set_checkpoint_pause(True)
    try:
        deadline = time.time() + TRUNCATE_RETRY_SECONDS
        while True:
            result = queue.checkpoint('TRUNCATE')
            if result is not None and result[0] == 0:
                logger.info('Queue WAL truncated (%.2f GB reclaimed).',
                            wal_bytes / 1e9)
                return
            if time.time() >= deadline:
                logger.warning('Queue WAL TRUNCATE still busy after %ds '
                               '(wal=%.2f GB); will retry next pass.',
                               TRUNCATE_RETRY_SECONDS,
                               queue.wal_size_bytes() / 1e9)
                return
            time.sleep(0.5)
    finally:
        queue.set_checkpoint_pause(False)


def _dump_worker_stacks(procs):
    """SIGUSR1 every live worker so it dumps all-thread stacks to its log.

    Workers register faulthandler on SIGUSR1 (erd_swarm._setup_logging), whose
    C-level handler dumps even from inside a native call (numpy, sqlite).
    Best-effort: a dead pid is skipped, and a worker deep in C returns its dump
    when the call unwinds."""
    for wid, (p, _started_at) in procs.items():
        if not p.is_alive():
            continue
        try:
            os.kill(p.pid, signal.SIGUSR1)
            logger.critical('Requested stack dump from worker-%d (pid=%d).',
                            wid, p.pid)
        except OSError as exc:
            logger.warning('Could not signal worker-%d (pid=%d): %s',
                           wid, p.pid, exc)


def _enforce_wal_hard_ceiling(queue, procs) -> bool:
    """Backstop for a quiesce/TRUNCATE that never wins: when the WAL breaches
    QUEUE_WAL_HARD_CEILING_BYTES, capture why and stop before it fills the disk
    and corrupts the database.  Returns True once it has latched the swarm
    down; the caller then stops."""
    wal_bytes = queue.wal_size_bytes()
    if wal_bytes < QUEUE_WAL_HARD_CEILING_BYTES:
        return False
    logger.critical(
        'Queue WAL %.2f GB breached hard ceiling %.2f GB — TRUNCATE never '
        'reclaimed it. Latching swarm down before the disk fills. Per-table '
        'WAL attribution and worker stacks are in the runtime/erd_worker_*.log files.',
        wal_bytes / 1e9, QUEUE_WAL_HARD_CEILING_BYTES / 1e9)
    now = time.time()
    for h in queue.heartbeats_with_branch():
        started = h['claim_started_at']
        updated = h['updated_at']
        logger.critical(
            '  worker-%s pid=%s cand=%s in_candidate=%ss since_heartbeat=%ss '
            'nodes=%s node_rate=%s/s',
            h['worker_id'], h['pid'], h['cur_candidate'],
            f'{now - started:.0f}' if started else '?',
            f'{now - updated:.0f}' if updated else '?',
            h['cur_nodes'], h['node_rate'])
    _dump_worker_stacks(procs)
    # Let workers flush their dumps to their logs before the supervisor tears
    # them down at the end of the run loop.
    time.sleep(2)
    try:
        queue.set_disk_stop(
            f'supervisor: queue WAL {wal_bytes / 1e9:.1f} GB breached hard '
            f'ceiling ({QUEUE_WAL_HARD_CEILING_BYTES / 1e9:.0f} GB)')
    except sqlite3.OperationalError as exc:
        logger.critical('Could not write disk_stop latch on WAL ceiling: %s',
                        exc)
    return True


def cmd_run(args):
    # First, before anything opens --cache: the two paths might be the same
    # file, and every operation below writes to --cache.  Checkpointing it and
    # running its migrations would rewrite the quarantined artifact — new
    # tables, moved rows, a changed mtime — and only then reach a refusal that
    # can no longer undo any of it.  Proving the artifact here also keeps a
    # refusal from being raised independently by every worker, which the
    # supervisor would respawn forever.
    if not _hint_cache_is_usable(args):
        return
    _checkpoint_cache_on_start(args.cache)
    # Apply any pending ScoreCache schema migrations single-threaded now, before
    # the worker processes open the cache concurrently — concurrent first-open
    # would race on ALTER TABLE ADD COLUMN ("duplicate column name").
    ScoreCache(args.cache, load_word_list(ANSWER_FILE),
               checkpoint_on_close=False).close()
    queue = ERDQueue(args.queue)
    latch = queue.disk_stop()
    if latch is not None:
        at = (
            datetime.fromtimestamp(latch['at']).isoformat(' ')
            if latch.get('at') else 'unknown time'
        )
        print(f'Refusing to start: disk-stop latch is set ({latch["reason"]}, '
              f'at {at}).\nFree disk space, then clear it with: '
              f'erd_search.py queue clear-disk-stop', file=sys.stderr)
        queue.close()
        return
    used_fraction = disk_stats(args.queue)['used_fraction']
    if used_fraction >= DISK_STOP_FRACTION:
        queue.set_disk_stop(f'startup: disk {100 * used_fraction:.1f}% full')
        print(f'Refusing to start: disk {100 * used_fraction:.1f}% full '
              f'(>= {100 * DISK_STOP_FRACTION:.0f}% stop threshold); '
              f'latching down.', file=sys.stderr)
        queue.close()
        return
    stale = queue.reset_stale_in_progress()
    nb, nc = queue.recover_active_branches()
    if stale or nb or nc:
        print(f'Recovery: {stale} pending rows reset, '
              f'{nb} active branches resumed, {nc} in-flight claims freed.')

    counts = queue.counts_by_status()
    if not counts.get('pending') and not counts.get('in_progress'):
        print('Warning: queue appears empty.  '
              'Run queue add to load branches before starting workers.',
              file=sys.stderr)
    # Telemetry epoch the workers will stamp their rows with.  Epoch 0 is the
    # single-candidate-atom baseline; a claiming-regime change calls set_epoch.
    print(f'Telemetry epoch: {queue.epoch}')
    queue.close()

    _setup_supervisor_logging()
    logger.info('Supervisor starting: %d workers, recycle_hours=%.1f',
                args.workers, args.recycle_hours)

    stop_event = multiprocessing.Event()

    def _sighandler(signum, frame):
        logger.info('Supervisor received signal %d — stopping...', signum)
        print(f'\nReceived signal {signum}, draining workers...')
        stop_event.set()

    signal.signal(signal.SIGTERM, _sighandler)
    signal.signal(signal.SIGINT, _sighandler)

    procs: dict[int, tuple] = {}
    for i in range(args.workers):
        procs[i] = _spawn_worker(i, args, stop_event)
    logger.info('Started %d workers (supervisor pid=%d).', args.workers, os.getpid())

    q = ERDQueue(args.queue)
    last_checkpoint = time.time()
    last_disk_sample = 0.0
    last_invariant_check = time.time()
    while not stop_event.is_set():
        time.sleep(5)
        if stop_event.is_set():
            break

        try:
            if _disk_guard(q, args.queue):
                stop_event.set()
                break

            now = time.time()
            if now - last_disk_sample > DISK_SAMPLE_SECONDS:
                q.record_disk_sample(disk_stats(args.queue)['avail_bytes'])
                last_disk_sample = now

            if now - last_checkpoint > erd_swarm.CHECKPOINT_SECONDS:
                _supervisor_checkpoint(q)
                last_checkpoint = time.time()
            if now - last_invariant_check > INVARIANT_CHECK_SECONDS:
                # Purely diagnostic (six read queries, one log line): a
                # transient OperationalError here must not escalate to the
                # fail-stop handler below, which is for the load-bearing
                # writes elsewhere in this loop.
                try:
                    _check_opener_work_invariants(q)
                except sqlite3.OperationalError as exc:
                    logger.warning('Opener-work invariant check skipped: %s',
                                   exc)
                last_invariant_check = time.time()
            _maybe_quiesce_truncate(q)
            if _enforce_wal_hard_ceiling(q, procs):
                stop_event.set()
                break

            for wid, (p, started_at) in list(procs.items()):
                age = time.time() - started_at
                if not p.is_alive():
                    logger.info('Worker %d exited (age=%.0fs), respawning',
                                wid, age)
                    _reap_worker(q, wid)
                    procs[wid] = _spawn_worker(wid, args, stop_event)
                elif age > args.recycle_hours * 3600:
                    logger.info('Worker %d recycle-hours hit (age=%.0fs), '
                                'terminating and respawning', wid, age)
                    p.terminate()
                    p.join(timeout=10)
                    if p.is_alive():
                        logger.warning('Worker %d did not exit on SIGTERM; '
                                       'killing', wid)
                        p.kill()
                        p.join(timeout=10)
                    _reap_worker(q, wid)
                    procs[wid] = _spawn_worker(wid, args, stop_event)

            # Liveness-gated reclaim never frees work held by a live worker.
            freed = q.reclaim_stale_claims(args.worker_timeout_seconds)
            if freed:
                logger.info('Reclaimed %d stale candidate claim(s).', freed)
            counts = q.counts_by_status()
            in_flight = len(q.branches_in_progress())

            if (counts.get('pending', 0) == 0
                    and counts.get('in_progress', 0) == 0
                    and in_flight == 0
                    and counts):
                logger.info('Queue drained — all branches done.')
                print('\nQueue empty — all branches done.')
                stop_event.set()
        except sqlite3.OperationalError as exc:
            logger.critical('Queue database error in supervisor loop: %s — '
                            'stopping swarm.', exc)
            stop_event.set()

    logger.info('Supervisor stopping all workers...')
    for wid, (p, _) in procs.items():
        if p.is_alive():
            p.terminate()
    for wid, (p, _) in procs.items():
        p.join(timeout=30)
        if p.is_alive():
            # A worker that ignores SIGTERM must not outlive the supervisor:
            # every WAL/disk safeguard below assumes no writers remain, and an
            # orphaned writer can fill the disk unsupervised.
            logger.warning('Worker %d did not exit on SIGTERM; killing.', wid)
            p.kill()
            p.join(timeout=10)
    # Workers are gone: the WAL truncates uncontested.
    q.checkpoint()
    q.close()
    logger.info('Supervisor exited.')
    print('All workers stopped.')


def _reap_worker(queue, worker_id: int):
    """Free a dead/killed worker's in-flight candidate claims and clear its
    heartbeat, BEFORE a replacement of the same name starts heartbeating.

    Without this, a respawned worker-N would refresh the 'worker-N' heartbeat,
    making the previous instance's orphaned (done=0) claims look like they're
    held by a live worker — so the liveness-gated reclaim would never free them
    and the affected branches would never finalize.
    """
    name = f'worker-{worker_id}'
    freed = queue.reclaim_claims_of_worker(name)
    queue.clear_heartbeat(name)
    if freed:
        logger.info('Reaped worker %d: freed %d candidate claim(s).',
                    worker_id, freed)


def _hint_cache_is_usable(args) -> bool:
    """Open and close the configured hint artifact, reporting any refusal.

    True when the run may proceed — including the ordinary case of no
    --hint-cache at all.  A refusal is printed and the run stops: a hint path
    the operator named but the swarm cannot honour must not turn into a silent
    hintless run.
    """
    try:
        hint = open_hint_cache(args.hint_cache, load_word_list(ANSWER_FILE),
                               args.cache)
    except HintCacheError as error:
        print(f'Refusing to start: {error}', file=sys.stderr)
        return False
    if hint is not None:
        print(f'Hint cache: {hint.db_path} '
              f'({hint.namespace_branch_count:,} branch rows for this answer '
              f'list, read-only, candidate order only)')
        hint.close()
    return True


def _spawn_worker(worker_id: int, args, stop_event):
    p = multiprocessing.Process(
        target=erd_swarm.swarm_worker,
        args=(worker_id, args.cache, args.queue, stop_event, args.workers),
        kwargs={'hint_cache_path': args.hint_cache},
        daemon=False,
        name=f'erd-worker-{worker_id}',
    )
    p.start()
    logger.info('Spawned worker-%d (pid=%d)', worker_id, p.pid)
    return p, time.time()


def _setup_supervisor_logging():
    log_path = DEFAULT_SEARCH_LOG_PATH
    h = logging.FileHandler(log_path)
    h.setFormatter(logging.Formatter(
        '%(asctime)s %(levelname)-7s %(message)s'))
    logger.addHandler(h)
    logger.setLevel(logging.INFO)
    print(f'Logging to {log_path}')


# ---------------------------------------------------------------------------
# reset-stale
# ---------------------------------------------------------------------------

def cmd_reset_stale(args):
    queue = ERDQueue(args.queue)
    n = queue.reset_stale_in_progress()
    queue.close()
    print(f'Reset {n} in_progress row(s) to pending.')


# ---------------------------------------------------------------------------
# queue reconcile-orphaned-ownership
# ---------------------------------------------------------------------------

def cmd_queue_reconcile_orphaned_ownership(args):
    queue = ERDQueue(args.queue)
    branch_ids = queue.reconcile_orphaned_branch_ownership()
    queue.close()
    if not branch_ids:
        print('No orphaned owned branches found.')
        return
    print(f'Demoted {len(branch_ids)} orphaned owned branch(es) to direct '
          f'(claimable without a live opener-work membership): '
          f'{", ".join(str(b) for b in branch_ids)}')


def _normalize_queue_cli_args(args):
    """Apply the queue-level path to nested mutation commands."""
    if args.cmd != 'queue':
        return
    if not hasattr(args, 'queue'):
        args.queue = args.queue_path or DEFAULT_QUEUE


def _normalize_epoch_cli_args(args):
    """Apply the epoch-level path to nested epoch commands."""
    if args.cmd != 'epoch':
        return
    if not hasattr(args, 'queue'):
        args.queue = args.queue_path or DEFAULT_QUEUE


# ---------------------------------------------------------------------------
# telemetry epoch
# ---------------------------------------------------------------------------

def _current_git_sha():
    """Return the current checkout's abbreviated commit SHA, if available."""
    import subprocess
    try:
        result = subprocess.run(
            ['git', 'rev-parse', '--short', 'HEAD'],
            capture_output=True, check=True, text=True,
        )
    except (OSError, subprocess.CalledProcessError):
        return None
    return result.stdout.strip() or None


def cmd_epoch_show(args):
    queue = ERDQueue(args.queue, initialize_schema=False)
    try:
        print(json.dumps(queue.epoch_metadata(), sort_keys=True))
    finally:
        queue.close()


def cmd_epoch_set(args):
    queue = ERDQueue(args.queue)
    try:
        now = int(time.time())
        live_workers = [
            row for row in queue.heartbeats_with_branch()
            if now - row['updated_at'] <= WORKER_LIVENESS_SECONDS
        ]
        if live_workers and not args.force:
            worker_ids = ', '.join(row['worker_id'] for row in live_workers)
            print(
                f'Refusing to change telemetry epoch while live workers are '
                f'heartbeating: {worker_ids}. Stop the swarm first, or use '
                f'--force to override.',
                file=sys.stderr,
            )
            return 1
        git_sha = args.git_sha if args.git_sha is not None else _current_git_sha()
        queue.set_epoch(
            args.epoch, label=args.label, git_sha=git_sha, notes=args.notes,
        )
        print(json.dumps(queue.epoch_metadata(), sort_keys=True))
        return 0
    finally:
        queue.close()


# ---------------------------------------------------------------------------
# CLI entry point
# ---------------------------------------------------------------------------

def main():
    parser = argparse.ArgumentParser(
        description=__doc__,
        formatter_class=argparse.RawDescriptionHelpFormatter)
    sub = parser.add_subparsers(dest='cmd', required=True)

    # -- run --
    p_run = sub.add_parser('run', help='Start the parallel precache supervisor')
    p_run.add_argument('--workers', type=int, default=6, metavar='N',
                       help='Number of swarm worker processes (default: 6)')
    p_run.add_argument('--cache', default=DEFAULT_CACHE, metavar='PATH')
    p_run.add_argument(
        '--hint-cache', default=None, metavar='PATH',
        help='quarantined historical cache, opened read-only and immutable '
             'and used only to order candidates.  Never a source of exact '
             'results, bounds, or losses, and never written to.')
    p_run.add_argument('--queue', default=DEFAULT_QUEUE, metavar='PATH')
    p_run.add_argument('--recycle-hours', type=float, default=3.0,
                       metavar='H',
                       help='Respawn each worker after H hours wall time '
                            '(default: 3).  Bounds per-worker ScoreCache '
                            'memory growth while in-progress work is preserved '
                            'in the queue and resumed by the fresh worker.')
    p_run.add_argument('--worker-timeout-seconds', type=int,
                       default=WORKER_LIVENESS_SECONDS,
                       metavar='S',
                       help='Declare a worker dead and reclaim its candidate '
                            'claims after S seconds of missed heartbeats '
                            '(default: 30).  Live workers heartbeat every '
                            '~2s regardless of how long a single candidate '
                            'takes, so only a crashed process triggers this.')

    # -- view --
    p_view = sub.add_parser(
        'view', help='View shared swarm reports',
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""Branch status
  A branch moves through one lifecycle, normally in order:

    unqueued -> queued -> evaluating -> finalizing -> done

    unqueued    discovered, but not scheduled or cached
    queued      scheduled and waiting for a worker
    evaluating  candidates are being evaluated
    finalizing  a worker is persisting the completed evaluation
    done        a reusable result has been persisted

  Worker presence only distinguishes evaluating and finalizing branches:
    active      a worker is on the branch right now
    waiting     no worker is on the branch right now

  Cached or trivial branches can move directly to done.  Recovery may
  briefly return an interrupted finalization to evaluating before retrying
  it.  Removing unfinished work returns the branch to unqueued.

  A word report derives its response groups from the answer list, so it is
  the only report that can show an unqueued branch; every other report reads
  its branches from the queue.  --branch-status unqueued is refused
  elsewhere rather than matching nothing.

  The overview answers what the swarm is doing now, so it defaults to
  --branch-status evaluating,finalizing --branch-worker-status active.
  Every other report starts unfiltered.  Use comma-separated values such as
  --branch-status queued,done, or use all to disable a filter.

  Worker status narrows only the stages that carry one, so a status filter
  selecting neither evaluating nor finalizing drops it rather than matching
  nothing.
""")
    p_view.add_argument('--watch', nargs='?', const=30.0,
                        type=_view_watch_interval, metavar='SECONDS')
    p_view.add_argument('--format', choices=('text', 'json', 'jsonl'),
                        default='text')
    p_view.add_argument('--no-color', action='store_true')
    p_view.add_argument('--queue-path', default=DEFAULT_QUEUE, metavar='PATH')
    p_view.add_argument('--cache-path', default=DEFAULT_CACHE, metavar='PATH')
    p_view.add_argument('--claims', action='store_true')
    p_view.add_argument('--answers', action='store_true')
    p_view.add_argument('--tree', action='store_true')
    view_kind = p_view.add_mutually_exclusive_group()
    view_kind.add_argument('--queue', dest='view_queue', action='store_true')
    view_kind.add_argument('--workers', action='store_true')
    view_kind.add_argument('--worker', metavar='N')
    view_kind.add_argument('--cache', dest='view_cache', action='store_true')
    view_kind.add_argument('--hotspots', action='store_true')
    view_kind.add_argument('--accuracy', action='store_true',
                           help='Show predicted-versus-actual candidate work')
    view_kind.add_argument(
        '--work-distribution', dest='work_distribution', action='store_true',
        help='Band branches by worker time to show where search work and '
             'coordination actually go')
    view_kind.add_argument('--leaderboard', action='store_true')
    view_kind.add_argument(
        '--openers', action='store_true',
        help='Show opener-work requests and branch ownership/lineage')
    view_kind.add_argument(
        '--root-progress', dest='root_progress', action='store_true',
        help='Show per-response-group work totals and a completion estimate '
             'for one opener')
    p_view.add_argument(
        '--inherited-cost', dest='inherited_cost', action='store_true',
        help='With --root-progress, roll up what other openers spent on the '
             'response groups this one inherited')
    p_view.add_argument(
        '--branch-status', type=_branch_status_filter, metavar='STATUSES',
        help='Comma-separated unqueued,queued,evaluating,finalizing,done, or '
             'all (unqueued selects only on a word report)')
    p_view.add_argument(
        '--branch-worker-status', type=_branch_worker_status_filter, metavar='STATUSES',
        help='Comma-separated active,waiting, or all (meaningful only for '
             'evaluating/finalizing branches)')
    p_view.add_argument(
        '--opener-state', type=_opener_state_filter, metavar='STATES',
        help='Comma-separated queued,active,complete, or all (--openers only)')
    p_view.add_argument('--minimum-answer-count', type=int, metavar='N')
    p_view.add_argument('--maximum-answer-count', type=int, metavar='N')
    p_view.add_argument('--budget', type=int, metavar='N')
    p_view.add_argument('--priority', type=int, metavar='N')
    p_view.add_argument('--sort',
                        choices=tuple(sorted(
                            {'default', 'age', 'size', 'workers', 'priority',
                             'nodes', 'slowest', *OPENER_SORT_FIELDS,
                             *ROOT_PROGRESS_SORT_FIELDS})))
    p_view.add_argument('--limit', type=int, metavar='N')
    p_view.add_argument(
        '--by', choices=(
            'nodes', 'age', 'size', 'workers', 'priority', 'slowest',
            'evaluated-candidates', 'one-level-erd-prunes',
            'two-level-erd-prunes',
            'cut-reuse', 'coordination'))
    p_view.add_argument('--epoch', type=int, metavar='N')
    p_view.add_argument('--opener', metavar='WORD',
                        help='Restrict --accuracy to one opener')
    p_view.add_argument('--accuracy-offset', type=int, default=0, metavar='N',
                        help='Skip N raw --accuracy rows before --limit')
    p_view.add_argument('--since-seconds', type=int, metavar='N')
    p_view.add_argument('--sample-size', type=int, metavar='N')
    p_view.add_argument('spine', nargs='*', metavar='SPINE')

    # -- queue --
    p_queue = sub.add_parser('queue', help='Manage queue work')
    p_queue.add_argument('--queue', dest='queue_path', default=None, metavar='PATH')
    qsub = p_queue.add_subparsers(dest='queue_cmd', required=True)

    # -- queue add --
    p_qa = qsub.add_parser('add',
                           help='Add branches for one or more words to the queue')
    qa_word = p_qa.add_mutually_exclusive_group(required=True)
    qa_word.add_argument('--word', nargs='+', metavar='WORD',
                         help='One or more guess words, space-separated '
                              '(e.g. --word salet crane raise)')
    qa_word.add_argument('--words-file', metavar='FILE',
                         help=f'File of words to add, one per line '
                              f'(default list: {WORDS_FILE})')
    p_qa.add_argument('--pattern', metavar='PAT',
                      help='Only add this specific response pattern for --word '
                           '(5 chars: g=green y=yellow -=gray).  '
                           'Omit to add all patterns for the word(s).')
    p_qa.add_argument('--priority', type=int, default=None, metavar='N',
                      help='Priority of the last word on the ladder.  Higher '
                           'numbers are worked sooner.  Omit to append the '
                           'batch below every word the queue still owes work, '
                           'so it preempts nothing already queued; name a '
                           'value to preempt deliberately.')
    p_qa.add_argument('--priority-step', type=int,
                      default=DEFAULT_PRIORITY_STEP, metavar='N',
                      help=f'Gap between adjacent words on the priority '
                           f'ladder (default: {DEFAULT_PRIORITY_STEP}), so the '
                           f'swarm finishes one word before starting the next.  '
                           f'0 queues every word at --priority, letting them '
                           f'all start at once.')
    p_qa.add_argument('--priority-words', nargs='+', metavar='WORD',
                      help='With --words-file: only these words are laddered '
                           'from --priority, the rest queued flat at 0 (e.g. '
                           '--priority-words salet crane)')
    p_qa.add_argument('--max-branch-size', type=int, default=None, metavar='N',
                      help='Skip branches with more than N answer words '
                           '(default: unlimited)')
    p_qa.add_argument('--delete-erd-cache', action='store_true',
                      help='Delete any existing ERD cache entry for each '
                           'queued branch first, so it is recomputed instead '
                           'of being skipped as already-cached.  Also discards '
                           'the queue-side work state of branches already '
                           'completed — their candidate claims, bundle stats '
                           'and active-branch row — so they are claimable '
                           'again.  Branches a worker is still solving are '
                           'left alone.')
    p_qa.add_argument('--cache', default=DEFAULT_CACHE, metavar='PATH')
    p_qa.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    # -- queue clear --
    p_qc = qsub.add_parser('clear',
                            help='Wipe all queue state (does not touch the ERD cache)')
    p_qc.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')
    p_qc.add_argument('--yes', action='store_true',
                      help='Skip confirmation prompt')

    # -- queue remove --
    p_qr = qsub.add_parser('remove',
                            help='Remove a pending branch from the queue')
    p_qr.add_argument('--word', required=True, metavar='WORD')
    p_qr.add_argument('--pattern', required=True, metavar='PAT',
                      help='Response pattern (5 chars: g=green y=yellow -=gray)')
    p_qr.add_argument('--force', action='store_true',
                      help='Also cancel an in-progress branch (clears active '
                           'state so the worker moves on after its current candidate)')
    p_qr.add_argument('--cache', default=DEFAULT_CACHE, metavar='PATH')
    p_qr.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    # -- queue priority --
    p_qp = qsub.add_parser('priority',
                            help='Set the priority of a queued branch')
    qp_target = p_qp.add_mutually_exclusive_group(required=True)
    qp_target.add_argument('--word', metavar='WORD')
    qp_target.add_argument('--opener-word', metavar='WORD',
                           help='Set every ownerless open branch attributed '
                                'to this word')
    p_qp.add_argument('--pattern', metavar='PAT',
                      help='Response pattern (5 chars: g=green y=yellow -=gray)')
    p_qp.add_argument('--priority', required=True, type=int, metavar='N',
                      help='New priority (higher = worked sooner; '
                           'use values 0–999,999 for normal work)')
    p_qp.add_argument('--cache', default=DEFAULT_CACHE, metavar='PATH')
    p_qp.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    # -- queue opener-priority --
    p_qsp = qsub.add_parser(
        'opener-priority',
        help='Set the requested priority of an opener-work request')
    p_qsp.add_argument('--word', required=True, metavar='WORD')
    p_qsp.add_argument('--priority', required=True, type=int, metavar='N',
                       help='New requested priority (higher = worked sooner; '
                            'use values 0–999,999)')
    p_qsp.add_argument('--opener-work-id', type=int, default=None, metavar='N',
                       help='opener_work_id to disambiguate, when --word '
                            'owns more than one open request')
    p_qsp.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    # -- start --
    p_start = sub.add_parser(
        'start',
        help='Start the supervisor and the report web server via systemd '
             '(systemctl --user start wordle-erd wordle-report-server)')
    _add_service_scope_flags(p_start, 'Start')

    # -- stop --
    p_stop = sub.add_parser(
        'stop',
        help='Stop the supervisor and the report web server via systemd '
             '(systemctl --user stop wordle-erd wordle-report-server)')
    _add_service_scope_flags(p_stop, 'Stop')

    # -- restart --
    p_restart = sub.add_parser(
        'restart',
        help='Restart the supervisor and the report web server via systemd '
             '(systemctl --user restart wordle-erd wordle-report-server)')
    _add_service_scope_flags(p_restart, 'Restart')

    # -- queue reset-stale --
    p_rst = qsub.add_parser('reset-stale',
                             help='Reset in_progress rows to pending')
    p_rst.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    # -- queue reconcile-orphaned-ownership --
    p_qro = qsub.add_parser(
        'reconcile-orphaned-ownership',
        help='Demote open owned branches whose opener-work membership was '
             'lost while they were still open, making them claimable again')
    p_qro.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    p_cds = qsub.add_parser(
        'clear-disk-stop', help='Release the disk-stop latch so run can start'
    )
    p_cds.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    p_sds = qsub.add_parser(
        'set-disk-stop',
        help='Keep the swarm down across reboots and systemd restarts',
    )
    p_sds.add_argument('--reason', required=True, metavar='TEXT',
                       help='Reason shown when run refuses to start')
    p_sds.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    # -- epoch --
    p_epoch = sub.add_parser(
        'epoch', help='Show or change the telemetry epoch'
    )
    p_epoch.add_argument('--queue', dest='queue_path', default=None, metavar='PATH')
    esub = p_epoch.add_subparsers(dest='epoch_cmd', required=True)

    p_epoch_show = esub.add_parser(
        'show', help='Print the active telemetry epoch metadata'
    )
    p_epoch_show.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    p_epoch_set = esub.add_parser(
        'set', help='Change the active telemetry epoch while workers are stopped'
    )
    p_epoch_set.add_argument('epoch', type=int, metavar='N')
    p_epoch_set.add_argument('--label', metavar='TEXT')
    p_epoch_set.add_argument('--git-sha', metavar='SHA')
    p_epoch_set.add_argument('--notes', metavar='TEXT')
    p_epoch_set.add_argument(
        '--force', action='store_true',
        help='Allow the change despite live worker heartbeats',
    )
    p_epoch_set.add_argument('--queue', default=argparse.SUPPRESS, metavar='PATH')

    args = parser.parse_args()
    ensure_runtime_dir()
    if args.cmd == 'view' and args.format == 'json' and args.watch is not None:
        parser.error('--format json cannot be used with --watch; use jsonl')
    if (args.cmd == 'queue' and args.queue_cmd == 'priority'
            and args.word is not None and args.pattern is None):
        parser.error('--pattern is required with --word')
    if (args.cmd == 'queue' and args.queue_cmd == 'priority'
            and args.opener_word is not None and args.pattern is not None):
        parser.error('--pattern can only be used with --word')
    if args.cmd == 'view':
        try:
            args.branch_target = parse_report_branch_target(args.spine)
        except ValueError as error:
            parser.error(str(error))
        if args.limit is not None and args.limit < 1:
            parser.error('--limit must be at least 1')
        if (args.minimum_answer_count is not None
                and args.maximum_answer_count is not None
                and args.minimum_answer_count > args.maximum_answer_count):
            parser.error('--minimum-answer-count cannot exceed --maximum-answer-count')
        args.report_kind = (
            'queue' if args.view_queue else
            'workers' if args.workers or args.worker is not None else
            'cache' if args.view_cache else
            'hotspots' if args.hotspots else
            'work_distribution' if args.work_distribution else
            'accuracy' if args.accuracy else
            'leaderboard' if args.leaderboard else
            'openers' if args.openers else
            'root_progress' if args.root_progress else 'auto'
        )
        if args.by is not None and not args.hotspots:
            parser.error('--by requires --hotspots')
        if args.sample_size is not None and not (args.hotspots or args.accuracy):
            parser.error('--sample-size requires --hotspots or --accuracy')
        if args.since_seconds is not None and not (
                args.hotspots or args.accuracy or args.work_distribution):
            parser.error('--since-seconds requires --hotspots, --accuracy, or '
                         '--work-distribution')
        if args.epoch is not None and not (args.hotspots or args.root_progress
                                           or args.accuracy
                                           or args.work_distribution):
            parser.error('--epoch requires --hotspots, --accuracy, '
                         '--root-progress, or --work-distribution')
        if args.opener is not None and not args.accuracy:
            parser.error('--opener requires --accuracy')
        if args.accuracy_offset and not args.accuracy:
            parser.error('--accuracy-offset requires --accuracy')
        if args.accuracy_offset < 0:
            parser.error('--accuracy-offset cannot be negative')
        if args.since_seconds is not None and args.since_seconds < 1:
            parser.error('--since-seconds must be at least 1')
        if args.sample_size is not None and args.sample_size < 1:
            parser.error('--sample-size must be at least 1')
        hotspot_field = args.by or 'nodes'
        if args.hotspots and args.limit is None:
            args.limit = 10
        if args.accuracy and args.limit is None:
            args.limit = 20
        overview = is_overview_request(
            args.report_kind, args.branch_target.kind, args.tree
        )
        branch_statuses = args.branch_status
        if branch_statuses is None:
            branch_statuses = OVERVIEW_BRANCH_STATUSES if overview else ()
        branch_worker_statuses = args.branch_worker_status
        if branch_worker_statuses is None:
            branch_worker_statuses = (
                OVERVIEW_BRANCH_WORKER_STATUSES if overview else ()
            )
        args.filters = applied_branch_filters(ReportFilters(
            branch_statuses=branch_statuses,
            branch_worker_statuses=branch_worker_statuses,
            minimum_answer_count=args.minimum_answer_count,
            maximum_answer_count=args.maximum_answer_count,
            budget=args.budget,
            priority=args.priority,
            sort=args.sort,
            opener_states=args.opener_state or (),
            limit=args.limit,
        ))
        args.hotspot_field = hotspot_field if args.hotspots else None
        # The work distribution aggregates rather than samples, so it carries
        # no sample bound at all and defaults to the whole epoch.
        args.sample_size = (
            None if args.work_distribution
            else min(args.sample_size or 50_000, 1_000_000))
        if args.hotspots:
            args.since_seconds = args.since_seconds or 3600
        try:
            validate_report_request(ReportRequest(
                report_kind=args.report_kind,
                branch_target=args.branch_target,
                include_claims=args.claims,
                include_answers=args.answers,
                raw_row_offset=args.accuracy_offset,
                tree=args.tree,
                filters=args.filters,
                worker_id=args.worker,
                hotspot_field=args.hotspot_field,
                epoch=args.epoch,
                since_seconds=args.since_seconds,
                sample_size=args.sample_size,
                inherited_cost=args.inherited_cost,
            ))
        except ValueError as error:
            parser.error(str(error))
    _normalize_queue_cli_args(args)
    _normalize_epoch_cli_args(args)

    if args.cmd == 'queue':
        qdispatch = {
            'add': cmd_queue_add,
            'clear': cmd_queue_clear,
            'remove': cmd_queue_remove,
            'priority': cmd_queue_priority,
            'opener-priority': cmd_queue_opener_priority,
            'reset-stale': cmd_reset_stale,
            'reconcile-orphaned-ownership': cmd_queue_reconcile_orphaned_ownership,
            'clear-disk-stop': cmd_queue_clear_disk_stop,
            'set-disk-stop': cmd_queue_set_disk_stop,
        }
        if args.queue_cmd == 'add':
            try:
                qdispatch[args.queue_cmd](args)
            except ValueError as error:
                parser.error(str(error))
        else:
            qdispatch[args.queue_cmd](args)
        return

    if args.cmd == 'epoch':
        epoch_dispatch = {
            'show': cmd_epoch_show,
            'set': cmd_epoch_set,
        }
        exit_code = epoch_dispatch[args.epoch_cmd](args)
        if isinstance(exit_code, int) and exit_code != 0:
            sys.exit(exit_code)
        return

    dispatch = {
        'start': cmd_start,
        'stop': cmd_stop,
        'restart': cmd_restart,
        'run': cmd_run,
        'view': cmd_view,
    }
    dispatch[args.cmd](args)


if __name__ == '__main__':
    main()