goodwe-addon/goodwe_controller/tools/symmetric_message_scan.py

#!/usr/bin/env python3
"""SAFETY-38 round 3 - the SYMMETRIC-PREDICATE FAMILY, enumerated by machine.

WHAT IT LOOKS FOR, in one sentence: a call whose argument ORDER is load-bearing
for a MESSAGE and for nothing else - so swapping the arguments cannot change a
single decision, is invisible to every behavioural check BY CONSTRUCTION, and
still inverts what an operator reads on the line they are woken by.

⚠️ WHY IT IS AN INSTRUMENT AND NOT A LIST. MUTATION.md habit 12: "a hand-written
scope is the same defect as a hand-written fixture". SAFETY-38 met this family
twice - `SignWatchdog.evaluate()` called at `main.py:1620` (round 2) and at
`main.py:1560` (round 3), ONE CALL SITE OVER - and both times the enumeration
was written by hand and both times it was short. So the scope is computed.

------------------------------------------------------------------------------
THE TEST IT APPLIES

For every function in `app/` and every pair (a, b) of its parameters:

    rename a -> b and b -> a throughout the body, ERASE every string literal
    and f-string, CANONICALISE the commutative operators, and compare the two
    ASTs. If they are EQUAL, nothing the function DECIDES can depend on which
    of the two values arrived first.

Then, if some f-string in that function mentions BOTH names, the order is
load-bearing for the MESSAGE ALONE, and the pair is reported.

⚠️ THE SYMMETRY IS USUALLY ONE CALL DEEP, WHICH IS WHY THIS IS A FIXPOINT AND
NOT A SINGLE PASS. `SignWatchdog.evaluate()` does not compare its two arguments
itself; it hands them to `watchdog.disagrees()`, whose body is
`commanded_w * against_w < 0 and abs(commanded_w) > min_w and
abs(against_w) > min_w`. An intraprocedural scan sees `disagrees(a, b, ...)`
become `disagrees(b, a, ...)`, calls that a difference, and reports NOTHING -
which is exactly what the first version of this file did, on a tree where the
answer was already known. So symmetric pairs discovered in one round are used
to canonicalise CALL ARGUMENTS in the next, and the scan iterates until the set
stops growing.

CANONICALISATION, and each entry is here because something in `app/` needs it:
  * `a * b`, `a + b`               - operand order sorted (`disagrees`)
  * `x and y`, `x or y`            - term order sorted (`disagrees`' conjunction)
  * `a == b`, `a != b`             - operand order sorted
  * `min(a, b)`, `max(a, b)`       - argument order sorted (`clamp_to_rating`)
  * a tuple/list/set LITERAL being iterated - element order sorted
    (`rail_ceiling_w`'s `for bound in (dev_min, dev_max)`)
  * arguments at a KNOWN SYMMETRIC POSITION PAIR of a called function - sorted
    (the fixpoint)

Equal ASTs after the rename is a SUFFICIENT condition for behavioural
invariance, not a necessary one. This scan therefore UNDER-REPORTS and says so
- see "WHAT THIS CANNOT SEE". That is the conservative direction: everything it
names is genuinely in the family, and the gap is stated with its shape.

Run: `python3 tools/symmetric_message_scan.py [--quiet]` from goodwe_controller/.
Exit status is 0 always; it is a census, not a gate.
"""
from __future__ import annotations

import ast
import itertools
import pathlib
import sys

APP = pathlib.Path(__file__).resolve().parent.parent / "app"

# {function name: {(i, j) positional index pairs that are interchangeable}}
_SYM: dict[str, set[tuple[int, int]]] = {}
# ⚠️ AND THE SAME THING FOR KEYWORD-ONLY PARAMETERS, WHICH THE FIRST FIXPOINT
# DROPPED. `freeze.rail_ceiling_w(max_w, *, dev_min=None, dev_max=None)` folds
# its two device bounds with `for bound in (dev_min, dev_max): ceiling = min(
# ceiling, abs(bound))` - perfectly symmetric - and `rail_unreachable()` prints
# `"a range of {dev_min}...{dev_max} W"`. A scan that only followed POSITIONAL
# arguments called that pair asymmetric and hid a genuine member of the family
# behind an implementation detail of its own traversal.
_SYM_KW: dict[str, set[tuple[str, str]]] = {}


