OPT-11 P1 review: a mode never beats the peak floor; SAFETY-03 kept at the rail

- arbiter: a mode now gives way only as far as a site limit above P_MODE
  needs (_unit_modes at N>1, _site_mode at N=1). Before, the modes were
  released only when the peak floor bound S, so a trade won above the floor
  kept them and placed less than the floor (an Off battery at N=1 went to 0 W
  under a 2000 W floor), and a capped grid charge released an Off battery.
  While released, held batteries weigh 0 and carry only the shortfall.
- control: a side of the loop bound that reaches the rail is left to the
  rail, so SAFETY-03's i_clamped still fires with every battery on.
- opt11_identity: add maintenance scenarios and a 6000-case arbiter digest
  (peak floor, trades, maintenance, refusals); a5dddad digests pinned.
- tests: test_arbiter part 11, test_control (g), test_modes (0/1 refused,
  loop bound counts live units only). DOCS and DECISIONS say the new rule.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UBtqPrG1mDLXzTEBnhEkkV

goodwe_controller/tools/opt11_identity.py

@@ -14,7 +14,8 @@ printed on a5dddad (they are written there, with that provenance).
 
 Scenarios: N=1 and N=2, a house that steps, a surplus of solar, a battery
 reaching its cut-out level and, at N=2, a battery that refuses (SAFETY-73) and
-is probed.
+is probed; the same with both batteries due for maintenance; and the arbiter
+alone on 6000 seeded claim lists (peak floor, trades, maintenance, refusals).
 """
 
 import asyncio
@@ -187,6 +188,11 @@ def digest(rows) -> str:
     return hashlib.sha256(repr(rows).encode()).hexdigest()
 
 
+# Both batteries due for their monthly cycle at the run's first cycle, so the
+# maintenance slot, its pins and its shaping are in every row (PR #149 review:
+# the scenarios above never reach maintenance).
+_MAINT = {"maintenance_enabled": True, "maintenance_start_hour": 11}
+
 SCENARIOS = {
     "n1": dict(n=1),
     "n2": dict(n=2),
@@ -194,11 +200,78 @@ SCENARIOS = {
     # option without turning it on runs. The old tree ignores the key.
     "n1_off": dict(n=1, extra={"battery_modes_enabled": False}),
     "n2_off": dict(n=2, extra={"battery_modes_enabled": False}),
+    "n1_maint": dict(n=1, extra=_MAINT),
+    "n2_maint": dict(n=2, extra=_MAINT),
 }
 
 
+def resolve_rows(cases=6000, seed=1149):
+    """The arbiter alone, on seeded random claim lists with no mode claim:
+    the peak floor, a trade, the loop, maintenance pins and shaping, SAFETY-73
+    refusal limits and probe pins, packs, ratings and safety stops, at one to
+    three inverters. The peak guard needs a P1 meter the cycle runs above do
+    not have, so its claim is graded here. Uses only a5dddad's API."""
+    import random
+    from math import inf
+    from app import arbiter as a
+    rnd = random.Random(seed)
+    rows = []
+    for _ in range(cases):
+        n = rnd.choice((1, 2, 3))
+        units = ("inverter-a", "inverter-b", "inverter-c")[:n]
+
+        def act(src, u):
+            return a.actuator_for(src, u, units)
+        cl = []
+        for u in units:
+            r = rnd.choice((1000.0, 2500.0, 5000.0))
+            cl.append(a.Claim.limit("supervised", a.P_SAFETY, -r, r, "max_w",
+                                    actuator=act("supervised", u)))
+            x = rnd.random()
+            if x < 0.15:
+                cl.append(a.Claim.limit("pack", a.P_SAFETY, *rnd.choice(
+                    ((0.0, inf), (-inf, 0.0))), actuator=act("pack", u)))
+            elif x < 0.25:
+                cl.append(a.Claim.limit("refused", a.P_SAFETY, *rnd.choice(
+                    ((-inf, 0.0), (0.0, inf))), "refused",
+                    actuator=act("refused", u)))
+            elif x < 0.32:
+                cl.append(a.Claim.set("refused", a.P_SAFETY,
+                                      rnd.choice((200.0, -200.0)), "probing",
+                                      actuator=act("refused", u)))
+            elif x < 0.42:
+                cl.append(a.Claim.set("maintenance", a.P_MAINTENANCE,
+                                      rnd.choice((2000.0, -2000.0)), "drain",
+                                      actuator=act("maintenance", u)))
+            elif x < 0.5:
+                cl.append(a.Claim.limit("maintenance", a.P_MAINT_SHAPE,
+                                        hi=-rnd.choice((500.0, 2000.0)),
+                                        actuator=act("maintenance", u)))
+        y = rnd.random()
+        if y < 0.8:
+            cl.append(a.Claim.set("loop", a.P_LOOP, rnd.uniform(-8000, 8000)))
+        if 0.5 < y < 0.9:
+            cl.append(a.Claim.set("arbitrage", a.P_ARBITRAGE,
+                                  rnd.uniform(-6000, 6000)))
+        if rnd.random() < 0.6:
+            cl.append(a.Claim.limit("peak", a.P_PEAK,
+                                    lo=rnd.uniform(-4000, 6000)))
+        if rnd.random() < 0.05:
+            cl.append(a.Claim.set("safety", a.P_SAFETY, 0.0, "stopped"))
+        rnd.shuffle(cl)
+        r = a.resolve(cl, units=units if n > 1 else None,
+                      weights=(None if rnd.random() < 0.5 else
+                               {u: rnd.choice((0.0, 0.5, 1.0)) for u in units}))
+        rows.append((r.target_w, r.winner, r.bound_by, r.contradiction,
+                     tuple(sorted((r.shares or {}).items())), r.allocated,
+                     r.allocation.explain() if r.allocation else None))
+    return rows
+
+
 def digests() -> dict:
-    return {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+    out = {k: digest(run(**v)[0]) for k, v in SCENARIOS.items()}
+    out["resolve"] = digest(resolve_rows())
+    return out
 
 
 if __name__ == "__main__":
@@ -208,5 +281,9 @@ if __name__ == "__main__":
             print(f"{k}: {len(rows)} cycles, loop reasons "
                   f"{sorted({r[3] for r in rows})}, arbiter "
                   f"{sorted({r[6].split(' ->')[0] for r in rows})}")
+        rr = resolve_rows()
+        print(f"resolve: {len(rr)} claim lists, winners "
+              f"{sorted({r[1] for r in rr})}, bound by "
+              f"{sorted({str(r[2]).split('(')[0] for r in rr})}")
     for k, v in digests().items():
         print(f"{k} {v}")