"""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