Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432#!/usr/bin/env python3# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""Four-way native-sol conformance check."""
from __future__ import annotations
import astimport importlibimport inspectimport sysimport tomllibfrom collections.abc import Callable, Iterablefrom dataclasses import dataclassfrom functools import lru_cachefrom pathlib import Pathfrom typing import Any
from flask import Flask
import solstone.convey.reasons as reasonsfrom solstone.apps.activities.routes import activities_bpfrom solstone.apps.awareness.routes import awareness_bpfrom solstone.apps.body.routes import body_bpfrom solstone.apps.curation.routes import curation_bpfrom solstone.apps.entities.routes import entities_bpfrom solstone.apps.network.routes import network_bpfrom solstone.apps.settings.routes import settings_bpfrom solstone.apps.sol.routes import sol_bpfrom solstone.apps.speakers.routes import speakers_bpfrom solstone.apps.support.routes import support_bpfrom solstone.apps.thinking.routes import thinking_bpfrom solstone.apps.transcripts.routes import transcripts_bpfrom solstone.convey.chat import chat_bpfrom solstone.convey.contract.assemble import build_document, rule_to_openapi_pathfrom solstone.convey.health import bp as health_bpfrom solstone.convey.ledger import bp as ledger_bpfrom solstone.convey.profile import bp as profile_bpfrom solstone.convey.profile import profiles_bpfrom solstone.convey.reasons import Reasonfrom solstone.convey.root import bp as root_bp
try: from scripts.build_native_sol_inventory import ( AuthorityEntry, discover, is_private_app_authority, )except ModuleNotFoundError: # pragma: no cover - direct script execution path. from build_native_sol_inventory import ( # type: ignore[no-redef] AuthorityEntry, discover, is_private_app_authority, )
REPO_ROOT = Path(__file__).resolve().parents[1]import_bp = importlib.import_module("solstone.apps.import.routes").import_bpREASON_CODES_BY_NAME = { name: value.code for name, value in vars(reasons).items() if isinstance(value, Reason)}
@dataclass(frozen=True)class ContractOperation: operation_id: str method: str route: str reason_codes: frozenset[str]
@dataclass(frozen=True)class RawAuthorityEntry: authority: Path raw: dict[str, Any]
def check_conformance( *, root: Path = REPO_ROOT, document: dict[str, Any] | None = None, authorities: list[AuthorityEntry] | None = None, route_map: dict[tuple[str, str], Callable[..., Any]] | None = None,) -> list[str]: root = root.resolve() document = document if document is not None else build_document() authorities = authorities if authorities is not None else discover(root) route_map = route_map if route_map is not None else collect_flask_routes()
errors: list[str] = [] raw_authority_by_operation = load_raw_authority_entries(root) contract_by_operation = collect_contract_operations(document)
if not authorities: errors.append("native sol conformance discovered zero authorities") if not route_map: errors.append("native sol conformance discovered zero Flask routes") if not contract_by_operation: errors.append("native sol conformance discovered zero OpenAPI operations") if not any(authority.entry_type == "top-level-import" for authority in authorities): errors.append("native sol conformance missing top-level import authority") if not any(authority.entry_type == "top-level-link" for authority in authorities): errors.append("native sol conformance missing top-level link authority") if not any(authority.entry_type == "top-level-notify" for authority in authorities): errors.append("native sol conformance missing top-level notify authority")
for authority in sorted(authorities, key=lambda entry: entry.operation_id): raw_authority = raw_authority_by_operation.get(authority.operation_id) if authority.entry_type == "http": errors.extend(check_http_entry(authority, contract_by_operation, route_map)) elif authority.entry_type in {"moved-stub", "local"}: errors.extend(check_non_http_entry(authority, contract_by_operation)) elif authority.entry_type == "top-level-chat": errors.extend( check_top_level_backing_contracts( authority, raw_authority, contract_by_operation, route_map, "top-level-chat", "chat", ) ) elif authority.entry_type == "top-level-import": errors.extend( check_top_level_backing_contracts( authority, raw_authority, contract_by_operation, route_map, "top-level-import", "import", ) ) elif authority.entry_type == "top-level-notify": errors.extend(check_non_http_entry(authority, contract_by_operation)) elif authority.entry_type == "top-level-link": errors.extend(check_non_http_entry(authority, contract_by_operation)) else: errors.append( f"{authority.operation_id}: unsupported entry_type " f"{authority.entry_type!r}" ) return errors
def check_http_entry( authority: AuthorityEntry, contract_by_operation: dict[str, ContractOperation], route_map: dict[tuple[str, str], Callable[..., Any]],) -> list[str]: operation_id = authority.operation_id errors: list[str] = [] if authority.method is None or authority.route is None: errors.append(f"{operation_id}: HTTP authority must declare method and route") return errors if authority.contract_operation_id is None: errors.append( f"{operation_id}: HTTP authority must declare contract_operation_id" ) return errors
route_key = (authority.method, authority.route) view = route_map.get(route_key) if view is None: errors.append( f"{operation_id}: no Flask route for {authority.method} {authority.route}" )
contract = contract_by_operation.get(authority.contract_operation_id) if contract is None: errors.append( f"{operation_id}: no contract operation {authority.contract_operation_id}" ) return errors if contract.method != authority.method: errors.append( f"{operation_id}: contract method {contract.method!r} " f"!= authority {authority.method!r}" ) if contract.route != authority.route: errors.append( f"{operation_id}: contract route {contract.route!r} " f"!= authority {authority.route!r}" ) unknown_reasons = sorted(contract.reason_codes - set(REASON_CODES_BY_NAME.values())) if unknown_reasons: errors.append( f"{operation_id}: contract declares unknown reason codes {unknown_reasons!r}" )
if view is not None: route_reason_codes = route_error_reason_codes(view) if route_reason_codes != contract.reason_codes: errors.append( f"{operation_id}: route reason codes {sorted(route_reason_codes)!r} " f"!= contract {sorted(contract.reason_codes)!r}" ) return errors
def check_non_http_entry( authority: AuthorityEntry, contract_by_operation: dict[str, ContractOperation],) -> list[str]: errors: list[str] = [] operation_id = authority.operation_id if any( value is not None for value in ( authority.method, authority.route, authority.contract_operation_id, ) ): errors.append( f"{operation_id}: non-HTTP authority must not declare HTTP fields" ) if operation_id in contract_by_operation: errors.append(f"{operation_id}: non-HTTP authority must not have a contract") return errors
def check_top_level_backing_contracts( authority: AuthorityEntry, raw_authority: RawAuthorityEntry | None, contract_by_operation: dict[str, ContractOperation], route_map: dict[tuple[str, str], Callable[..., Any]], entry_type: str, label: str,) -> list[str]: errors = check_non_http_entry(authority, contract_by_operation) operation_id = authority.operation_id authority_ids = ( raw_authority.raw.get("backing_contract_operation_ids") if raw_authority is not None else None ) if not isinstance(authority_ids, list) or not authority_ids: errors.append(f"{operation_id}: {entry_type} must declare backing contracts") return errors for backing_id in authority_ids: if not isinstance(backing_id, str) or not backing_id: errors.append(f"{operation_id}: backing contract id must be a string") continue contract = contract_by_operation.get(backing_id) if contract is None: errors.append( f"{operation_id}: missing {label} backing contract {backing_id}" ) continue if (contract.method, contract.route) not in route_map: errors.append( f"{operation_id}: missing {label} backing route " f"{contract.method} {contract.route}" ) return errors
def collect_contract_operations( document: dict[str, Any],) -> dict[str, ContractOperation]: output: dict[str, ContractOperation] = {} for route, methods in document["paths"].items(): for raw_method, operation in methods.items(): operation_id = operation["operationId"] reason_codes = { reason_code for response in operation.get("responses", {}).values() for reason_code in response.get("x-reason-codes", []) } output[operation_id] = ContractOperation( operation_id=operation_id, method=raw_method.upper(), route=route, reason_codes=frozenset(reason_codes), ) return output
def register_native_blueprints(app: Flask) -> None: """Register blueprints needed by currently ported native authorities."""
for blueprint in ( activities_bp, support_bp, health_bp, chat_bp, root_bp, body_bp, curation_bp, entities_bp, sol_bp, awareness_bp, ledger_bp, profile_bp, profiles_bp, import_bp, transcripts_bp, network_bp, settings_bp, speakers_bp, thinking_bp, ): app.register_blueprint(blueprint)
def collect_flask_routes() -> dict[tuple[str, str], Callable[..., Any]]: app = Flask(__name__) register_native_blueprints(app) routes: dict[tuple[str, str], Callable[..., Any]] = {} for rule in app.url_map.iter_rules(): view = app.view_functions[rule.endpoint] route = rule_to_openapi_path(rule.rule) for method in sorted(rule.methods or ()): if method in {"HEAD", "OPTIONS"}: continue routes[(method, route)] = view return routes
def route_error_reason_codes(view: Callable[..., Any]) -> frozenset[str]: source = inspect.getsourcefile(view) if source is None: return frozenset() functions = module_functions(Path(source)) discovered = collect_function_reason_codes(view.__name__, functions, set()) return frozenset(discovered | annotated_route_reason_codes(view))
def annotated_route_reason_codes(view: Callable[..., Any]) -> set[str]: module = inspect.getmodule(view) annotations = getattr(module, "NATIVE_SOL_ROUTE_REASON_CODES", None) if not isinstance(annotations, dict): return set() values = annotations.get(view.__name__, set()) if not isinstance(values, (set, frozenset, list, tuple)): return set() return {str(value) for value in values}
@lru_cache(maxsize=None)def module_functions(source: Path) -> dict[str, ast.FunctionDef]: tree = ast.parse(source.read_text(encoding="utf-8"), filename=str(source)) return {node.name: node for node in tree.body if isinstance(node, ast.FunctionDef)}
def collect_function_reason_codes( function_name: str, functions: dict[str, ast.FunctionDef], seen: set[str],) -> set[str]: if function_name in seen: return set() seen.add(function_name) function = functions.get(function_name) if function is None: return set() reason_codes: set[str] = set() for node in ast.walk(function): if not isinstance(node, ast.Call): continue name = call_name(node.func) if name == "error_response" and node.args: reason_code = reason_code_from_ast(node.args[0]) if reason_code is not None: reason_codes.add(reason_code) elif ( isinstance(node.func, ast.Name) and node.func.id in functions and node.func.id != function_name ): reason_codes.update( collect_function_reason_codes(node.func.id, functions, seen) ) return reason_codes
def call_name(node: ast.expr) -> str | None: if isinstance(node, ast.Name): return node.id if isinstance(node, ast.Attribute): return node.attr return None
def reason_code_from_ast(node: ast.expr) -> str | None: if isinstance(node, ast.Name): return REASON_CODES_BY_NAME.get(node.id) if isinstance(node, ast.Attribute): return REASON_CODES_BY_NAME.get(node.attr) return None
def load_raw_authority_entries(root: Path) -> dict[str, RawAuthorityEntry]: output: dict[str, RawAuthorityEntry] = {} authority_paths = sorted( set((root / "solstone").glob("**/native/authority.toml")) | set((root / "solstone").glob("**/native/**/authority.toml")) ) for authority in authority_paths: if is_private_app_authority(authority, root): continue data = tomllib.loads(authority.read_text(encoding="utf-8")) for raw in data.get("entries", []): if not isinstance(raw, dict): continue operation_id = raw.get("operation_id") if isinstance(operation_id, str): output[operation_id] = RawAuthorityEntry(authority=authority, raw=raw) return output
def format_errors(errors: Iterable[str]) -> str: return "\n".join(f"- {error}" for error in errors)
def main() -> int: errors = check_conformance() if errors: print("native sol conformance failed:", file=sys.stderr) print(format_errors(errors), file=sys.stderr) return 1 print("native sol conformance ok") return 0
if __name__ == "__main__": raise SystemExit(main())