diff --git a/tools/check_loader_catalog_sync.py b/tools/check_loader_catalog_sync.py index e22ebf5d8..b99008747 100644 --- a/tools/check_loader_catalog_sync.py +++ b/tools/check_loader_catalog_sync.py @@ -1,11 +1,18 @@ #!/usr/bin/env python3 -"""Check sync between runtime loaders, model_specs packages, and model_manager_v2. +"""Check sync between runtime loaders, model_specs packages, model_manager_v2, +and the WebUI catalog. Schema correctness is owned by the typed model-spec validator. This script only checks cross-system drift: - registered loader families vs model_specs families - model_specs packages vs model_manager_v2 package output +- model_specs packages vs the packages the native package manager publishes +- webui/configs/models_catalog.json vs model_specs (family, download id, + install path, task/mode vocabulary) and vs the packages a catalog entry can + actually reach +- webui/configs/model_params.json parameter groups vs the request options the + owning spec documents See docs/maintainers/loader_and_catalog.md. """ @@ -24,14 +31,27 @@ CMAKE_PATH = REPO_ROOT / "CMakeLists.txt" REGISTRY_PATH = REPO_ROOT / "src" / "framework" / "runtime" / "registry.cpp" SPECS_DIR = REPO_ROOT / "model_specs" +CATALOG_PATH = REPO_ROOT / "webui" / "configs" / "models_catalog.json" +MODEL_PARAMS_PATH = REPO_ROOT / "webui" / "configs" / "model_params.json" +SESSION_PATH = REPO_ROOT / "src" / "framework" / "runtime" / "session.cpp" _LOADER_CALL_RE = re.compile(r"\bmake_([a-z0-9_]+)_loader(?:\s*\(\s*\))?") +_VALUE_COMPARE_RE = re.compile(r'value\s*==\s*"([a-z0-9_]+)"') BUNDLED_LOADERS_WITHOUT_SPEC = { "marblenet_vad", "silero_vad", } +# Bundled loaders ship inside the repository instead of a downloadable package, +# so their catalog entries carry no download id and point outside models/. +BUNDLED_CATALOG_PATH_PREFIX = "assets/framework/models/" + +# src/framework/package_manager/manager.cpp drops every package whose download +# kind is not a Hugging Face snapshot, so those packages are invisible to the +# server and to the WebUI install flow no matter what model_manager_v2 lists. +INSTALLABLE_DOWNLOAD_KIND = "huggingface_snapshot" + @dataclass(frozen=True) class SpecPackage: @@ -40,6 +60,43 @@ class SpecPackage: format: str target_directory: str default: bool + download_kind: str = "" + files: tuple[str, ...] = () + strip_prefix: str = "" + + @property + def installable(self) -> bool: + return self.download_kind == INSTALLABLE_DOWNLOAD_KIND + + @property + def install_paths(self) -> set[str]: + """Catalog paths that resolve to this package once it is installed. + + Either the install directory itself or one of the files the package + drops into it. Both spellings appear in models_catalog.json and both + are accepted by the loaders. + """ + root = normalize_catalog_path(f"models/{self.target_directory}") + paths = {root} + prefix = normalize_catalog_path(self.strip_prefix) + for remote in self.files: + relative = normalize_catalog_path(remote) + if prefix and relative.startswith(f"{prefix}/"): + relative = relative[len(prefix) + 1:] + if relative: + paths.add(normalize_catalog_path(f"{root}/{relative}")) + return paths + + +@dataclass(frozen=True) +class CatalogEntry: + index: int + id: str + family: str + path: str + task: str + mode: str + download_id: str def rel(path: Path) -> str: @@ -49,6 +106,27 @@ def rel(path: Path) -> str: return str(path) +def normalize_catalog_path(value: str) -> str: + text = value.strip().replace("\\", "/") + while text.startswith("./"): + text = text[2:] + text = re.sub(r"/+", "/", text) + return text.rstrip("/") + + +def parse_string_vocabulary(text: str, signature: str) -> set[str]: + """Collect the `value == "..."` literals a session.cpp parser accepts.""" + start = text.find(signature) + if start < 0: + raise ValueError(f"'{signature}' not found") + end = text.find("\n}\n", start) + body = text[start:] if end < 0 else text[start:end] + values = set(_VALUE_COMPARE_RE.findall(body)) + if not values: + raise ValueError(f"'{signature}' declares no accepted values") + return values + + def parse_loader_declarations(text: str, comment_prefix: str) -> tuple[set[str], set[str]]: active: set[str] = set() commented: set[str] = set() @@ -110,6 +188,11 @@ def load_spec_packages(specs_dir: Path) -> tuple[dict[str, dict[str, Any]], dict errors.append(f"{rel(path)}: duplicate spec family '{family}'") specs_by_family[family] = spec + spec_defaults = spec.get("package_defaults") + default_download = {} + if isinstance(spec_defaults, dict) and isinstance(spec_defaults.get("download"), dict): + default_download = spec_defaults["download"] + packages = spec.get("packages", []) if packages is None: packages = [] @@ -130,16 +213,72 @@ def load_spec_packages(specs_dir: Path) -> tuple[dict[str, dict[str, Any]], dict f"model_specs/{packages_by_id[package_id].family}.json" ) continue + download = package.get("download") + if not isinstance(download, dict): + download = {} + files = package.get("files") + if not isinstance(files, list): + files = [] packages_by_id[package_id] = SpecPackage( family=family, id=package_id, format=str(package.get("format") or ""), target_directory=str(package.get("target_directory") or ""), default=package.get("default") is True, + download_kind=str(download.get("kind") or default_download.get("kind") or ""), + files=tuple(str(item) for item in files if isinstance(item, str)), + strip_prefix=str(package.get("strip_prefix") or ""), ) return specs_by_family, packages_by_id, errors +def recommended_package_id(spec: dict[str, Any]) -> str: + ui = spec.get("ui") + if not isinstance(ui, dict): + return "" + value = ui.get("recommended_package") + return value if isinstance(value, str) else "" + + +def load_catalog(path: Path) -> tuple[list[CatalogEntry], list[str]]: + errors: list[str] = [] + try: + payload = load_json(path) + except json.JSONDecodeError as exc: + return [], [f"{rel(path)}: invalid JSON: {exc}"] + if not isinstance(payload, dict) or not isinstance(payload.get("models"), list): + return [], [f"{rel(path)}: top-level JSON must be an object with a 'models' list"] + + entries: list[CatalogEntry] = [] + seen_ids: set[str] = set() + for index, row in enumerate(payload["models"]): + if not isinstance(row, dict): + errors.append(f"{rel(path)}: models[{index}] must be an object") + continue + entry_id = row.get("id") + if not isinstance(entry_id, str) or not entry_id: + errors.append(f"{rel(path)}: models[{index}] missing id") + continue + if entry_id in seen_ids: + errors.append(f"{rel(path)}: duplicate catalog entry id '{entry_id}'") + continue + seen_ids.add(entry_id) + family = row.get("family") + if not isinstance(family, str) or not family: + errors.append(f"{rel(path)}: catalog entry '{entry_id}' missing family") + continue + entries.append(CatalogEntry( + index=index, + id=entry_id, + family=family, + path=str(row.get("path") or ""), + task=str(row.get("task") or ""), + mode=str(row.get("mode") or ""), + download_id=str(row.get("download_id") or ""), + )) + return entries, errors + + def load_manager_packages(specs_dir: Path) -> tuple[dict[str, Any], list[str]]: sys.path.insert(0, str(REPO_ROOT / "tools")) import model_manager_v2 # noqa: E402 @@ -210,6 +349,289 @@ def check_manager_sync(spec_packages: dict[str, SpecPackage], manager_packages: return errors, warnings +def check_native_manager_sync(spec_packages: dict[str, SpecPackage]) -> list[str]: + """Report packages model_manager_v2 lists but the native manager drops. + + tools/model_manager_v2.py flattens every declared package; the server keeps + only Hugging Face snapshots (src/framework/package_manager/manager.cpp), so + the two package counts are not interchangeable. + """ + warnings: list[str] = [] + for package in sorted(spec_packages.values(), key=lambda item: item.id): + if package.installable: + continue + warnings.append( + f"model_specs/{package.family}.json package '{package.id}' has download.kind=" + f"'{package.download_kind or 'missing'}' and is not published by the native " + f"package manager (server-side package list excludes it)" + ) + return warnings + + +def check_catalog_sync( + catalog_entries: list[CatalogEntry], + specs_by_family: dict[str, dict[str, Any]], + spec_packages: dict[str, SpecPackage], + task_kinds: set[str], + run_modes: set[str], + catalog_path: Path, +) -> tuple[list[str], list[str]]: + """Check webui/configs/models_catalog.json against model_specs. + + A catalog entry resolves to exactly one package through its download id. + The WebUI then offers every GGUF package that installs into the resolved + package's target directory (webui/native/src/lib/catalog.ts), so a target + directory no entry resolves into is unreachable from the UI. + """ + errors: list[str] = [] + warnings: list[str] = [] + name = rel(catalog_path) + + packages_by_family: dict[str, list[SpecPackage]] = {} + for package in spec_packages.values(): + packages_by_family.setdefault(package.family, []).append(package) + + reachable_ids: set[str] = set() + families_in_catalog: set[str] = set() + + for entry in catalog_entries: + if entry.task and entry.task not in task_kinds: + errors.append( + f"{name}: catalog entry '{entry.id}' task '{entry.task}' is not accepted by " + f"parse_voice_task_kind (expected one of {', '.join(sorted(task_kinds))})" + ) + if entry.mode and entry.mode not in run_modes: + errors.append( + f"{name}: catalog entry '{entry.id}' mode '{entry.mode}' is not accepted by " + f"parse_run_mode (expected one of {', '.join(sorted(run_modes))})" + ) + + if entry.family in BUNDLED_LOADERS_WITHOUT_SPEC: + if entry.download_id: + errors.append( + f"{name}: catalog entry '{entry.id}' is a bundled loader but names " + f"download_id '{entry.download_id}'" + ) + if not normalize_catalog_path(entry.path).startswith(BUNDLED_CATALOG_PATH_PREFIX): + errors.append( + f"{name}: catalog entry '{entry.id}' is bundled and must point inside " + f"{BUNDLED_CATALOG_PATH_PREFIX} (path '{entry.path}')" + ) + continue + + if entry.family not in specs_by_family: + errors.append( + f"{name}: catalog entry '{entry.id}' names family '{entry.family}' " + f"which has no model_specs/{entry.family}.json" + ) + continue + families_in_catalog.add(entry.family) + + family_packages = packages_by_family.get(entry.family, []) + recommended_id = recommended_package_id(specs_by_family[entry.family]) + resolved = spec_packages.get(entry.download_id) if entry.download_id else None + degraded = False + if not entry.download_id: + warnings.append( + f"{name}: catalog entry '{entry.id}' has no download_id; its install location " + f"cannot be checked against model_specs/{entry.family}.json" + ) + elif resolved is None: + errors.append( + f"{name}: catalog entry '{entry.id}' download_id '{entry.download_id}' is not a " + f"packages[].id in model_specs/{entry.family}.json" + ) + elif resolved.family != entry.family: + errors.append( + f"{name}: catalog entry '{entry.id}' download_id '{entry.download_id}' belongs to " + f"model_specs/{resolved.family}.json, not to family '{entry.family}'" + ) + resolved = None + if resolved is None and entry.download_id: + # Fall back to the family recommendation so a broken download id + # does not also suppress the path and reachability checks. The + # download id error above is the one to fix first. + fallback = spec_packages.get(recommended_id) + if fallback is not None and fallback.family == entry.family: + resolved = fallback + degraded = True + + if resolved is None: + continue + + if entry.path: + candidate = normalize_catalog_path(entry.path) + accepted = resolved.install_paths + source = ( + f"model_specs/{entry.family}.json recommended package '{resolved.id}'" + if degraded else f"package '{resolved.id}'" + ) + if candidate not in accepted: + lowered = {value.lower() for value in accepted} + if candidate.lower() in lowered: + warnings.append( + f"{name}: catalog entry '{entry.id}' path '{entry.path}' differs in case " + f"from the install location of {source}" + ) + else: + errors.append( + f"{name}: catalog entry '{entry.id}' path '{entry.path}' is neither the " + f"target_directory nor an installed file of {source} " + f"(expected models/{resolved.target_directory}[/])" + ) + + exposed = [ + package for package in family_packages + if package.target_directory == resolved.target_directory + and package.format == "gguf" and package.installable + ] + if not exposed: + warnings.append( + f"{name}: catalog entry '{entry.id}' resolves to package '{resolved.id}' " + f"(format={resolved.format or 'unknown'}), which the native model manager " + f"cannot install; the entry has no install choice" + ) + reachable_ids.update(package.id for package in exposed) + + if (not degraded and recommended_id and resolved.id != recommended_id + and not resolved.default): + recommended = spec_packages.get(recommended_id) + if recommended is not None and recommended.target_directory == resolved.target_directory: + warnings.append( + f"{name}: catalog entry '{entry.id}' names package '{resolved.id}' while " + f"model_specs/{entry.family}.json recommends '{recommended_id}' from the " + f"same install location" + ) + + for family in sorted(specs_by_family): + family_packages = packages_by_family.get(family, []) + installable = [ + package for package in family_packages + if package.format == "gguf" and package.installable + ] + if not installable: + # Families whose packages are all non-distributable (download.kind + # is not a Hugging Face snapshot) are deliberately absent from the + # UI, so their absence is not reported. + continue + if family not in families_in_catalog: + warnings.append( + f"model_specs/{family}.json publishes {len(installable)} installable GGUF " + f"package(s) but no {name} entry exposes the family" + ) + continue + + directories: dict[str, list[str]] = {} + for package in installable: + directories.setdefault(package.target_directory, []).append(package.id) + for directory in sorted(directories): + if any(package_id in reachable_ids for package_id in sorted(directories[directory])): + continue + warnings.append( + f"model_specs/{family}.json installs {', '.join(sorted(directories[directory]))} " + f"into '{directory}', which no {name} entry reaches" + ) + + recommended_id = recommended_package_id(specs_by_family[family]) + recommended = spec_packages.get(recommended_id) if recommended_id else None + if recommended_id and (recommended is None or recommended.family != family): + errors.append( + f"model_specs/{family}.json ui.recommended_package '{recommended_id}' is not a " + f"packages[].id in that spec" + ) + elif recommended is not None and not (recommended.format == "gguf" and recommended.installable): + errors.append( + f"model_specs/{family}.json ui.recommended_package '{recommended_id}' is " + f"format={recommended.format or 'unknown'} download.kind=" + f"'{recommended.download_kind or 'missing'}' and can never be offered by the UI" + ) + elif recommended_id and recommended_id not in reachable_ids: + errors.append( + f"model_specs/{family}.json ui.recommended_package '{recommended_id}' is not " + f"reachable from any {name} entry" + ) + for package in sorted(installable, key=lambda item: item.id): + if package.default and package.id not in reachable_ids: + errors.append( + f"model_specs/{family}.json default package '{package.id}' is not reachable " + f"from any {name} entry" + ) + return errors, warnings + + +def check_model_params_sync( + params: Any, + specs_by_family: dict[str, dict[str, Any]], + catalog_entries: list[CatalogEntry], + params_path: Path, +) -> tuple[list[str], list[str]]: + """Check webui/configs/model_params.json controls against the specs. + + Parameter groups are keyed by spec family or by catalog entry id. A control + whose name is absent from the owning spec's options.request is dead: strict + loaders reject the unknown key and the request fails, lenient ones drop it + silently. Only families that actually document their request options are + checked, because an empty options.request means "undocumented", not + "unsupported". + """ + errors: list[str] = [] + warnings: list[str] = [] + name = rel(params_path) + if not isinstance(params, dict): + return [f"{name}: top-level JSON must be an object"], warnings + + family_by_entry_id = {entry.id: entry.family for entry in catalog_entries} + for group, controls in params.items(): + if not isinstance(controls, list): + continue + family = group if group in specs_by_family else family_by_entry_id.get(group, "") + if not family and group in BUNDLED_LOADERS_WITHOUT_SPEC: + # Bundled families own no spec, so they are absent from + # specs_by_family, and their catalog ids are hyphenated while the + # group is keyed by the family name. Resolve them by name. + family = group + if not family: + errors.append( + f"{name}: parameter group '{group}' matches no model_specs family and no " + f"catalog entry id" + ) + continue + if family in BUNDLED_LOADERS_WITHOUT_SPEC: + # Bundled loaders have no spec by design, so there is no declared + # request surface to check a control against. They also validate + # nothing at runtime, so an unknown key is dropped rather than + # rejected: the controls are safe, just unverifiable from here. + continue + spec = specs_by_family.get(family) + if spec is None: + errors.append( + f"{name}: parameter group '{group}' resolves to family '{family}' " + f"which has no model_specs/{family}.json" + ) + continue + options = spec.get("options") + request = options.get("request") if isinstance(options, dict) else None + if not isinstance(request, list) or not request: + continue + declared = { + option.get("name") for option in request + if isinstance(option, dict) and isinstance(option.get("name"), str) + } + for control in controls: + if not isinstance(control, dict): + continue + control_name = control.get("name") + if not isinstance(control_name, str) or not control_name: + errors.append(f"{name}: parameter group '{group}' has a control without a name") + continue + if control_name not in declared: + warnings.append( + f"{name}: parameter group '{group}' control '{control_name}' is not in " + f"model_specs/{family}.json options.request" + ) + return errors, warnings + + class _SyncCheckSelfTests(unittest.TestCase): def test_parse_loader_declarations(self) -> None: text = """ @@ -225,18 +647,171 @@ def test_loader_json_family_parse(self) -> None: families = loader_families_from_json([{"family": "a"}, {"family": "b"}]) self.assertEqual(families, {"a", "b"}) + def test_parse_string_vocabulary(self) -> None: + text = ( + 'VoiceTaskKind parse_voice_task_kind(const std::string & value) {\n' + ' if (value == "vad") {\n return VoiceTaskKind::Vad;\n }\n' + ' if (value == "tts") {\n return VoiceTaskKind::Tts;\n }\n' + ' throw std::runtime_error("unsupported task: " + value);\n' + '}\n' + 'RunMode parse_run_mode(const std::string & value) {\n' + ' if (value == "offline") {\n return RunMode::Offline;\n }\n' + '}\n' + ) + self.assertEqual( + parse_string_vocabulary(text, "parse_voice_task_kind(const std::string & value)"), + {"vad", "tts"}, + ) + self.assertEqual( + parse_string_vocabulary(text, "parse_run_mode(const std::string & value)"), + {"offline"}, + ) + with self.assertRaises(ValueError): + parse_string_vocabulary(text, "parse_missing(const std::string & value)") + + def test_package_install_paths(self) -> None: + package = _self_test_package( + "demo_q8_0", + files=("Demo-GGUF/demo-q8_0.gguf", "Demo-GGUF/config.json"), + strip_prefix="Demo-GGUF", + ) + self.assertEqual(package.install_paths, { + "models/Demo-GGUF", + "models/Demo-GGUF/demo-q8_0.gguf", + "models/Demo-GGUF/config.json", + }) + + def test_catalog_download_id_must_be_a_package_id(self) -> None: + errors, _ = _self_test_catalog_check( + [_self_test_entry(download_id="demo")], + {"demo_q8_0": _self_test_package("demo_q8_0")}, + ) + self.assertTrue(any("download_id 'demo' is not a packages[].id" in error for error in errors)) + + def test_catalog_path_must_match_the_package(self) -> None: + packages = {"demo_q8_0": _self_test_package("demo_q8_0", files=("demo-q8_0.gguf",))} + errors, _ = _self_test_catalog_check( + [_self_test_entry(path="models/Demo")], packages) + self.assertTrue(any("path 'models/Demo' is neither" in error for error in errors)) + errors, _ = _self_test_catalog_check( + [_self_test_entry(path="models/Demo-GGUF/demo-q8_0.gguf")], packages) + self.assertEqual(errors, []) + + def test_catalog_rejects_unknown_task_and_family(self) -> None: + errors, _ = _self_test_catalog_check( + [_self_test_entry(task="nope"), _self_test_entry(entry_id="x", family="ghost")], + {"demo_q8_0": _self_test_package("demo_q8_0")}, + ) + self.assertTrue(any("task 'nope' is not accepted" in error for error in errors)) + self.assertTrue(any("no model_specs/ghost.json" in error for error in errors)) + + def test_catalog_reports_unreachable_directory_and_default(self) -> None: + packages = { + "demo_q8_0": _self_test_package("demo_q8_0", default=True), + "demo_extra_q8_0": _self_test_package( + "demo_extra_q8_0", target_directory="Demo-GGUF/extra"), + } + _, warnings = _self_test_catalog_check([_self_test_entry()], packages) + self.assertTrue(any("which no" in warning and "extra" in warning for warning in warnings)) + errors, _ = _self_test_catalog_check( + [_self_test_entry(download_id="demo_extra_q8_0", path="models/Demo-GGUF/extra")], + packages, + ) + self.assertTrue(any("default package 'demo_q8_0' is not reachable" in e for e in errors)) + + def test_catalog_reports_unsupported_downloads_as_not_installable(self) -> None: + packages = {"demo_q8_0": _self_test_package("demo_q8_0", download_kind="unsupported")} + errors, warnings = _self_test_catalog_check([], packages) + self.assertEqual(errors, []) + self.assertEqual([w for w in warnings if "no model_specs" in w], []) + self.assertEqual(check_native_manager_sync(packages) != [], True) + + def test_model_params_group_must_resolve(self) -> None: + specs = {"demo": {"family": "demo", "options": {"request": [{"name": "speed"}]}}} + errors, warnings = check_model_params_sync( + {"demo": [{"name": "speed"}, {"name": "gone"}], "ghost": [{"name": "speed"}]}, + specs, + [], + MODEL_PARAMS_PATH, + ) + self.assertTrue(any("parameter group 'ghost' matches no" in error for error in errors)) + self.assertTrue(any("control 'gone' is not in" in warning for warning in warnings)) + + +def _self_test_package( + package_id: str, + *, + family: str = "demo", + target_directory: str = "Demo-GGUF", + files: tuple[str, ...] = (), + strip_prefix: str = "", + default: bool = False, + download_kind: str = INSTALLABLE_DOWNLOAD_KIND, +) -> SpecPackage: + return SpecPackage( + family=family, + id=package_id, + format="gguf", + target_directory=target_directory, + default=default, + download_kind=download_kind, + files=files, + strip_prefix=strip_prefix, + ) + + +def _self_test_entry( + *, + entry_id: str = "demo", + family: str = "demo", + path: str = "models/Demo-GGUF", + task: str = "tts", + mode: str = "offline", + download_id: str = "demo_q8_0", +) -> CatalogEntry: + return CatalogEntry( + index=0, + id=entry_id, + family=family, + path=path, + task=task, + mode=mode, + download_id=download_id, + ) + + +def _self_test_catalog_check( + entries: list[CatalogEntry], + packages: dict[str, SpecPackage], +) -> tuple[list[str], list[str]]: + specs = {"demo": {"family": "demo", "ui": {"recommended_package": "demo_q8_0"}}} + return check_catalog_sync( + entries, specs, packages, {"tts", "asr"}, {"offline", "streaming"}, CATALOG_PATH) + + +def advisory(message: str) -> str: + return f"{message} (advisory; --strict-catalog makes this fail)" + def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--cmake", type=Path, default=CMAKE_PATH, help="Path to top-level CMakeLists.txt") parser.add_argument("--registry", type=Path, default=REGISTRY_PATH, help="Path to registry.cpp") parser.add_argument("--specs-dir", type=Path, default=SPECS_DIR, help="Directory containing model spec JSON files") + parser.add_argument("--catalog", type=Path, default=CATALOG_PATH, help="Path to models_catalog.json") + parser.add_argument("--model-params", type=Path, default=MODEL_PARAMS_PATH, help="Path to model_params.json") + parser.add_argument("--session", type=Path, default=SESSION_PATH, help="Path to session.cpp") parser.add_argument( "--loader-json", type=Path, default=None, help="Optional audiocpp_cli --list-loaders --json output. Use '-' to read stdin.", ) + parser.add_argument( + "--strict-catalog", + action="store_true", + help="Fail on WebUI catalog and model_params findings instead of reporting them as advisories", + ) parser.add_argument("--self-test", action="store_true", help="Run built-in unit tests and exit") args = parser.parse_args() @@ -270,35 +845,83 @@ def main() -> int: print("error: no active loaders found", file=sys.stderr) return 2 + for path, label in ((args.catalog, "catalog"), (args.model_params, "model params"), (args.session, "session")): + if not path.is_file(): + print(f"error: {label} not found: {path}", file=sys.stderr) + return 2 + try: + session_text = args.session.read_text(encoding="utf-8") + task_kinds = parse_string_vocabulary(session_text, "parse_voice_task_kind(const std::string & value)") + run_modes = parse_string_vocabulary(session_text, "parse_run_mode(const std::string & value)") + except Exception as exc: + print(f"error: failed to read task kinds from {rel(args.session)}: {exc}", file=sys.stderr) + return 2 + specs_by_family, spec_packages, spec_errors = load_spec_packages(args.specs_dir) manager_packages, manager_errors = load_manager_packages(args.specs_dir) + catalog_entries, catalog_errors = load_catalog(args.catalog) errors.extend(spec_errors) errors.extend(manager_errors) + errors.extend(catalog_errors) errors.extend(check_loader_spec_sync(active_loaders, specs_by_family)) manager_sync_errors, manager_sync_warnings = check_manager_sync(spec_packages, manager_packages) errors.extend(manager_sync_errors) warnings.extend(manager_sync_warnings) + warnings.extend(check_native_manager_sync(spec_packages)) + catalog_sync_errors, catalog_sync_warnings = check_catalog_sync( + catalog_entries, specs_by_family, spec_packages, task_kinds, run_modes, args.catalog) + # The catalog pass is new, and the catalog it inspects predates it. Reporting + # its findings as advisories by default means this check can land without + # rewriting the data it measures; --strict-catalog is what CI turns on once + # the catalog corrections have followed. + if args.strict_catalog: + errors.extend(catalog_sync_errors) + else: + warnings.extend(advisory(error) for error in catalog_sync_errors) + warnings.extend(catalog_sync_warnings) + try: + params_payload = load_json(args.model_params) + except json.JSONDecodeError as exc: + errors.append(f"{rel(args.model_params)}: invalid JSON: {exc}") + params_payload = {} + params_errors, params_warnings = check_model_params_sync( + params_payload, specs_by_family, catalog_entries, args.model_params) + if args.strict_catalog: + errors.extend(params_errors) + else: + warnings.extend(advisory(error) for error in params_errors) + warnings.extend(params_warnings) + server_packages = sum(1 for package in spec_packages.values() if package.installable) print( f"active_loaders={len(active_loaders)} commented_loaders={len(commented_loaders)} " f"specs={len(specs_by_family)} packages={len(spec_packages)} " - f"manager_packages={len(manager_packages)}" + f"manager_packages={len(manager_packages)} server_packages={server_packages} " + f"catalog_entries={len(catalog_entries)} task_kinds={len(task_kinds)}" ) for warning in warnings: print(f"warning: {warning}") if errors: - print("loader/spec sync failed:", file=sys.stderr) + print("loader/spec/catalog sync failed:", file=sys.stderr) for error in errors: print(f" - {error}", file=sys.stderr) print( "\nFix: keep model_specs/*.json, model_manager_v2.py, registered loaders, " - "and published default GGUF packages aligned. Schema-level validation " - "belongs to the typed model-spec validator, and WebUI placement is checked separately.", + "published default GGUF packages, and webui/configs/models_catalog.json aligned. " + "A catalog entry must name a real packages[].id and install into that package's " + "location. Schema-level validation belongs to the typed model-spec validator.", file=sys.stderr, ) return 1 - print("ok: runtime loaders, model_specs, and model_manager_v2 are in sync") + advisories = sum(1 for warning in warnings if warning.endswith("--strict-catalog makes this fail)")) + if advisories: + print( + f"ok: runtime loaders, model_specs and model_manager_v2 are in sync; " + f"{advisories} WebUI catalog advisories reported (--strict-catalog fails on them)" + ) + else: + print("ok: runtime loaders, model_specs, model_manager_v2, and the WebUI catalog are in sync") return 0