class _Canon(ast.NodeTransformer):
    """Rename one pair, erase every string, and sort what commutes."""

    COMMUTATIVE_BINOP = (ast.Mult, ast.Add, ast.BitAnd, ast.BitOr, ast.BitXor)
    COMMUTATIVE_CMP = (ast.Eq, ast.NotEq)
    COMMUTATIVE_CALL = ("min", "max")

    def __init__(self, a: str = "", b: str = ""):
        self.a, self.b = a, b

    # -- the rename ---------------------------------------------------------
    def visit_Name(self, node):
        self.generic_visit(node)
        if self.a and node.id == self.a:
            node.id = self.b
        elif self.b and node.id == self.b:
            node.id = self.a
        return node

    def visit_arg(self, node):
        if self.a and node.arg == self.a:
            node.arg = self.b
        elif self.b and node.arg == self.b:
            node.arg = self.a
        return node

    # -- the message is exactly what is ALLOWED to differ --------------------
    def visit_JoinedStr(self, node):
        return ast.Constant(value="<msg>")

    def visit_Constant(self, node):
        if isinstance(node.value, str):
            return ast.Constant(value="<msg>")
        return node

    # -- canonicalisation ---------------------------------------------------
    def visit_BinOp(self, node):
        self.generic_visit(node)
        if isinstance(node.op, self.COMMUTATIVE_BINOP):
            l, r = ast.dump(node.left), ast.dump(node.right)
            if l > r:
                node.left, node.right = node.right, node.left
        return node

    def visit_BoolOp(self, node):
        self.generic_visit(node)
        node.values = sorted(node.values, key=ast.dump)
        return node

    def visit_Compare(self, node):
        self.generic_visit(node)
        if (len(node.ops) == 1
                and isinstance(node.ops[0], self.COMMUTATIVE_CMP)):
            l, r = ast.dump(node.left), ast.dump(node.comparators[0])
            if l > r:
                node.left, node.comparators[0] = node.comparators[0], node.left
        return node

    def visit_For(self, node):
        self.generic_visit(node)
        if isinstance(node.iter, (ast.Tuple, ast.List, ast.Set)):
            node.iter.elts = sorted(node.iter.elts, key=ast.dump)
        return node

    def visit_Call(self, node):
        self.generic_visit(node)
        name = (node.func.id if isinstance(node.func, ast.Name)
                else node.func.attr if isinstance(node.func, ast.Attribute)
                else "")
        if name in self.COMMUTATIVE_CALL and len(node.args) == 2 \
                and not node.keywords:
            if ast.dump(node.args[0]) > ast.dump(node.args[1]):
                node.args = [node.args[1], node.args[0]]
            return node
        # the fixpoint: sort the arguments at a known symmetric position pair
        for (i, j) in sorted(_SYM.get(name, ())):
            if j < len(node.args):
                if ast.dump(node.args[i]) > ast.dump(node.args[j]):
                    node.args[i], node.args[j] = node.args[j], node.args[i]
        # ...and at a known symmetric KEYWORD pair
        for (ka, kb) in sorted(_SYM_KW.get(name, ())):
            ia = next((k for k, w in enumerate(node.keywords)
                       if w.arg == ka), None)
            ib = next((k for k, w in enumerate(node.keywords)
                       if w.arg == kb), None)
            if ia is not None and ib is not None:
                va, vb = node.keywords[ia].value, node.keywords[ib].value
                if ast.dump(va) > ast.dump(vb):
                    node.keywords[ia].value, node.keywords[ib].value = vb, va
        return node


def _canon(src: str, a: str = "", b: str = "") -> str:
    return ast.dump(_Canon(a, b).visit(ast.parse(src)))


# ⚠️ THE SINK IS HALF THE CLAIM, AND THE FIRST VERSION OF THIS SCAN OMITTED IT.
# Erasing every f-string makes a function whose ONLY job is to FORMAT one look
# perfectly symmetric - `store.unit_key(unit, key)` and
# `mqtt.unit_state_key(slug, key)` both came back as "the swap changes nothing"
# when in fact the string IS the return value: swapping there renames a STORE
# KEY and an MQTT TOPIC, which is a behaviour change every ordinary check can
# see, and MAINT-05 is the card about exactly that. The family this file is
# looking for is narrower - the string goes to an OPERATOR and to nobody else -
# so each sentence is classified by where it GOES.
_LOG_SINKS = {"debug", "info", "warning", "warn", "error", "exception",
              "critical", "log", "log_event", "notify", "persistent_notify"}
