cpsat_solver.py

"""Third exact backend: Google OR-Tools CP-SAT via AddCircuit.

Same objective and semantics as every other backend (Section 3.3 of
the technical documentation), but modeled WITHOUT big-M rows: CP-SAT
states each conditional constraint as a reified implication ("if this
arc is driven then T_j >= T_i + service + travel"), and per-adjuster
tours as AddCircuit constraints over eligibility-pruned arc sets. The
logic-heavy constructs that are the compact MILP's bottleneck (Section
3.5) - and that exposed two wrong Gurobi certificates (seeds 3003 and
404) - are native here.

Like the MILP backend it is an anytime exact solver: at the time limit
it returns the best incumbent plus a bound; given enough time it
returns a proof. Unlike Gurobi it has no license size limits (Apache
2.0, ships inside ortools), so it delivers proofs on the Hugging Face
Space where HiGHS runs out of steam - the seed-505 referee instance
(Gurobi ~90 s, HiGHS: no proof) is proven in ~3.5 s. All data is
integer minutes, which is CP-SAT's native currency.

Both optional toggles are modeled: the lunch break as the same
pre/post/on-arc placement disjunction the MILP uses (reified, no
big-M), workload balance as the linear beta * span term.
"""
from ortools.sat.python import cp_model

import config
from data_gen import Adjuster, Claim, arrival_ranges
from solver import Route, Solution, Stop

W = config.MAX_WAIT_MINUTES
LUNCH = config.LUNCH_BREAK
BETA = config.BALANCE_COEFFICIENT


