xai-org/x-algorithm · error · ValueError

All weights must be positive

Error message

All weights must be positive

What it means

PriorityTaskGenerator uses integer weights for weighted round-robin polling, so every weight must be a positive integer (>0). Any zero or negative weight would break the polling arithmetic, so the constructor validates them all up front.

Source

Thrown at grox/core/generators/task_generator.py:75

    def _poll(self) -> AsyncGenerator[TaskPayload | None, None]:
        pass

    async def ack(self, result: TaskResult):
        pass

    def identify_task_origin(self, result: TaskResult) -> str | None:
        return self.TASK_GENERATOR_TYPE

    def on_terminal_failure(self, result: TaskResult) -> None:
        pass


class PriorityTaskGenerator(TaskGenerator):
    def __init__(self, generators: list[tuple[TaskGenerator, int]]):
        if not generators:
            raise ValueError("No generators provided")
        if any(weight <= 0 for _, weight in generators):
            raise ValueError("All weights must be positive")
        super().__init__(None)
        self._generators: dict[str, TaskGenerator] = {}
        self._weights: dict[str, int] = {}
        for i, (gen, weight) in enumerate(generators):
            label = f"GEN_{i}"
            self._generators[label] = gen
            self._weights[label] = weight
        self._result_cache: dict[str, str] = {}
        logger.info(
            f"Initialized priority task generator with {list(zip(self._generators.keys(), [gen.__class__.__name__ for gen in self._generators.values()], self._weights.values(), strict=True))}"
        )

    async def start(self) -> None:
        logger.info("Starting priority task generators")
        await asyncio.gather(*[gen.start() for gen in self._generators.values()])
        self._streams = {label: gen.poll() for label, gen in self._generators.items()}
        logger.info("Priority task generators started")

View on GitHub (pinned to 24c60942c5)

Solutions

  1. Remove zero/negative-weight entries from the list before constructing (to disable a generator, don't include it).
  2. Clamp or default computed weights to at least 1.
  3. Validate config weights at load time with a clear error.

Example fix

# before
pg = PriorityTaskGenerator([(gen_a, 0), (gen_b, 3)])

# after
pg = PriorityTaskGenerator([(gen_b, 3)])  # omit disabled generators
Defensive patterns

Strategy: validation

Validate before calling

bad = [(g, w) for g, w in generators if w <= 0]
assert not bad, f"non-positive weights: {bad}"
pg = PriorityTaskGenerator(generators)

Type guard

def all_weights_positive(generators: list[tuple]) -> bool:
    return all(w > 0 for _, w in generators)

Try / catch

try:
    pg = PriorityTaskGenerator(gens)
except ValueError as e:
    if "weights must be positive" in str(e):
        gens = [(g, max(w, 1)) for g, w in gens]
        pg = PriorityTaskGenerator(gens)
    else:
        raise

Prevention

When it happens

Trigger: Passing a (generator, weight) tuple with weight 0 (e.g. to 'disable' a generator), a negative number, or a computed weight that evaluates to <=0.

Common situations: Config weights parsed as 0 for disabled entries; arithmetic that computes weights and underflows to 0/negative; misunderstanding 0 as 'lowest priority' instead of 'invalid'.

Understand the failure class

Background: "Must be a positive integer", "Invalid value", "Unsupported": the invalid-argument-value error family, when a library rejects the value you pass — this error's family across 35 libraries.

Related errors

AI-assisted analysis of xai-org/x-algorithm@24c60942c5 (2026-08-28). Data as JSON: /api/errors/3489ad87a139c9df. Report an issue: GitHub.