_MSG_ATTRS = {"reason", "message", "msg", "banner", "detail", "note",
              "explanation", "why", "text"}


def _sink_of(fn, target, follow=True):
    """RETURN | MESSAGE | OTHER - where this f-string node ends up.

    `follow` is the ONE step of data flow below, and it is off on the recursive
    call so that two locals that interpolate each other cannot loop forever.
    """
    for stmt in ast.walk(fn):
        for child in ast.iter_child_nodes(stmt):
            pass
    # a parent map is cheaper than repeated walks
    parent = {}
    for node in ast.walk(fn):
        for child in ast.iter_child_nodes(node):
            parent[id(child)] = node
    cur = target
    while id(cur) in parent:
        up = parent[id(cur)]
        if isinstance(up, ast.Return):
            return "RETURN"
        if isinstance(up, ast.Raise):
            return "MESSAGE"
        if isinstance(up, (ast.Assign, ast.AnnAssign, ast.AugAssign)):
            tgts = (up.targets if isinstance(up, ast.Assign) else [up.target])
            for t in tgts:
                if isinstance(t, ast.Attribute) and t.attr in _MSG_ATTRS:
                    return "MESSAGE"
                if isinstance(t, ast.Name) and t.id in _MSG_ATTRS:
                    return "MESSAGE"
            # ⚠️ ONE STEP OF DATA FLOW, AND IT IS NOT OPTIONAL. `main.py`'s
            # status page builds the device-range clause into a LOCAL - `_cap =
            # "; ".join(f"... accepts only {inv.dev_min}...{inv.dev_max} W" ...)`
            # - and only then concatenates it into the row's `text`. Stopping at
            # the assignment scores that OTHER and drops a member this card
            # MEASURED as a live survivor (0 bytes, 21/21 green). So: if the
            # name assigned here is interpolated into some other sentence in the
            # same function that IS a message, this one is a message too.
            if follow:
                for t in tgts:
                    if isinstance(t, ast.Name) and _feeds_a_message(fn, t.id,
                                                                    up):
                        return "MESSAGE"
            return "OTHER"
        if isinstance(up, ast.Call):
            name = (up.func.attr if isinstance(up.func, ast.Attribute)
                    else up.func.id if isinstance(up.func, ast.Name) else "")
            if name in _LOG_SINKS:
                return "MESSAGE"
            if name == "append":
                return "MESSAGE"
        cur = up
    return "OTHER"


def _feeds_a_message(fn, name, origin):
    """Is `name` interpolated into some OTHER sentence that IS a message?"""
    for node in ast.walk(fn):
        if not isinstance(node, ast.JoinedStr) or node is origin:
            continue
        used = any(isinstance(x, ast.Name) and x.id == name
                   for v in node.values if isinstance(v, ast.FormattedValue)
                   for x in ast.walk(v.value))
        if used and _sink_of(fn, node, follow=False) == "MESSAGE":
            return True
    return False


def _fstring_pairs(fn, params):
    """{(a, b): [(lineno, sink), ...]} for every f-string mentioning both."""
    out: dict[tuple[str, str], list[tuple[int, str]]] = {}
    for node in ast.walk(fn):
        if not isinstance(node, ast.JoinedStr):
            continue
        used = set()
        for v in node.values:
            if isinstance(v, ast.FormattedValue):
                for x in ast.walk(v.value):
                    if isinstance(x, ast.Name) and x.id in params:
                        used.add(x.id)
        if len(used) < 2:
            continue
        sink = _sink_of(fn, node)
        for a, b in itertools.combinations(sorted(used), 2):
            out.setdefault((a, b), []).append((node.lineno, sink))
    return out


def _functions():
    for path in sorted(APP.glob("*.py")):
        tree = ast.parse(path.read_text())
        for fn in ast.walk(tree):
            if isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
                yield path, fn


