admin/goodwe-addon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

SAFETY-25: pin the claim-source set, so the tripwire is one
checks / unit (pull_request) Successful in 3m53s
Details

Browse Source 
The durability claim this PR shipped was FALSE, and the reviewer of #31
disproved it by building the fix. Part 10 said, in four committed places,
that it "goes RED the day a loop-side floor is built, so the exposure
cannot change size under a green suite". The reviewer implemented the
memo's own recommended Option B in main.py - a
`Claim.limit("reserve", P_LOOP, hi=0.0)` filed under a `loop_soc_floor` -
and the ENTIRE SUITE STAYED GREEN. Part 10 green. Mutation 13 went on
printing "NOTHING PREVENTS IT, ON PURPOSE" and the same 21 % -> 10.0 %,
on a tree where something DID prevent it. Pushed to an absurd
always-binding 95 % reserve it was caught only by SAFETY-24's pre-existing
part 7 wiring checks; part 10 was still green.

THE CAUSE IS STRUCTURAL, not a missing assertion. `replay_loop()`
hand-builds main.py's claim list as a literal
`[supervised, loop, peakguard.claim(...)]`, so it is blind to any claim
main.py ADDS - which is exactly where AC 2 would be built. That is this
repository's named failure mode: a check that observed something other
than what it claimed to observe, shipped as a guarantee inside the
document that recommends the change.

THE FIX IS 10f: THE CLAIM-SOURCE SET, PINNED, two ways because the two
fail differently.

  * STATICALLY, over `Controller.cycle()`'s own AST. Every
    `claims.append(...)` in it must be one the scan can read - an
    unreadable append FAILS rather than counting as nothing, which is how
    a scan like this goes quietly vacuous - and the sources it can file
    must be exactly {device, supervised, failsafe, safety, peak,
    arbitrage, maintenance, loop}. A ninth takes part 10 red on the line
    it is written, in every configuration, whether or not any fixture
    here reaches the state that files it.
  * DYNAMICALLY, through REAL `Controller.cycle()`s with main.py's OWN
    `resolve` spied - not a list this file built and hoped matched -
    swept over the WHOLE 0..100 % SoC range in the two states 10a-10c
    replay. Idle: exactly {device, supervised, loop}. Floor-limited:
    exactly {device, supervised, loop, peak}. A claim conditioned on a
    low pack, which is the shape a loop-side floor has, cannot hide from
    a sweep that visits 0 %.

And the fold is checked rather than assumed: main.py's two P_SAFETY
rating limits are both symmetric LIMITs, so replay_loop() modelling them
as one `supervised` claim at the tighter bound is exact.

MEASURED AGAINST THE COUNTEREXAMPLE. Reproduced first on the unfixed
tree (green, as reported). On this tree the reviewer's Option B at its
recommended default takes part 10 RED on 3 checks, and the 95 % variant
on 6. Nine mutations were run in all, each against a freshly copied tree
with no `__pycache__` present, mounted `:ro`; all nine red, no survivors,
no `+0 B` delta. MUTATION.md records why a second round was needed: the
first round's mutations were all aimed at DISARMING the guard, while the
claim part 10 actually made was about the guard being FIXED.

The four false sentences are replaced with what is now true, and each
says WHICH half catches WHICH shape - 10f for an arbiter claim
(Options B and C), 10a-10c for a clamp inside control.compute() or the
guard (Option D), which the replay drives directly.

AC 2 IS STILL OPEN AND NOTHING HERE BUILDS IT. The loop-side floor was a
test mutant, reverted before commit. The memo still says "Status: OPEN.
This is a memo, not a decision."

Also, non-blocking, from the same review: the NOT READY banner had
inherited the new scope sentence and become a ~550-character paragraph
that the row three lines below repeated verbatim. A checks() row may now
carry an optional short `banner` form; the floor-limited row does, and
`status_payload()` prefers it. THE ROW IS UNCHANGED - shortening that
would have been weakening AC 3 - and rows without the key are
unaffected. Graded, and mutated: dropping the short form goes red.

Merged origin/release/1.0 first (CI-02), keeping both tickets' CHANGELOG
entries.