def solve(adjusters: list[Adjuster], claims: list[Claim],
          miles: list[list[float]], travel_min: list[list[int]],
          time_limit_s: int = 60, lunch_break: bool = False,
          balance: bool = False, workers: int = 8):
    """Returns (Solution | None, info dict). info carries the CP-SAT
    status name, wall time, proven flag, and the final lower bound."""
    n_adj = len(adjusters)
    m = cp_model.CpModel()

    def node(i):          # claim index -> travel-matrix node
        return n_adj + i

    elig = {k: [i for i, c in enumerate(claims)
                if config.is_eligible(a, c, miles[k][node(i)])]
            for k, a in enumerate(adjusters)}
    unservable = [c for i, c in enumerate(claims)
                  if not any(i in e for e in elig.values())]

    # Service-start time per claim: inside the window AND finishing by
    # its end - identical to the OR-Tools model's tightened ranges.
    T = {}
    for i, c in enumerate(claims):
        dom = cp_model.Domain.from_intervals(
            [list(r) for r in arrival_ranges(c)])
        T[i] = m.new_int_var_from_domain(dom, f"T_{i}")

    drop = {i: m.new_bool_var(f"drop_{i}") for i in range(len(claims))}
    visit, arcs_by_k, L, E, used = {}, {}, {}, {}, {}
    for k, a in enumerate(adjusters):
        L[k] = m.new_int_var(a.shift_start, a.shift_end, f"L_{k}")
        E[k] = m.new_int_var(a.shift_start, a.shift_end, f"E_{k}")
        m.add(E[k] >= L[k])
        arcs, lits = [], {}
        unused = m.new_bool_var(f"unused_{k}")
        used[k] = unused.Not()
        arcs.append((0, 0, unused))
        for li, i in enumerate(elig[k], start=1):
            skip = m.new_bool_var(f"skip_{k}_{i}")
            arcs.append((li, li, skip))
            visit[k, i] = skip.Not()
            m.add_implication(unused, skip)
        for li, i in enumerate(elig[k], start=1):
            t_hi = travel_min[k][node(i)]
            lit = m.new_bool_var(f"x_{k}_h_{i}")
            arcs.append((0, li, lit))
            lits["h", i] = lit
            m.add(T[i] >= L[k] + t_hi).only_enforce_if(lit)
            m.add(T[i] <= L[k] + t_hi + W).only_enforce_if(lit)
            t_ih = travel_min[node(i)][k]
            lit2 = m.new_bool_var(f"x_{k}_{i}_h")
            arcs.append((li, 0, lit2))
            lits[i, "h"] = lit2
            m.add(E[k] >= T[i] + claims[i].service_minutes + t_ih
                  ).only_enforce_if(lit2)
            for lj, j in enumerate(elig[k], start=1):
                if i == j:
                    continue
                t_ij = travel_min[node(i)][node(j)]
                lit3 = m.new_bool_var(f"x_{k}_{i}_{j}")
                arcs.append((li, lj, lit3))
                lits[i, j] = lit3
                dep = T[i] + claims[i].service_minutes
                m.add(T[j] >= dep + t_ij).only_enforce_if(lit3)
                m.add(T[j] <= dep + t_ij + W).only_enforce_if(lit3)
        m.add_circuit(arcs)
        m.add(E[k] == L[k]).only_enforce_if(unused)
        arcs_by_k[k] = lits

    for i in range(len(claims)):
        m.add(sum(visit[k, i] for k in range(n_adj) if (k, i) in visit)
              + drop[i] == 1)

    if lunch_break:
        # Same placement disjunction as the exact MILP, anchored to
        # the ROUTE span: the break ends before the route leaves home
        # (pre), starts after it returns (post), or rides on exactly
        # one traversed arc - INCLUDING the home legs - where travel
        # and break must both fit between departure and arrival. An
        # earlier draft anchored pre/post to service times and skipped
        # the home arcs; adversarial review produced an instance where
        # that relaxation certified a break overlapping the drive home.
        for k in range(n_adj):
            B = m.new_int_var(LUNCH["earliest_start"],
                              LUNCH["latest_start"], f"B_{k}")
            pre = m.new_bool_var(f"pre_{k}")
            post = m.new_bool_var(f"post_{k}")
            placements = [pre, post]
            dur = LUNCH["duration"]
            m.add(B + dur <= L[k]).only_enforce_if([pre, used[k]])
            m.add(B >= E[k]).only_enforce_if([post, used[k]])
            for (i, j), lit in arcs_by_k[k].items():
                on = m.new_bool_var(f"on_{k}_{i}_{j}")
                m.add_implication(on, lit)
                if i == "h":
                    dep = L[k]
                    t_ij = travel_min[k][node(j)]
                    arr = T[j]
                elif j == "h":
                    dep = T[i] + claims[i].service_minutes
                    t_ij = travel_min[node(i)][k]
                    arr = E[k]
                else:
                    dep = T[i] + claims[i].service_minutes
                    t_ij = travel_min[node(i)][node(j)]
                    arr = T[j]
                m.add(B >= dep).only_enforce_if(on)
                m.add(B + dur <= arr).only_enforce_if(on)
                m.add(arr >= dep + t_ij + dur).only_enforce_if(on)
                placements.append(on)
            m.add(sum(placements) == 1).only_enforce_if(used[k])

    travel_terms = []
    for k in range(n_adj):
        for (i, j), lit in arcs_by_k[k].items():
            fr = k if i == "h" else node(i)
            to = k if j == "h" else node(j)
            travel_terms.append(travel_min[fr][to] * lit)
    pen_terms = [config.effective_penalty(c.priority, c.age_days) * drop[i]
                 for i, c in enumerate(claims)]
    obj = sum(travel_terms) + sum(pen_terms)
    if balance:
        obj += BETA * sum(E[k] - L[k] for k in range(n_adj))
    m.minimize(obj)

    s = cp_model.CpSolver()
    s.parameters.max_time_in_seconds = float(time_limit_s)
    s.parameters.num_workers = workers
    status = s.solve(m)
    info = {"status": s.status_name(status),
            "proven": status == cp_model.OPTIMAL,
            "wall_s": round(s.wall_time, 1)}
    if status not in (cp_model.OPTIMAL, cp_model.FEASIBLE):
        return None, info
    info["bound"] = int(s.best_objective_bound)

    routes = []
    for k, a in enumerate(adjusters):
        succ = {i: j for (i, j), lit in arcs_by_k[k].items()
                if s.value(lit)}
        stops, cur, prev_node = [], succ.get("h"), k
        total_mi, total_tm, total_sv = 0.0, 0, 0
        while cur is not None and cur != "h":
            c = claims[cur]
            mi = miles[prev_node][node(cur)]
            tm = travel_min[prev_node][node(cur)]
            stops.append(Stop(c, s.value(T[cur]),
                              s.value(T[cur]) + c.service_minutes, mi, tm))
            total_mi += mi
            total_tm += tm
            total_sv += c.service_minutes
            prev_node = node(cur)
            cur = succ.get(cur)
        if stops:
            total_mi += miles[prev_node][k]
            total_tm += travel_min[prev_node][k]
        # Route start/end: L/E carry no objective pressure without the
        # balance term, so CP-SAT may leave them anywhere in the wait
        # slack. Restamp the tight OR-Tools convention (leave as late
        # as possible, return as early as possible) unless a toggle
        # makes the model's own values load-bearing (break anchoring /
        # span cost).
        if not stops:
            start = end = a.shift_start
        elif lunch_break or balance:
            start, end = s.value(L[k]), s.value(E[k])
        else:
            start = stops[0].arrival_min - stops[0].travel_min_from_prev
            end = stops[-1].departure_min + travel_min[prev_node][k]
        routes.append(Route(a, stops, start, end,
                            total_mi, total_tm, total_sv))
    dropped = [claims[i] for i in range(len(claims)) if s.value(drop[i])]
    sol = Solution(routes, dropped, unservable, int(s.objective_value),
                   sum(r.total_miles for r in routes),
                   sum(r.total_travel_min for r in routes))
    return sol, info