def _positional(fn):
    return [a.arg for a in fn.args.posonlyargs + fn.args.args
            if a.arg != "self"]


def _all_params(fn):
    return _positional(fn) + [a.arg for a in fn.args.kwonlyargs]


def _body_src(fn) -> str:
    """The function's BODY alone.

    ⚠️ NOT `ast.unparse(fn)`. The signature lists the parameters IN ORDER, so
    renaming a pair inside it reorders `arguments.args` and every function in
    the repository compares unequal to itself - which is what the first version
    of this scan did, silently, and it reported an empty family on a tree whose
    answer was already known. The claim being tested is about what the body
    DECIDES, and the body is what is compared.
    """
    return ast.unparse(ast.Module(body=list(fn.body), type_ignores=[]))


def _symmetric(fn, a, b) -> bool:
    src = _body_src(fn)
    return _canon(src) == _canon(src, a, b)


def build_fixpoint(max_rounds=6):
    """Fill `_SYM` until it stops growing. Returns the number of rounds."""
    for round_no in range(1, max_rounds + 1):
        grew = False
        for _path, fn in _functions():
            pos = _positional(fn)
            kws = [a.arg for a in fn.args.kwonlyargs]
            if len(pos) + len(kws) < 2:
                continue
            found = set()
            for i, j in itertools.combinations(range(len(pos)), 2):
                if _symmetric(fn, pos[i], pos[j]):
                    found.add((i, j))
            found_kw = set()
            for ka, kb in itertools.combinations(sorted(kws), 2):
                if _symmetric(fn, ka, kb):
                    found_kw.add((ka, kb))
            if found - _SYM.get(fn.name, set()):
                _SYM.setdefault(fn.name, set()).update(found)
                grew = True
            if found_kw - _SYM_KW.get(fn.name, set()):
                _SYM_KW.setdefault(fn.name, set()).update(found_kw)
                grew = True
        if not grew:
            return round_no
    return max_rounds


def scan():
    rows = []
    for path, fn in _functions():
        params = _all_params(fn)
        if len(params) < 2:
            continue
        kwonly = {a.arg for a in fn.args.kwonlyargs}
        for (a, b), hits in sorted(_fstring_pairs(fn, params).items()):
            rows.append(dict(
                module=path.name, func=fn.name, line=fn.lineno, a=a, b=b,
                msg_lines=sorted(set(hits)),
                sinks={s for _l, s in hits},
                symmetric=_symmetric(fn, a, b),
                positional=not (a in kwonly or b in kwonly)))
    return rows


# =============================================================================
# CENSUS B - THE SENTENCE THE CALLER BUILDS
# =============================================================================
# ⚠️ CENSUS A ABOVE ONLY SEES A CALLEE FORMATTING ITS OWN PARAMETERS, AND THAT
# WAS GAP 2 OF ITS OWN "WHAT THIS CANNOT SEE" LIST. The other half of the family
# never passes through a function boundary at all: an operator sentence built
# where it is printed, interpolating two values that are INTERCHANGEABLE BY
# SHAPE. `main.py`'s status page says
#
#     f"{inv.setpoint_entity or inv.name} accepts only {inv.dev_min}...{inv.dev_max} W"
#
# and transposing those two costs ZERO BYTES, changes nothing any check can
# observe, and tells the operator a range backwards. No predicate is involved,
# so census A's symmetry test has nothing to test.
#
# THE TEST: inside one MESSAGE-sink f-string, take every pair of interpolated
# expressions, blank every attribute NAME and every string CONSTANT, and compare.
# If the blanked forms are EQUAL while the originals DIFFER, the two expressions
# are the same shape with one leaf changed - `inv.dev_min` / `inv.dev_max`,
# `site_dev_bound("dev_min")` / `site_dev_bound("dev_max")` - which is exactly
# the pair a transposition swaps. Same direction as census A: it under-reports
# (two values of the same KIND spelled differently are missed) rather than
# inventing members.


class _Blank(ast.NodeTransformer):
    def visit_Attribute(self, node):
        self.generic_visit(node)
        node.attr = "<leaf>"
        return node

    def visit_Constant(self, node):
        if isinstance(node.value, str):
            return ast.Constant(value="<leaf>")
        return node


def _blanked(expr):
    return ast.dump(_Blank().visit(ast.parse(ast.unparse(expr))))