Full suite green: 17/17 in goodwe-addon-ci (test_ci_gate.py needs `git`,
which CI-02 added to .gitea/ci/Dockerfile but the local image lacks; run
with it installed and it passes).

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
glenn schrooyen
2026-09-01 05:04:22 +02:00
co-authored by Claude Opus 5
parent 3cb9d3b7d9
commit e498dd4c1c
5 changed files with 329 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files
goodwe_controller/test_peakguard.py
+251 -4
View File
@@ -55,6 +55,15 @@ NOT allowed to be tested at:
sweeps `peak_soc_floor` over the whole schema range to show the answer
does not move by one part in ten thousand, and pins the four
operator-facing surfaces that now say so.
⚠️ THE REPLAY ALONE IS NOT THE TRIPWIRE, and believing it was is the
defect PR #31's reviewer found: `replay_loop()` hand-builds main.py's
claim list, so it is blind to any claim main.py ADDS - which is where a
loop-side floor would be built. 10f is what closes that. It pins the set
of claim sources `Controller.cycle()` can file, statically over its AST
and dynamically through real cycles swept over the whole SoC range, so a
ninth source takes this part red. A floor built inside `control.compute()`
or the guard instead moves 10a-10c's numbers directly, because the replay
drives both.
11. MUTATION. Eleven one-line defeats of these two guards, each run against
the checks above to show they are not vacuous.
@@ -110,6 +119,7 @@ from app.arbiter import (
)
from app.control import Tuning, compute, peak_at_risk
from app.freeze import FreezeWatchdogConfig, looks_frozen, rail_w
from app import main as app_main
from app.main import Controller
from app.p1 import (
PLAUSIBLE_MAX_W, P1Ingest, QUARTER_S, QuarterAverager, make_sample,
@@ -1869,6 +1879,15 @@ print("10. SAFETY-25: the loop discharges past this floor, and how far past")
# deleting or inverting these checks along with
# DECISIONS/SAFETY-25-loop-soc-floor.md's open status.
#
# ⚠️ AND IT IS 10f THAT MAKES THAT SENTENCE TRUE, NOT 10a-10c. The replay
# below hand-builds main.py's claim list, so on its own it is blind to a
# floor filed as a NEW CLAIM - the memo's own Option B, and PR #31's
# reviewer proved the blindness by building it: the whole suite stayed
# green. 10f pins the claim-source set so that shape goes red too. The two
# halves cover the two places a loop-side floor can be built: 10f for an
# arbiter claim in Controller.cycle(), 10a-10c for a clamp inside
# control.compute() or peakguard, which the replay drives directly.
#
# ⚠️ AND IT DRIVES THE SHIPPED CODE, not a transcription of it: control.compute()
# at the shipped tuning, peakguard.decide(), peakguard.claim() and
# arbiter.resolve(), in main.py's order. Break any of them and this part moves -
@@ -2190,6 +2209,36 @@ check("the STATUS PAGE row for a floor-limited guard says the floor bounds the "
and "BOUNDS THE PEAK GUARD ONLY" in rows25[0]["text"].upper()
and "not a pack-wide reserve" in rows25[0]["text"]
and "BMS cut-out" in rows25[0]["text"])
# ⚠️ THE ROW KEEPS THE WHOLE SENTENCE; THE BANNER DOES NOT. status_payload()
# renders the first not-ok/not-warn row as the page headline, so before this the
# loudest element on the page was the same 550-character paragraph the row three
# lines below repeats verbatim. The scope wording is not weakened anywhere it
# matters - it is the ROW that an operator reads to learn what the floor bounds,
# and the row is asserted in full above.
_pay25 = c25.status_payload()
print(f" banner ({len(_pay25['banner'])} chars): {_pay25['banner']}")
# The guard's own line, which the banner has always carried and which is not
# this card's to shorten: `explain()` is SAFETY-24's sentence about a quarter
# that is going over the ceiling.
_expl25 = rows25[0]["text"].split(". NOTE:")[0]
check("the NOT READY banner is the guard's own line plus a SHORT scope tag - "
"under 90 characters of it - and NOT the row's whole paragraph repeated "
"at the top of the page. A headline that is a paragraph is a headline "
"nobody reads, and this one is followed three lines later by the same "
"text verbatim",
_pay25["level"] == "bad"
and _pay25["banner"].startswith("NOT READY")
and _expl25 in _pay25["banner"]
and "BOUNDS THE PEAK GUARD ONLY" in _pay25["banner"].upper()
and "not a pack-wide reserve" not in _pay25["banner"]
and len(_pay25["banner"]) - len(_expl25) < 90)
check("...and shortening the banner did NOT shorten the row: the full AC 3 "
"sentence is still what the checks list carries, which is the surface the "
"operator reads to find out what the floor does not bound",
len(rows25[0]["text"]) > 300
and "not a pack-wide reserve" in rows25[0]["text"]
and "BMS cut-out" in rows25[0]["text"])
_body25 = hass25.last_message("goodwe_peak_soc_floor")
print(f" notification body tail: ...{_body25[-180:]}")
check("...and so does the persistent notification, which outlives the page and "
@@ -2200,6 +2249,198 @@ check("...and so does the persistent notification, which outlives the page and "
and "BMS cut-out" in _body25)
# --- 10f. THE CLAIM-SOURCE SET, PINNED --------------------------------------
# ⚠️ THIS IS WHAT MAKES 10a-10c A TRIPWIRE RATHER THAN A SNAPSHOT, and it is
# here because the first cut of this part did NOT have it and the tripwire
# sentence above was therefore FALSE. `replay_loop()` hand-builds main.py's
# claim list as a literal `[supervised, loop, peakguard.claim(...)]`, so it is
# structurally BLIND to any claim main.py ADDS - which is exactly where AC 2
# would be built. PR #31's reviewer proved it by building the fix: they
# implemented the memo's own Option B in main.py - a
# `Claim.limit("reserve", P_LOOP, hi=0.0)` filed under a `loop_soc_floor` - and
# the whole suite stayed GREEN, this part included, while mutation 13 went on
# printing "NOTHING PREVENTS IT, ON PURPOSE" and the same 21 % -> 10.0 % on a
# tree where something DID prevent it. Pushed to an always-binding 95 % reserve
# it was still green here. That is this repository's named failure mode - a
# check that observed something other than what it claimed to observe - and it
# had been committed as a guarantee in four places.
#
# So the MODEL is now pinned to the MODELLED THING, two ways, because the two
# fail differently:
#
# * STATICALLY, over `Controller.cycle()`'s own AST. Every `claims.append(...)`
# in it must be one this scan can read, and the set of sources it can file
# must be exactly the eight below. A ninth - a `reserve`, whatever it ends
# up called - takes this red on the line it is written, in every
# configuration, whether or not any fixture here reaches the state that
# files it.
# * DYNAMICALLY, through REAL `Controller.cycle()`s with main.py's OWN
# `resolve` spied, swept over the WHOLE state-of-charge range in the two
# states 10a-10c replay. The static scan pins the NAMES; this pins that the
# names actually reach the arbiter and that nothing else does. A claim that
# appears only at a low SoC - precisely the shape a loop-side floor has -
# cannot hide from a sweep that visits 0 %.
#
# The third way a loop-side floor could be built is INSIDE `control.compute()`
# or the guard (the memo's Option D), and that needs nothing new here: the
# replay above drives both, so it moves the numbers in 10a-10c directly.
# The whole inventory of claim sources main.py can file, in any configuration.
# ⚠️ ADDING TO THIS SET IS A DELIBERATE ACT. If you are here because you added a
# claim to `Controller.cycle()` and this went red, that is the tripwire working:
# read DECISIONS/SAFETY-25-loop-soc-floor.md, answer AC 2 in it, and update
# part 10 and mutation 13 rather than this line alone.
CLAIM_SOURCES = {"device", "supervised", "failsafe", "safety", "peak",
"arbitrage", "maintenance", "loop"}
# The two appends that file a claim BUILT ELSEWHERE in the method. Named here so
# that a THIRD such indirection reads as an unrecognised append rather than as a
# silent gap in the scan.
CLAIM_BUILDERS = {"peak_claim": "peak", "arb_claim": "arbitrage"}
_main_tree = ast.parse((HERE / "app" / "main.py").read_text(encoding="utf-8"))
_cycle_def = next(
f
for cls in ast.walk(_main_tree)
if isinstance(cls, ast.ClassDef) and cls.name == "Controller"
for f in cls.body
if isinstance(f, (ast.FunctionDef, ast.AsyncFunctionDef)) and f.name == "cycle")
def claim_source(node):
"""The source string an appended expression files, or None if unreadable."""
if (isinstance(node, ast.Call) and isinstance(node.func, ast.Attribute)
and isinstance(node.func.value, ast.Name)
and node.func.value.id == "Claim"
and node.func.attr in ("set", "limit") and node.args
and isinstance(node.args[0], ast.Constant)
and isinstance(node.args[0].value, str)):
return node.args[0].value
if isinstance(node, ast.Name):
return CLAIM_BUILDERS.get(node.id)
return None
_appends = [n for n in ast.walk(_cycle_def)
if isinstance(n, ast.Call) and isinstance(n.func, ast.Attribute)
and n.func.attr == "append" and isinstance(n.func.value, ast.Name)
and n.func.value.id == "claims" and len(n.args) == 1]
_filed = [claim_source(n.args[0]) for n in _appends]
print(f" main.py Controller.cycle() files {len(_appends)} claims, from "
f"{len(CLAIM_SOURCES)} sources: "
f"{', '.join(sorted({s for s in _filed if s}))}")
check("every `claims.append(...)` in main.py's cycle is one this scan can READ "
"- a claim filed through a shape it cannot parse has to fail here rather "
"than be counted as nothing, which is how a scan like this goes quietly "
"vacuous", len(_appends) >= 10 and None not in _filed)
check("...and the sources that cycle can file are EXACTLY these eight. THIS is "
"what makes part 10 a tripwire: replay_loop() hand-builds the claim list, "
"so it cannot see a claim main.py adds - and a loop-side SoC reserve is a "
"NINTH source. Filing one takes this line red on the line it is written",
set(_filed) == CLAIM_SOURCES)
def cycle_claims(run):
"""The claim list main.py's OWN `resolve` was handed, from a real cycle.
⚠️ SPIED AT `app.main.resolve`, not reconstructed. The point of the whole
sub-part is that this file must stop believing its own transcription of
main.py's claim list, so the observation has to be taken from main.py's
call site and nowhere else.
"""
seen = []
real = app_main.resolve
def _spy(claims):
seen.append(list(claims))
return real(claims)
app_main.resolve = _spy
try:
out = run()
finally:
app_main.resolve = real
return out, (seen[-1] if seen else [])
def run_cycle(ctl):
asyncio.get_event_loop_policy().new_event_loop().run_until_complete(
ctl.cycle())
return ctl
# The IDLE state, which is what 10a and 10b replay: a sustained house load well
# under the ceiling, so the guard files nothing and the loop is alone with the
# safety limits. One controller, cycled once per state of charge - the load is
# the evening's own mean, so this is the evening's cycle at every SoC it passes
# through and at every SoC it does not.
EVE_MEAN_W = sum(EVE_W) / len(EVE_W)
c_src, _ = built(TMP / "sources.db")
c_src.auto, c_src.batt, c_src.soc = True, 0.0, 100.0
_blk_src = datetime.fromtimestamp(
int(datetime.now(timezone.utc).timestamp() // QUARTER_S) * QUARTER_S,
timezone.utc)
for _i in range(0, int((datetime.now(timezone.utc)
- _blk_src).total_seconds()) + 1, 5):
c_src.p1.submit(sample(_blk_src + timedelta(seconds=_i), EVE_MEAN_W))
c_src.grid = EVE_MEAN_W
# ⚠️ THE WHOLE RANGE, INCLUDING 0 %. A reserve conditioned on a low pack is
# invisible to a sweep that starts at a comfortable one, and "conditioned on a
# low pack" is the definition of the claim this is watching for.
SRC_SWEEP = sorted({float(s) for s in range(0, 101, 5)}
| {BMS_FLOOR, SOC_FLOOR, EVE_START, EV["end_soc"]})
IDLE_SRC = {}
for _soc in SRC_SWEEP:
c_src.soc = _soc
_, _cl = cycle_claims(lambda: run_cycle(c_src))
IDLE_SRC[_soc] = frozenset(c.source for c in _cl)
IDLE_SETS = set(IDLE_SRC.values())
print(f" a real idle cycle at {len(SRC_SWEEP)} states of charge "
f"({min(SRC_SWEEP):.0f}..{max(SRC_SWEEP):.0f} %) files: "
+ " | ".join(sorted(", ".join(sorted(s)) for s in IDLE_SETS)))
check("a REAL Controller.cycle() in the state 10a and 10b replay - auto, P1 "
"healthy, the guard idle under an 830 W load - hands the arbiter exactly "
"{device, supervised, loop}, at EVERY state of charge from 0 to 100 %. "
"Not one source appears at a low pack that does not appear at a full one, "
"which is the claim a loop-side reserve would break",
IDLE_SETS == {frozenset({"device", "supervised", "loop"})}
and len(SRC_SWEEP) >= 21 and min(SRC_SWEEP) == 0.0)
# The ENGAGED state, which is what 10c replays: the guard over its ceiling and
# floor-limited on an empty pack. Two states of charge, both under the floor,
# because that is the band a loop-side reserve would be conditioned on.
HOT_SRC = set()
for _soc in (SOC_FLOOR - 1.0, 0.0):
(c_h, _h, _n, _r), _cl = cycle_claims(
lambda: wired_over_limit(TMP / f"sources-hot-{_soc:.0f}.db",
LIVE_MAX_W * 0.3, soc=_soc))
HOT_SRC.add(frozenset(c.source for c in _cl))
_hot_limited = c_h.peak_guard is not None and c_h.peak_guard.floor_limited
print(" a real FLOOR-LIMITED cycle files: "
+ " | ".join(sorted(", ".join(sorted(s)) for s in HOT_SRC)))
check("...and with the guard actually FLOOR-LIMITED - 10c's state, on an empty "
"pack - the set gains the guard's own claim and NOTHING ELSE. So the "
"three sources replay_loop() hand-builds are the whole cycle in both of "
"the states it replays, rather than a subset of it that happened to "
"match", _hot_limited
and HOT_SRC == {frozenset({"device", "supervised", "loop", "peak"})})
# THE FOLD, stated rather than assumed. replay_loop() files ONE `supervised`
# limit where main.py files two - `device`, from the inverter's own reported
# range, and `supervised`, from `max_w`. Both are symmetric P_SAFETY limits, so
# the tighter one binds and modelling them as one is exact, not an
# approximation. Checked on the claims the real cycle actually filed.
_safety_lims = [c for c in _cl if c.source in ("device", "supervised")]
check("...and replay_loop() folding main.py's TWO P_SAFETY rating limits into "
"its one `supervised` claim is exact rather than convenient: both are "
"symmetric LIMIT claims at P_SAFETY, so the tighter of the pair binds and "
"one claim at the tighter bound is the same constraint",
len(_safety_lims) == 2
and all(c.kind == "limit" and c.priority == P_SAFETY
and near(c.lo, -c.hi) and c.hi > 0.0 for c in _safety_lims)
and near(LOOP_MAX_W, read_option("max_w")))
# ==============================================================================
print()
print("11. mutation: what defeats these guards, and which check bites")
@@ -2318,9 +2559,13 @@ MUT.append(("floor-limited reported but the floor still filed at `wanted`",
#
# ⚠️ SAFETY-25 CHANGED WHAT "NOT CAUGHT" MEANS HERE, AND ONLY THAT. Nothing
# PREVENTS it - the code is unchanged and a `lo` limit still cannot cap a
# discharge - but part 10 now MEASURES it on the committed capture, so the
# exposure can no longer change size under a green suite. That is the whole
# difference between an uncaught mutation and an unrecorded one.
# discharge - but part 10 now MEASURES it on the committed capture and 10f
# PINS THE CLAIM-SOURCE SET around it, so the exposure can no longer change
# size under a green suite: file the reserve and 10f goes red on the line
# it is written. That is the whole difference between an uncaught mutation
# and an unrecorded one. ⚠️ THE PIN IS LOAD-BEARING FOR THIS ROW. Without
# it this rationale was FALSE - PR #31's reviewer built the memo's Option B
# and this row went on printing "NOTHING PREVENTS IT" under a green suite.
_loop_unbounded = near(r_disch.target_w, 1800.0, 1e-9)
MUT.append(("the grid-following loop discharging past this floor",
not _loop_unbounded,
@@ -2330,7 +2575,9 @@ MUT.append(("the grid-following loop discharging past this floor",
"that the loop has no SoC guard. SAFETY-25 part 10 now MEASURES "
"it: one sustained-load evening, "
f"{EVE_START:.0f} % -> {EV['end_soc']:.1f} % on the reference pack, "
f"identical at every peak_soc_floor the schema accepts"))
f"identical at every peak_soc_floor the schema accepts - and 10f "
f"pins the claim-source set, so building the floor takes this "
f"part RED instead of leaving this row quietly stale"))
for label, caught, where in MUT:
print(f" {'caught' if caught else 'NOT CAUGHT':>10} {label}")