"""Static tripwire for analytics methods orphaned from production callers.
This would have caught the Device Connected regression in 02524b7c0b: its
qualified Flutter call sites fell from two to one. It also rejects the dead
macOS shape removed with this checker: a production-called empty
reportAllSettingsIfNeeded() plus its unreachable collectAllSettings() helper.
This is deliberately lexical, not behavioral coverage. It recognizes the
repository's three analytics access boundaries, ignores comments and strings,
and ratchets current call-site counts. It does not prove that a call executes.
"""
from __future__ import annotations
import json
import sys
from dataclasses import dataclass
from pathlib import Path
ROOT = Path(__file__).resolve().parents[2]
BASELINE = ROOT / ".github/scripts/analytics_reachability_baseline.json"
@dataclass(frozen=True)
class Token:
value: str
kind: str = "punct"
@dataclass(frozen=True)
class Method:
name: str
public: bool
body: tuple[Token, ...]
def lex(source: str) -> list[Token]:
"""Tokenize enough syntax to distinguish code from comments and strings."""
tokens: list[Token] = []
i = 0
while i < len(source):
char = source[i]
if char.isspace():
i += 1
continue
if source.startswith("//", i):
i = source.find("\n", i)
if i < 0:
break
continue
if source.startswith("/*", i):
depth = 1
i += 2
while i < len(source) and depth:
if source.startswith("/*", i):
depth += 1
i += 2
elif source.startswith("*/", i):
depth -= 1
i += 2
else:
i += 1
continue
if char in "'\"`":
quote = char
triple = source.startswith(char * 3, i)
delimiter = char * (3 if triple else 1)
i += len(delimiter)
value: list[str] = []
while i < len(source):
if source.startswith(delimiter, i):
i += len(delimiter)
break
if source[i] == "\\" and not triple and i + 1 < len(source):
value.append(source[i + 1])
i += 2
else:
value.append(source[i])
i += 1
tokens.append(Token("".join(value), "string"))
continue
if char.isalpha() or char in "_$":
end = i + 1
while end < len(source) and (source[end].isalnum() or source[end] in "_$"):
end += 1
tokens.append(Token(source[i:end], "ident"))
i = end
continue
if source.startswith("=>", i):
tokens.append(Token("=>"))
i += 2
continue
tokens.append(Token(char))
i += 1
return tokens
def matching(tokens: list[Token], start: int, opening: str, closing: str) -> int:
depth = 0
for index in range(start, len(tokens)):
if tokens[index].value == opening:
depth += 1
elif tokens[index].value == closing:
depth -= 1
if depth == 0:
return index
raise ValueError(f"unclosed {opening}")
def statement_end(tokens: list[Token], start: int) -> int:
"""Find a semicolon outside nested closures/collections."""
depths = {"(": 0, "[": 0, "{": 0}
pairs = {")": "(", "]": "[", "}": "{"}
for index in range(start, len(tokens)):
value = tokens[index].value
if value in depths:
depths[value] += 1
elif value in pairs:
depths[pairs[value]] -= 1
elif value == ";" and not any(depths.values()):
return index
raise ValueError("unterminated statement")
def class_body(tokens: list[Token], name: str) -> tuple[int, int]:
for index in range(len(tokens) - 2):
if tokens[index].value == "class" and tokens[index + 1].value == name:
brace = next(i for i in range(index + 2, len(tokens)) if tokens[i].value == "{")
return brace, matching(tokens, brace, "{", "}")
raise ValueError(f"class {name} not found")
def dart_methods(source: str) -> list[Method]:
tokens = lex(source)
start, end = class_body(tokens, "AnalyticsManager")
methods: list[Method] = []
depth = 1
index = start + 1
while index < end:
value = tokens[index].value
if value == "{":
depth += 1
elif value == "}":
depth -= 1
elif (
depth == 1
and tokens[index].kind == "ident"
and index + 1 < end
and tokens[index + 1].value == "("
and value != "AnalyticsManager"
):
close = matching(tokens, index + 1, "(", ")")
after = close + 1
while after < end and tokens[after].value in {"async", "sync", "*"}:
after += 1
if after < end and tokens[after].value in {"{", "=>"}:
if tokens[after].value == "{":
body_end = matching(tokens, after, "{", "}")
body = tuple(tokens[after + 1 : body_end])
else:
body_end = statement_end(tokens, after + 1)
body = tuple(tokens[after + 1 : body_end])
methods.append(Method(value, not value.startswith("_"), body))
index = body_end
index += 1
return methods
def swift_methods(source: str) -> list[Method]:
tokens = lex(source)
start, end = class_body(tokens, "AnalyticsManager")
methods: list[Method] = []
depth = 1
index = start + 1
while index < end:
value = tokens[index].value
if value == "{":
depth += 1
elif value == "}":
depth -= 1
elif depth == 1 and value == "func" and index + 1 < end:
name = tokens[index + 1].value
modifier_start = index - 1
while modifier_start > start and tokens[modifier_start].value not in {"}", "{", ";", "var", "let"}:
modifier_start -= 1
modifiers = {token.value for token in tokens[modifier_start + 1 : index]}
brace = next(i for i in range(index + 2, end) if tokens[i].value == "{")
body_end = matching(tokens, brace, "{", "}")
methods.append(
Method(
name, not modifiers.intersection({"private", "fileprivate"}), tuple(tokens[brace + 1 : body_end])
)
)
index = body_end
index += 1
return methods
def typescript_methods(source: str) -> list[Method]:
tokens = lex(source)
methods: list[Method] = []
for index in range(len(tokens) - 2):
if tokens[index].value != "export" or tokens[index + 1].value != "function":
continue
name = tokens[index + 2].value
opening = next(i for i in range(index + 3, len(tokens)) if tokens[i].value == "(")
closing = matching(tokens, opening, "(", ")")
brace = next(i for i in range(closing + 1, len(tokens)) if tokens[i].value == "{")
body_end = matching(tokens, brace, "{", "}")
methods.append(Method(name, True, tuple(tokens[brace + 1 : body_end])))
return methods
def sequence_call_counts(tokens: list[Token], prefixes: tuple[tuple[str, ...], ...]) -> dict[str, int]:
counts: dict[str, int] = {}
values = [token.value for token in tokens]
for index in range(len(values)):
for prefix in prefixes:
end = index + len(prefix)
if tuple(values[index:end]) != prefix or end + 2 >= len(values):
continue
if values[end] == "." and tokens[end + 1].kind == "ident" and values[end + 2] == "(":
name = values[end + 1]
counts[name] = counts.get(name, 0) + 1
return counts
def dart_call_counts(sources: list[str]) -> dict[str, int]:
counts: dict[str, int] = {}
prefixes = (
("AnalyticsManager", "(", ")"),
("AnalyticsManager",),
("PlatformManager", ".", "instance", ".", "analytics"),
)
for source in sources:
for name, count in sequence_call_counts(lex(source), prefixes).items():
counts[name] = counts.get(name, 0) + count
return counts
def swift_call_counts(sources: list[str]) -> dict[str, int]:
counts: dict[str, int] = {}
for source in sources:
for name, count in sequence_call_counts(lex(source), (("AnalyticsManager", ".", "shared"),)).items():
counts[name] = counts.get(name, 0) + count
return counts
def windows_imports(tokens: list[Token]) -> dict[str, str]:
aliases: dict[str, str] = {}
index = 0
while index < len(tokens):
if tokens[index].value != "import":
index += 1
continue
try:
opening = next(i for i in range(index + 1, len(tokens)) if tokens[i].value in {"{", ";"})
except StopIteration:
break
if tokens[opening].value != "{":
index = opening + 1
continue
closing = matching(tokens, opening, "{", "}")
if closing + 2 >= len(tokens) or tokens[closing + 1].value != "from":
index = closing + 1
continue
path = tokens[closing + 2]
if path.kind != "string" or path.value.rsplit("/", 1)[-1] not in {"analytics", "analytics.ts"}:
index = closing + 1
continue
cursor = opening + 1
while cursor < closing:
if tokens[cursor].kind != "ident":
cursor += 1
continue
imported = tokens[cursor].value
local = imported
if cursor + 2 < closing and tokens[cursor + 1].value == "as":
local = tokens[cursor + 2].value
cursor += 3
else:
cursor += 1
aliases[local] = imported
index = closing + 3
return aliases
def windows_call_counts(sources: list[str]) -> dict[str, int]:
counts: dict[str, int] = {}
for source in sources:
tokens = lex(source)
aliases = windows_imports(tokens)
for index in range(len(tokens) - 1):
imported = aliases.get(tokens[index].value)
if imported and tokens[index + 1].value == "(":
counts[imported] = counts.get(imported, 0) + 1
return counts
def is_test_seam(name: str) -> bool:
"""Return whether a method is a test-only observation hook, not an emitter.
Installing or forwarding to a scoped test capture is production-unreachable
by construction, so auditing it only forces one baseline entry per seam. The
emitters that call these hooks are still audited normally.
"""
return name.endswith("ForTests") or name.endswith("ForTesting")
def emitters(methods: list[Method]) -> list[Method]:
return [method for method in methods if not is_test_seam(method.name)]
def private_incoming(methods: list[Method]) -> dict[str, int]:
private_names = {method.name for method in methods if not method.public}
incoming = dict.fromkeys(private_names, 0)
for caller in methods:
for index in range(len(caller.body) - 1):
callee = caller.body[index].value
if callee in private_names and caller.name != callee and caller.body[index + 1].value == "(":
incoming[callee] += 1
return incoming
def audit_platform(
platform: str,
methods: list[Method],
calls: dict[str, int],
baseline: dict[str, dict[str, object]],
) -> list[str]:
errors: list[str] = []
by_name = {method.name: method for method in methods}
if len(by_name) != len(methods):
errors.append(f"{platform}: overloaded analytics methods are unsupported by this name-based tripwire")
return errors
orphans = set(baseline["public_orphans"].get(platform, []))
private_orphans = set(baseline["private_orphans"].get(platform, []))
minimums = baseline["multi_call_minimums"].get(platform, {})
for stale in sorted((orphans | private_orphans | set(minimums)) - set(by_name)):
errors.append(f"{platform}.{stale}: stale baseline entry; remove it")
incoming = private_incoming(methods)
for method in methods:
if not method.body:
errors.append(f"{platform}.{method.name}: empty analytics method")
if method.public:
count = calls.get(method.name, 0)
if count == 0:
if method.name not in orphans:
errors.append(f"{platform}.{method.name}: no production call site")
continue
if method.name in orphans:
errors.append(f"{platform}.{method.name}: now has a caller; remove it from public_orphans")
minimum = int(minimums.get(method.name, 1))
if count < minimum:
errors.append(f"{platform}.{method.name}: production call sites fell {minimum}->{count}")
elif count > minimum and count > 1:
errors.append(
f"{platform}.{method.name}: production call sites grew {minimum}->{count}; "
"raise multi_call_minimums to ratchet the gain"
)
else:
count = incoming.get(method.name, 0)
if count == 0 and method.name not in private_orphans:
errors.append(f"{platform}.{method.name}: unreachable private analytics helper")
elif count > 0 and method.name in private_orphans:
errors.append(f"{platform}.{method.name}: now reachable; remove it from private_orphans")
return errors
def load_platforms(root: Path) -> dict[str, tuple[list[Method], dict[str, int]]]:
dart_manager = root / "app/lib/utils/analytics/analytics_manager.dart"
swift_manager = root / "desktop/macos/Desktop/Sources/AnalyticsManager.swift"
windows_manager = root / "desktop/windows/src/renderer/src/lib/analytics.ts"
dart_sources = [
path.read_text(encoding="utf-8") for path in (root / "app/lib").rglob("*.dart") if path != dart_manager
]
swift_sources = [
path.read_text(encoding="utf-8")
for path in (root / "desktop/macos/Desktop/Sources").rglob("*.swift")
if path != swift_manager
]
windows_sources = [
path.read_text(encoding="utf-8")
for path in (root / "desktop/windows/src/renderer/src").rglob("*")
if path.suffix in {".ts", ".tsx"}
and ".test." not in path.name
and ".spec." not in path.name
and path != windows_manager
]
return {
"flutter": (emitters(dart_methods(dart_manager.read_text(encoding="utf-8"))), dart_call_counts(dart_sources)),
"macos": (emitters(swift_methods(swift_manager.read_text(encoding="utf-8"))), swift_call_counts(swift_sources)),
"windows": (
emitters(typescript_methods(windows_manager.read_text(encoding="utf-8"))),
windows_call_counts(windows_sources),
),
}
def suggested_baseline(platforms: dict[str, tuple[list[Method], dict[str, int]]]) -> dict[str, dict[str, object]]:
baseline: dict[str, dict[str, object]] = {
"public_orphans": {},
"private_orphans": {},
"multi_call_minimums": {},
}
for platform, (methods, calls) in platforms.items():
incoming = private_incoming(methods)
baseline["public_orphans"][platform] = sorted(
method.name for method in methods if method.public and calls.get(method.name, 0) == 0
)
baseline["private_orphans"][platform] = sorted(
method.name for method in methods if not method.public and incoming.get(method.name, 0) == 0
)
baseline["multi_call_minimums"][platform] = {
method.name: calls[method.name]
for method in sorted(methods, key=lambda item: item.name)
if method.public and calls.get(method.name, 0) > 1
}
return baseline
def main() -> int:
platforms = load_platforms(ROOT)
if "--print-baseline" in sys.argv:
print(json.dumps(suggested_baseline(platforms), indent=2, sort_keys=True))
return 0
baseline = json.loads(BASELINE.read_text(encoding="utf-8"))
errors = [
error
for platform, (methods, calls) in platforms.items()
for error in audit_platform(platform, methods, calls, baseline)
]
if errors:
print("analytics reachability static tripwire failed:", file=sys.stderr)
for error in errors:
print(f" - {error}", file=sys.stderr)
return 1
print("analytics reachability static tripwire passed")
return 0
if __name__ == "__main__":
raise SystemExit(main())