def scan_caller():
    rows = []
    for path in sorted(APP.glob("*.py")):
        tree = ast.parse(path.read_text())
        for fn in ast.walk(tree):
            if not isinstance(fn, (ast.FunctionDef, ast.AsyncFunctionDef)):
                continue
            for node in ast.walk(fn):
                if not isinstance(node, ast.JoinedStr):
                    continue
                # ⚠️ MESSAGE SINKS ONLY, AND THE FIRST VERSION OF THIS
                # CENSUS SAID "anything but a RETURN". That let in
                # `p1.py`'s `self.url = f"http://{self.host}:{self.port}…"` -
                # an f-string that is DATA, used to make a request, where a
                # transposition is an ordinary behavioural change every check
                # can see. Same distinction as census A's RETURN bucket: a
                # formatted string is only in this family when its only
                # consumer is a person.
                if _sink_of(fn, node) != "MESSAGE":
                    continue
                exprs = [v.value for v in node.values
                         if isinstance(v, ast.FormattedValue)]
                for a, b in itertools.combinations(exprs, 2):
                    ua, ub = ast.unparse(a), ast.unparse(b)
                    if ua == ub:
                        continue
                    if _blanked(a) == _blanked(b):
                        rows.append((path.name, node.lineno, fn.name, ua, ub))
    # de-duplicate: the same sentence reached through several enclosing defs
    return sorted(set(rows), key=lambda r: (r[0], r[1], r[3]))


# ⚠️ THE UNIT OF COVERAGE IS THE CALL SITE, NOT THE FUNCTION - which is the
# whole lesson of SAFETY-38 round 3. `SignWatchdog.evaluate()` had its operand
# order pinned and the defect still shipped, because the pin was on the call two
# lines below the one a one-inverter roster actually reaches. So a family member
# is reported WITH EVERY PLACE IT IS CALLED.
def call_sites(name):
    out = []
    for path in sorted(APP.glob("*.py")):
        tree = ast.parse(path.read_text())
        for node in ast.walk(tree):
            if not isinstance(node, ast.Call):
                continue
            called = (node.func.attr if isinstance(node.func, ast.Attribute)
                      else node.func.id if isinstance(node.func, ast.Name)
                      else "")
            if called == name:
                # ⚠️ THE RECEIVER IS PRINTED, because matching on the bare
                # attribute name conflates methods: `evaluate` is the name of
                # SAFETY-02's, SAFETY-18's and SAFETY-31's watchdogs alike, and
                # a list of four lines with no receiver tells a reader nothing
                # about which detector they are looking at. `inv.sign_wd` and
                # `self.freeze_wd` are the disambiguation, and they are the
                # caller's own words.
                out.append(f"{path.name}:{node.lineno} "
                           f"`{ast.unparse(node.func)}`")
    return out


