Something went wrong. Try again.
personal memory agent
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737#!/usr/bin/env python3# SPDX-License-Identifier: AGPL-3.0-only# Copyright (c) 2026 sol pbc
"""Check native-sol parity coverage against current HTTP authorities."""
from __future__ import annotations
import jsonimport subprocessimport sysfrom collections import Counterfrom pathlib import Pathfrom typing import Any
try: from scripts.build_native_sol_inventory import ( FINAL_HTTP_TOTAL, FINAL_STUB_COUNTS, FINAL_TOP_LEVEL_CHAT_TOTAL, FINAL_TOP_LEVEL_IMPORT_TOTAL, FINAL_TOP_LEVEL_LINK_TOTAL, FINAL_TOP_LEVEL_NOTIFY_TOTAL, REPO_ROOT, discover, )except ModuleNotFoundError: # pragma: no cover - direct script execution path. from build_native_sol_inventory import ( # type: ignore[no-redef] FINAL_HTTP_TOTAL, FINAL_STUB_COUNTS, FINAL_TOP_LEVEL_CHAT_TOTAL, FINAL_TOP_LEVEL_IMPORT_TOTAL, FINAL_TOP_LEVEL_LINK_TOTAL, FINAL_TOP_LEVEL_NOTIFY_TOTAL, REPO_ROOT, discover, )
SCHEMA = "native-sol-applicability-v1"APPLICABILITY = REPO_ROOT / "core/fixtures/native-sol/applicability.json"PARITY_DIR = REPO_ROOT / "core/fixtures/native-sol/parity"RUST_MANIFEST = REPO_ROOT / "core/Cargo.toml"
def main() -> int: errors = check_coverage() if errors: print("native sol coverage failed:", file=sys.stderr) for error in errors: print(f"- {error}", file=sys.stderr) return 1 print("native sol coverage ok") return 0
def check_coverage(root: Path = REPO_ROOT) -> list[str]: entries = discover(root) required = { entry.operation_id for entry in entries if entry.surface == "sol-call" and entry.entry_type == "http" } required_stubs = { entry.operation_id for entry in entries if entry.surface == "sol-call" and entry.entry_type in {"moved-stub", "local"} } required_top_level_chat = { entry.operation_id for entry in entries if entry.surface == "sol-chat" and entry.entry_type == "top-level-chat" } required_top_level_import = { entry.operation_id for entry in entries if entry.surface == "sol-import" and entry.entry_type == "top-level-import" } required_top_level_notify = { entry.operation_id for entry in entries if entry.surface == "sol-notify" and entry.entry_type == "top-level-notify" } required_top_level_link = { entry.operation_id for entry in entries if entry.surface == "sol-link" and entry.entry_type == "top-level-link" } required_dispatch = ( required | required_stubs | required_top_level_chat | required_top_level_import | required_top_level_link | required_top_level_notify ) vectors = load_vectors(PARITY_DIR) resolved = resolve_vectors(PARITY_DIR, vectors) applicability, applicability_errors = load_applicability(APPLICABILITY)
errors: list[str] = [] errors.extend(applicability_errors) if len(required) != FINAL_HTTP_TOTAL: errors.append( f"current HTTP authority count {len(required)} != {FINAL_HTTP_TOTAL}" ) stub_counts = Counter( entry.entry_type for entry in entries if entry.surface == "sol-call" and entry.entry_type in FINAL_STUB_COUNTS ) for entry_type, expected in sorted(FINAL_STUB_COUNTS.items()): if stub_counts[entry_type] != expected: errors.append( f"current {entry_type} authority count {stub_counts[entry_type]} " f"!= {expected}" ) if len(required_top_level_chat) != FINAL_TOP_LEVEL_CHAT_TOTAL: errors.append( f"current top-level chat authority count {len(required_top_level_chat)} " f"!= {FINAL_TOP_LEVEL_CHAT_TOTAL}" ) if not applicability_errors: entries = applicability["entries"] keys = set(entries) http_count = applicability.get("http_count") if http_count != FINAL_HTTP_TOTAL: errors.append( f"applicability http_count {http_count!r} != {FINAL_HTTP_TOTAL}" ) if http_count != len(entries): errors.append( f"applicability http_count {http_count!r} != entries {len(entries)}" ) if http_count != len(required): errors.append( f"applicability http_count {http_count!r} != current HTTP " f"authority count {len(required)}" ) errors.extend(compare_sets("applicability keys", required, keys)) top_level_entries = applicability.get("top_level_entries", {}) if not isinstance(top_level_entries, dict): errors.append("applicability top_level_entries must be an object") top_level_entries = {} import_keys = set(top_level_entries) errors.extend( compare_sets( "top-level import applicability keys", required_top_level_import, import_keys, ) )
if len(required_top_level_import) != FINAL_TOP_LEVEL_IMPORT_TOTAL: errors.append( f"current top-level import authority count {len(required_top_level_import)} " f"!= {FINAL_TOP_LEVEL_IMPORT_TOTAL}" ) if len(required_top_level_notify) != FINAL_TOP_LEVEL_NOTIFY_TOTAL: errors.append( f"current top-level notify authority count {len(required_top_level_notify)} " f"!= {FINAL_TOP_LEVEL_NOTIFY_TOTAL}" ) if len(required_top_level_link) != FINAL_TOP_LEVEL_LINK_TOTAL: errors.append( f"current top-level link authority count {len(required_top_level_link)} " f"!= {FINAL_TOP_LEVEL_LINK_TOTAL}" ) if not required_dispatch: errors.append("native dispatch authority set is empty") resolved_operations = { str(item["operation_id"]) for item in resolved.values() if item.get("operation_id") is not None } missing_dispatch = sorted(required_dispatch - resolved_operations) if missing_dispatch: errors.append( "production aggregate dispatch missing required operations " f"{missing_dispatch!r}" )
buckets = collect_buckets(vectors, resolved, required, {"http"}, errors) for bucket_name in ("request_binding", "success", "failure"): errors.extend(compare_sets(bucket_name, required, buckets[bucket_name])) import_buckets = collect_buckets( vectors, resolved, required_top_level_import, {"top-level-import"}, errors, ) for bucket_name in ("request_binding", "success", "failure"): errors.extend( compare_sets( f"top-level import {bucket_name}", required_top_level_import, import_buckets[bucket_name], ) ) notify_buckets = collect_buckets( vectors, resolved, required_top_level_notify, {"top-level-notify"}, errors, ) for bucket_name in ("notification_binding", "success", "failure"): errors.extend( compare_sets( f"top-level notify {bucket_name}", required_top_level_notify, notify_buckets[bucket_name], ) ) link_buckets = collect_buckets( vectors, resolved, required_top_level_link, {"top-level-link"}, errors, ) for bucket_name in ("success", "failure"): errors.extend( compare_sets( f"top-level link {bucket_name}", required_top_level_link, link_buckets[bucket_name], ) )
if not applicability_errors: errors.extend( check_applicability_requirements( applicability["entries"], vectors, resolved, buckets ) ) errors.extend( check_top_level_import_cases( applicability.get("top_level_entries", {}), vectors, resolved, ) ) return errors
def load_applicability(path: Path) -> tuple[dict[str, Any], list[str]]: try: payload = json.loads(path.read_text(encoding="utf-8")) except FileNotFoundError: return {}, [f"{path.relative_to(REPO_ROOT)} is missing"] except json.JSONDecodeError as error: return {}, [f"{path.relative_to(REPO_ROOT)} is malformed JSON: {error}"] errors: list[str] = [] if payload.get("schema") != SCHEMA: errors.append(f"applicability schema must be {SCHEMA!r}") entries = payload.get("entries") if not isinstance(entries, dict): errors.append("applicability entries must be an object") else: for operation_id, entry in entries.items(): if not isinstance(operation_id, str) or not operation_id: errors.append("applicability operation ids must be non-empty strings") if not isinstance(entry, dict): errors.append(f"{operation_id}: applicability entry must be an object") continue for key in ( "path", "group", "pagination", "mutation", "upload", "env_default", "confirmation", "consent", "dry_run", "multi_request", ): if key not in entry: errors.append(f"{operation_id}: missing applicability field {key}") top_level_entries = payload.get("top_level_entries", {}) if top_level_entries is not None and not isinstance(top_level_entries, dict): errors.append("applicability top_level_entries must be an object") return payload, errors
def load_vectors(directory: Path) -> dict[str, dict[str, Any]]: vectors: dict[str, dict[str, Any]] = {} for path in sorted(directory.glob("*.jsonl")): for line_number, line in enumerate( path.read_text(encoding="utf-8").splitlines(), 1 ): if not line.strip(): continue vector = json.loads(line) vector_id = str(vector.get("id") or "") if not vector_id: raise ValueError( f"{path.relative_to(REPO_ROOT)}:{line_number}: missing id" ) if vector_id in vectors: raise ValueError( f"{path.relative_to(REPO_ROOT)}:{line_number}: duplicate id {vector_id}" ) vector["_fixture"] = str(path.relative_to(REPO_ROOT)) vectors[vector_id] = vector return vectors
def resolve_vectors( directory: Path, vectors: dict[str, dict[str, Any]]) -> dict[str, dict[str, Any]]: paths = sorted(directory.glob("*.jsonl")) completed = subprocess.run( [ "cargo", "run", "--quiet", "--manifest-path", str(RUST_MANIFEST), "-p", "solstone-core-sol-client-cli", "--bin", "resolve_parity_leaves", "--", *(str(path) for path in paths), ], check=False, cwd=REPO_ROOT, text=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE, ) if completed.returncode != 0: raise RuntimeError( "production parity leaf resolver failed:\n" + completed.stderr ) resolved: dict[str, dict[str, Any]] = {} for line in completed.stdout.splitlines(): if not line.strip(): continue item = json.loads(line) vector_id = str(item["id"]) if vector_id not in vectors: raise RuntimeError(f"resolver returned unknown vector id {vector_id}") resolved[vector_id] = item missing = sorted(set(vectors) - set(resolved)) if missing: raise RuntimeError(f"resolver omitted vectors {missing!r}") return resolved
def collect_buckets( vectors: dict[str, dict[str, Any]], resolved: dict[str, dict[str, Any]], required: set[str], entry_types: set[str], errors: list[str],) -> dict[str, set[str]]: buckets: dict[str, set[str]] = { "request_binding": set(), "notification_binding": set(), "success": set(), "failure": set(), } for vector_id, vector in vectors.items(): operation_id = resolved[vector_id].get("operation_id") entry_type = resolved[vector_id].get("entry_type") if vector.get("surface", "sol-call") == "sol-call" and operation_id is None: errors.append( f"{vector_id}: argv did not resolve through production dispatch" ) continue if entry_type not in entry_types or operation_id not in required: continue expected = vector.get("expected") or {} requests = expected.get("requests") if isinstance(expected, dict) else None notifications = ( expected.get("notifications") if isinstance(expected, dict) else None ) if pins_request_shape(requests): buckets["request_binding"].add(operation_id) if pins_notification_shape(notifications): buckets["notification_binding"].add(operation_id) if is_failure_vector(vector, expected): buckets["failure"].add(operation_id) elif is_success_vector( expected, vector.get("transport", {}).get("requests", []) ): buckets["success"].add(operation_id) return buckets
def pins_notification_shape(notifications: Any) -> bool: return isinstance(notifications, list) and any( isinstance(item, str) and item.endswith("\n") for item in notifications )
def pins_request_shape(requests: Any) -> bool: if not isinstance(requests, list): return False return any( isinstance(request, dict) and isinstance(request.get("method"), str) and isinstance(request.get("path"), str) for request in requests )
def is_failure_vector(vector: dict[str, Any], expected: Any) -> bool: if isinstance(expected, dict) and expected.get("exit") != 0: return True for request in vector.get("transport", {}).get("requests", []): if not isinstance(request, dict): continue if "fault" in request: return True status = ( request.get("response", {}).get("status") if isinstance(request.get("response"), dict) else None ) if status is not None and not (200 <= int(status) <= 299): return True return False
def is_success_vector(expected: Any, transport_requests: Any) -> bool: if not isinstance(expected, dict) or expected.get("exit") != 0: return False if not isinstance(transport_requests, list): return False for request in transport_requests: if not isinstance(request, dict) or "fault" in request: return False response = request.get("response") if not isinstance(response, dict): return False status = int(response.get("status", 200)) if not (200 <= status <= 299): return False return True
def compare_sets(label: str, required: set[str], actual: set[str]) -> list[str]: errors: list[str] = [] missing = sorted(required - actual) extra = sorted(actual - required) if missing: errors.append(f"{label} missing {missing!r}") if extra: errors.append(f"{label} extra {extra!r}") return errors
def check_top_level_import_cases( entries: Any, vectors: dict[str, dict[str, Any]], resolved: dict[str, dict[str, Any]],) -> list[str]: if not isinstance(entries, dict): return [] errors: list[str] = [] for operation_id, entry in sorted(entries.items()): if not isinstance(entry, dict): errors.append( f"{operation_id}: top-level applicability entry must be an object" ) continue case_ids = entry.get("case_ids", {}) if not isinstance(case_ids, dict): errors.append(f"{operation_id}: top-level case_ids must be an object") continue for case_name, ids in sorted(case_ids.items()): if not isinstance(ids, list) or not ids: errors.append( f"{operation_id}: top-level case {case_name} must be non-empty" ) continue for vector_id in ids: if vector_id not in vectors: errors.append( f"{operation_id}: top-level case {case_name} unknown vector {vector_id!r}" ) continue mapped = resolved[vector_id].get("operation_id") if mapped != operation_id: errors.append( f"{operation_id}: top-level case {case_name} vector " f"{vector_id!r} maps to {mapped!r}" ) return errors
def check_applicability_requirements( entries: dict[str, dict[str, Any]], vectors: dict[str, dict[str, Any]], resolved: dict[str, dict[str, Any]], buckets: dict[str, set[str]],) -> list[str]: errors: list[str] = [] for operation_id, entry in sorted(entries.items()): for dimension in ( "pagination", "upload", "env_default", "dry_run", "multi_request", ): payload = entry.get(dimension) if not isinstance(payload, dict) or not payload.get("enabled"): continue errors.extend( check_named_cases(operation_id, dimension, payload, vectors, resolved) ) if bool(entry.get("upload", {}).get("enabled")): upload = entry["upload"] operation_vectors = mapped_vectors(vectors, resolved, operation_id) if not any( "UPLOAD" in expected_request_methods(vector) for vector in operation_vectors ): errors.append(f"{operation_id}: upload=true but no UPLOAD vector") errors.extend( check_required_case_names( operation_id, "upload", upload, ["missing_file", "unreadable_file", "rejection_before_mutation"], ) ) if upload.get("file_count") == "multi" and not any( has_later_upload_failure(vector) for vector in operation_vectors ): errors.append( f"{operation_id}: upload multi has no later-file-failure vector" ) if bool(entry.get("multi_request", {}).get("enabled")): multi_request = entry["multi_request"] operation_vectors = mapped_vectors(vectors, resolved, operation_id) if not any( expected_request_count(vector) > 1 for vector in operation_vectors ): errors.append( f"{operation_id}: multi_request=true but no multi-request vector" ) if multi_request_can_exceed_one( multi_request.get("request_count") ) and not any( has_later_boundary_failure(vector) for vector in operation_vectors ): errors.append( f"{operation_id}: multi_request has no later-boundary-failure vector" ) if bool(entry.get("env_default", {}).get("enabled")): env_default = entry["env_default"] mode = env_default.get("mode", "required") if mode in {"valid_absent", "valid-absent"}: required_cases = ["explicit", "valid_absent"] elif mode == "required": required_cases = ["explicit", "absent"] elif mode in {"required_with_invalid", "required-with-invalid"}: required_cases = ["explicit", "absent", "invalid"] else: errors.append( f"{operation_id}: env_default.mode {mode!r} is unsupported" ) required_cases = [] errors.extend( check_required_case_names( operation_id, "env_default", env_default, required_cases ) ) if bool(entry.get("dry_run", {}).get("enabled")): if operation_id not in buckets["success"]: errors.append(f"{operation_id}: dry_run=true but no success coverage") if operation_id not in buckets["failure"]: errors.append(f"{operation_id}: dry_run=true but no failure coverage") return errors
def check_required_case_names( operation_id: str, dimension: str, payload: dict[str, Any], required_cases: list[str],) -> list[str]: case_ids = payload.get("case_ids", {}) if not isinstance(case_ids, dict): return [] errors = [] for case_name in required_cases: ids = case_ids.get(case_name) if not isinstance(ids, list) or not ids: errors.append( f"{operation_id}: {dimension} missing required case {case_name}" ) return errors
def check_named_cases( operation_id: str, dimension: str, payload: dict[str, Any], vectors: dict[str, dict[str, Any]], resolved: dict[str, dict[str, Any]],) -> list[str]: errors: list[str] = [] case_ids = payload.get("case_ids", {}) if not isinstance(case_ids, dict): errors.append(f"{operation_id}: {dimension}.case_ids must be an object") return errors for case_name, ids in sorted(case_ids.items()): if not isinstance(ids, list) or not ids: errors.append( f"{operation_id}: {dimension}.case_ids.{case_name} must be non-empty" ) continue for vector_id in ids: if vector_id not in vectors: errors.append( f"{operation_id}: {dimension}.case_ids.{case_name} unknown " f"vector {vector_id!r}" ) continue mapped = resolved[vector_id].get("operation_id") if mapped != operation_id: errors.append( f"{operation_id}: {dimension}.case_ids.{case_name} vector " f"{vector_id!r} maps to {mapped!r}" ) return errors
def mapped_vectors( vectors: dict[str, dict[str, Any]], resolved: dict[str, dict[str, Any]], operation_id: str,) -> list[dict[str, Any]]: return [ vector for vector_id, vector in vectors.items() if resolved[vector_id].get("operation_id") == operation_id ]
def expected_request_methods(vector: dict[str, Any]) -> set[str]: requests = vector.get("expected", {}).get("requests", []) methods: set[str] = set() for request in requests if isinstance(requests, list) else []: if isinstance(request, dict) and isinstance(request.get("method"), str): methods.add(request["method"]) return methods
def expected_request_count(vector: dict[str, Any]) -> int: requests = vector.get("expected", {}).get("requests", []) return len(requests) if isinstance(requests, list) else 0
def multi_request_can_exceed_one(value: Any) -> bool: if isinstance(value, int): return value > 1 if not isinstance(value, str): return False if value == "single": return False if ".." not in value: try: return int(value) > 1 except ValueError: return True _minimum, maximum = value.split("..", 1) if maximum == "n": return True try: return int(maximum) > 1 except ValueError: return True
def has_later_boundary_failure(vector: dict[str, Any]) -> bool: requests = transport_requests(vector) for index, request in enumerate(requests): if index == 0 or not request_failed(request): continue if all(request_succeeded(previous) for previous in requests[:index]): return True return False
def has_later_upload_failure(vector: dict[str, Any]) -> bool: requests = transport_requests(vector) upload_indices = [ index for index, request in enumerate(requests) if isinstance(request, dict) and request.get("method") == "UPLOAD" ] for upload_position, request_index in enumerate(upload_indices): if upload_position == 0 or not request_failed(requests[request_index]): continue prior_uploads = (requests[index] for index in upload_indices[:upload_position]) if all(request_succeeded(request) for request in prior_uploads): return True return False
def transport_requests(vector: dict[str, Any]) -> list[dict[str, Any]]: requests = vector.get("transport", {}).get("requests", []) return [request for request in requests if isinstance(request, dict)]
def request_failed(request: dict[str, Any]) -> bool: if "fault" in request: return True status = response_status(request) return status is not None and not (200 <= status <= 299)
def request_succeeded(request: dict[str, Any]) -> bool: if "fault" in request: return False status = response_status(request) return status is not None and 200 <= status <= 299
def response_status(request: dict[str, Any]) -> int | None: response = request.get("response") if not isinstance(response, dict): return None return int(response.get("status", 200))
if __name__ == "__main__": raise SystemExit(main())