Python Constraints

A constraint provider receives a ConstraintFactory and returns a list of named rules. Each rule starts from an entity stream, filters or joins rows, assigns a penalty or reward, and ends with .named(...).

@constraint_provider
def constraints(factory: ConstraintFactory):
    return [
        factory.for_each(Shift)
        .filter(lambda shift: shift.required and shift.nurse is None)
        .penalize(HardSoftScore.ONE_HARD)
        .named("required shift is unassigned")
    ]

Use small callback functions or lambdas for the business rule itself. Keep data loading, validation, and reporting outside the constraint provider.

Supported Stream Shapes

The supported terminal operations are penalize(...), reward(...), and named(...).

Unary Constraints

Use unary constraints for rules on one entity at a time.

(
    factory.for_each(Shift)
    .filter(lambda shift: shift.employee_idx is None)
    .penalize(HardSoftScore.ONE_HARD)
    .named("unassigned shift")
)

The filter callback must return bool.

Joins

Use stream-level join(...) for pair constraints. joiner.equal(...) compares one key on both sides. joiner.equal_bi(...) compares separate left and right keys. A key may be a Python callback or an attribute-name string.

from solverforge import joiner

(
    factory.for_each(Shift)
    .filter(lambda shift: shift.employee_idx is not None)
    .join(
        Employee,
        joiner.equal_bi(lambda shift: shift.employee_idx, lambda employee: employee.index),
    )
    .filter(lambda shift, employee: shift.required_skill not in employee.skills)
    .penalize(HardSoftScore.ONE_HARD)
    .named("missing required skill")
)

You may also join against a filtered right-hand stream:

active_employees = factory.for_each(Employee).filter(lambda employee: employee.active)

(
    factory.for_each(Shift)
    .join(
        active_employees,
        joiner.equal_bi(lambda shift: shift.employee_idx, lambda employee: employee.index),
    )
)

Joiners preserve Python equality semantics.

Use attribute strings when the equality key is a stable imported row field:

(
    factory.for_each(Shift)
    .join(Employee, joiner.equal_bi("employee_idx", "index"))
    .filter(lambda shift, employee: shift.required_skill not in employee.skills)
    .penalize(HardSoftScore.ONE_HARD)
    .named("missing required skill")
)

SolverForge specializes string equality natively only for planning scalar slots and stable fields stored directly on every imported row. Properties, shadow-derived values, containers, unsupported scalar values, and callback keys retain live Python attribute access and equality.

Grouped Counts

group_by(...) groups rows by a Python key callback. The filter on the grouped stream receives the grouped key and the count for that key.

(
    factory.for_each(Shift)
    .filter(lambda shift: shift.employee_idx is not None)
    .group_by(lambda shift: shift.employee_idx)
    .filter(lambda employee_idx, shift_count: shift_count > 5)
    .penalize(HardSoftScore.ONE_SOFT)
    .named("too many shifts per employee")
)

Use grouped counts for simple capacity, frequency, or load rules.

indexed_presence(...) is available as a grouped collector when the grouped stream needs run or range presence scoring by ordinal index:

from solverforge import indexed_presence

(
    factory.for_each(Shift)
    .group_by(lambda shift: shift.employee_idx, indexed_presence(lambda shift: shift.day))
    .penalize(
        lambda employee_idx, presence: HardSoftScore.of_soft(
            presence.complement_runs(0, 7).len()
            + (1 if presence.any_in(5, 7) else 0)
        )
    )
    .named("missing coverage days")
)

Balance

balance(...) scores imbalance across keys returned by the callback.

(
    factory.for_each(Shift)
    .filter(lambda shift: shift.employee_idx is not None)
    .balance(lambda shift: shift.employee_idx)
    .penalize(HardSoftScore.ONE_SOFT)
    .named("balance employee assignments")
)

Balance scoring is useful for spreading assignments across employees, machines, vehicles, or other owner keys.

List Constraints

Use for_each_unassigned_element(...) when a planning-list model must score elements that are not currently assigned to any owner list.

(
    factory.for_each_unassigned_element(Vehicle, "delivery_order")
    .penalize(HardSoftScore.ONE_HARD)
    .named("all deliveries assigned")
)

Use list_precedence_makespan(...) for list variables that declare element_owner, precedence_duration, and precedence_successors metadata. Each source can be a callback or a solution-level sequence field indexed by element ID. The native scorer computes makespan from that declared metadata and keeps ordinary Python constraints available for additional hard or soft penalties.

factory.list_precedence_makespan(Machine, "operations").named("job shop makespan")

This is a dedicated terminal builder: finish it with .named(...), rather than adding stream filters or .penalize(...). With HardSoftScore, the hard score penalizes missing, duplicate, invalid, or wrong-owner assignments and invalid fixed successor IDs. The soft score is negative makespan over the combined fixed-successor and owner-list adjacency graph. A cycle adds a hard penalty for the whole element set and contributes zero makespan; a zero soft score therefore does not establish feasibility.

List edits compare old and new adjacency edges, preserving unchanged edges and refreshing affected downstream timings. The paired retract/insert update carries its precedence score delta on insertion; an undone trial restores the original score. This scorer composes with ordinary unary, joined, grouped, balance, and unassigned-list rules in the same provider.

Weights

Weights may be score objects, integers, score-level sequences, or callbacks.

(
    factory.for_each(Shift)
    .filter(lambda shift: shift.is_weekend)
    .penalize(HardSoftScore.of_soft(10))
    .named("weekend assignment")
)

For callback weights, return a score object or integer compatible with the solution score family:

(
    factory.for_each(Shift)
    .filter(lambda shift: shift.employee_idx is None)
    .penalize(lambda shift: HardSoftScore.of_hard(shift.priority))
    .named("priority weighted unassigned shift")
)

Compiled Evaluation Boundary

Python callbacks remain the constraint authoring surface. The binding compiles a native evaluator when schema evidence proves it has identical semantics. Safe specializations include:

• simple fixed-weight unary constraints
• unassigned-list scoring
• list precedence/makespan metadata
• proven stable string-key equality joins

Dispatch is per constraint plan, not all-or-nothing for the provider. A callback rule in a mixed constraint set does not force native unary, eligible equality join, or precedence plans through the generic Python-row evaluator. Only plans that need Python rows materialize them. List precedence has its own compiled plan variant for full evaluation and incremental updates; it does not decode generic stream filters or weights. Metadata callbacks can still run inside that specialized scorer, so native dispatch does not mean callback-free execution.

Incremental precedence state is separate for each rule, even when multiple precedence rules target the same owner type. The scorer retains each plan’s metadata access and graph state across route edits and clears that state on reset. Mixed grouped, unassigned-list, and precedence rules keep their own caches and contribute once each to the total score.

Callback filters, callback-computed weights, computed properties, unsupported key values, and stateful callables remain on the Python callback path. Closures, bound defaults, partials, methods, callable instances, and other stateful callbacks compile per invocation rather than being retained across solves.

Only capture-free functions with canonical module provenance and no defaults or function-owned metadata can share a schema/runtime plan. Mutable values in that module namespace remain live, and callbacks from different module namespaces do not share a plan merely because their code objects match.

Python Stream API

Use stream-level for_each(...).join(...) and for_each(...).group_by(...) for joins and grouped counts. These top-level factory methods are not Python APIs yet:

• ConstraintFactory.join(...)
• ConstraintFactory.group_by(...)
• ConstraintFactory.if_exists(...)
• ConstraintFactory.if_not_exists(...)
• ConstraintFactory.flattened(...)