def main():
    quiet = "--quiet" in sys.argv
    rounds = build_fixpoint()
    rows = scan()
    fam = [r for r in rows if r["symmetric"] and "MESSAGE" in r["sinks"]
           and "RETURN" not in r["sinks"]]
    valued = [r for r in rows if r["symmetric"] and "RETURN" in r["sinks"]]
    print("SYMMETRIC-PREDICATE FAMILY - argument order load-bearing for a "
          "MESSAGE only")
    print("=" * 78)
    print(f"fixpoint reached after {rounds} round(s); "
          f"{sum(len(v) for v in _SYM.values())} interchangeable positional "
          f"and {sum(len(v) for v in _SYM_KW.values())} keyword pair(s) "
          f"across {len(set(_SYM) | set(_SYM_KW))} function(s) in app/")
    print(f"functions formatting >=2 of their own parameters into one "
          f"sentence: {len(rows)} pair(s)")
    print(f"of those, pairs whose SWAP LEAVES EVERY DECISION IDENTICAL: "
          f"{len(fam) + len(valued)}")
    print(f"  ...of which the sentence is RETURNED (so the swap is a VALUE "
          f"change an ordinary check can see): {len(valued)}")
    print(f"  ...and the sentence goes ONLY TO AN OPERATOR - THE FAMILY: "
          f"{len(fam)}")
    print()
    print("THE FAMILY - a swap here is invisible to every behavioural check "
          "BY CONSTRUCTION:")
    for r in sorted(fam, key=lambda r: (r["module"], r["line"])):
        kind = ("POSITIONAL - a transposition at a call site costs ZERO BYTES"
                if r["positional"] else
                "KEYWORD-ONLY at the definition - a transposition has to be "
                "written out by name, which is not free and not silent")
        print(f"  {r['module']}:{r['line']}  {r['func']}()   pair "
              f"({r['a']}, {r['b']})")
        print(f"      sentence at line(s): "
              f"{', '.join(f'{l} -> {s}' for l, s in r['msg_lines'])}")
        print(f"      {kind}")
        sites = call_sites(r["func"])
        print(f"      CALL SITES ({len(sites)}) - each one is a separate pin, "
              f"and the receiver says which detector it is:")
        for site in sites or ["(none inside app/)"]:
            print(f"        {site}")
    print()
    print("SYMMETRIC, BUT THE SENTENCE IS THE RETURN VALUE - READ THIS BUCKET "
          "BY HAND.")
    print("A returned string is sometimes a KEY (a swap renames a store key or "
          "an MQTT topic,")
    print("which is behaviour every ordinary check can see - MAINT-05 is the "
          "card about it)")
    print("and sometimes a SENTENCE (a swap is message-only after all, and it "
          "IS this family):")
    for r in sorted(valued, key=lambda r: (r["module"], r["line"])):
        print(f"  {r['module']}:{r['line']}  {r['func']}()   pair "
              f"({r['a']}, {r['b']})")
    print()
    print("=" * 78)
    print("CENSUS B - SENTENCES THE CALLER BUILDS, interpolating two values of "
          "the SAME SHAPE")
    print("A transposition inside one of these is ZERO BYTES and no predicate "
          "is involved at")
    print("all, so nothing behavioural can ever see it. Same rule: assert the "
          "ORDERED PAIR.")
    cb = scan_caller()
    print(f"candidates: {len(cb)}")
    for mod, line, fn, ua, ub in cb:
        print(f"  {mod}:{line}  in {fn}()")
        print(f"      {ua}   <->   {ub}")
    print()
    print("WHAT THIS CANNOT SEE - the shape of each gap, named rather than "
          "rounded off:")
    print("  1. SEMANTIC symmetry no canonicaliser above spells. The list of "
          "commutative")
    print("     forms is finite and hand-written; a function symmetric only "
          "through a form")
    print("     not on it is REJECTED and never appears. This under-reports.")
    print("  2. In CENSUS B, two interpolated values of the same KIND that "
          "are not the same")
    print("     SHAPE - `a.lo` beside `hi`, or one of them behind a local "
          "variable. Only")
    print("     one-leaf-apart expressions are paired. `%s`-style logging "
          "(`_LOG.error('%s")
    print("     ... %s', a, b)`) and non-f-string payloads are not scanned at "
          "all.")
    print("  3. Anything outside `app/` - `sim/`, `tools/`, the firmware, the "
          "witness.")
    print("  4. WHETHER each call site is pinned. The sites are listed above; "
          "whether a")
    print("     check asserts the ORDERED PAIR at each one is a fact about "
          "`test_*.py` and is")
    print("     recorded in DECISIONS/SAFETY-38-site-standing-pin.md section "
          "7. Call sites")
    print("     outside `app/` are not listed either, and the site list matches "
          "on the bare")
    print("     METHOD NAME - three different watchdogs are all called "
          "`evaluate`, so the")
    print("     receiver is printed beside each one and a reader has to pick.")
    print("  5. ⚠️ IT CAN ALSO OVER-REPORT, IN ONE EXACT WAY. The sentence is "
          "ERASED before")
    print("     the comparison, so a swap that would make the FORMATTING "
          "ITSELF fail - a")
    print("     `DAY_NAMES[start]` where `start` is 1320 - is scored as "
          "'changes no decision'.")
    print("     `_raise_gap`'s (start, weekday) and (end, weekday) pairs are "
          "that: mixed")
    print("     types, and a real transposition there raises IndexError rather "
          "than lying")
    print("     quietly. The hazard is a SAME-TYPED pair; read the list with "
          "that in hand.")


if __name__ == "__main__":
    main()