diff --git a/data/gguf_fingerprints.json b/data/gguf_fingerprints.json index 2ab4fc2a3..ac99a748d 100644 --- a/data/gguf_fingerprints.json +++ b/data/gguf_fingerprints.json @@ -186,6 +186,35 @@ ], "repo": "unsloth/Qwen3.8-Flash-Next-GGUF" }, + "gyro/Gyro-S": { + "arch": "qwen4exp", + "tensors": 1224, + "bytes": 58475973120, + "expert_bytes": { + "TQK6": 16357785600, + "TQK7": 9437184000 + }, + "experts_bpw": 1.708, + "bpw": 2.644, + "files": [ + "Qwen3.8-Flash-Next-Gyro-S-TQ1_0.gguf" + ], + "repo": "agentionai/Qwen3.8-Flash-Next-Gyro-GGUF" + }, + "gyro/Gyro-M": { + "arch": "qwen4exp", + "tensors": 1224, + "bytes": 91954731520, + "expert_bytes": { + "TQ2_T": 32086425600 + }, + "experts_bpw": 2.125, + "bpw": 4.157, + "files": [ + "Qwen3.8-Flash-Next-Gyro-M-TQ2_0.gguf" + ], + "repo": "agentionai/Qwen3.8-Flash-Next-Gyro-GGUF" + }, "_roles": { "blk.N.attn_gate.weight": [ "IQ3_S", @@ -252,7 +281,9 @@ "IQ4_NL", "Q2_0", "Q5_1", - "Q8_0" + "Q8_0", + "TQ2_T", + "TQK7" ], "blk.N.ffn_down_shexp.weight": [ "IQ4_NL", @@ -271,7 +302,9 @@ "IQ4_XS", "Q2_0", "Q4_K", - "Q5_K" + "Q5_K", + "TQ2_T", + "TQK6" ], "blk.N.ffn_gate_inp.weight": [ "BF16", @@ -301,7 +334,9 @@ "IQ4_XS", "Q2_0", "Q4_K", - "Q5_K" + "Q5_K", + "TQ2_T", + "TQK6" ], "blk.N.ffn_up_shexp.weight": [ "IQ3_S", @@ -315,11 +350,13 @@ ], "blk.N.hc_attn_down.weight": [ "BF16", + "Q6_K", "Q8_0" ], "blk.N.hc_attn_inject.weight": [ "BF16", - "F32" + "F32", + "Q6_K" ], "blk.N.hc_attn_norm.weight": [ "F32" @@ -330,11 +367,13 @@ ], "blk.N.hc_ffn_down.weight": [ "BF16", + "Q6_K", "Q8_0" ], "blk.N.hc_ffn_inject.weight": [ "BF16", - "F32" + "F32", + "Q6_K" ], "blk.N.hc_ffn_norm.weight": [ "F32" @@ -362,6 +401,7 @@ "blk.N.ple_key.weight": [ "BF16", "Q2_0", + "Q6_K", "Q8_0" ], "blk.N.ple_norm_conv.weight": [ @@ -375,6 +415,7 @@ ], "blk.N.ple_value.weight": [ "BF16", + "Q6_K", "Q8_0" ], "blk.N.ssm_a": [ @@ -382,11 +423,13 @@ ], "blk.N.ssm_alpha.weight": [ "BF16", - "F32" + "F32", + "Q5_K" ], "blk.N.ssm_beta.weight": [ "BF16", - "F32" + "F32", + "Q5_K" ], "blk.N.ssm_conv1d.weight": [ "F32" @@ -414,6 +457,7 @@ ], "output_hc_down.weight": [ "BF16", + "IQ4_NL", "Q8_0" ], "output_hc_norm.weight": [ @@ -421,6 +465,7 @@ ], "output_hc_up.weight": [ "BF16", + "IQ4_NL", "Q8_0" ], "per_layer_token_embd.weight": [ @@ -430,6 +475,7 @@ "IQ3_S", "IQ4_XS", "Q3_K", + "Q6_K", "Q8_0" ] } diff --git a/strata_cli/cli.py b/strata_cli/cli.py index 4b3afca98..28b1f1891 100644 --- a/strata_cli/cli.py +++ b/strata_cli/cli.py @@ -514,9 +514,7 @@ def _models(a: argparse.Namespace) -> int: else: print(f"GGUF file set is incomplete or invalid: {', '.join(result['issues'])}", file=sys.stderr) return 5 - return _success({"plan": result}, machine=a.output == "json", - human=f"{result['file_count']} GGUF file(s): {result['file_set_status']}\n" - f"Adoption: {result['adoption_status']}\nNo files changed") + return _success({"plan": result}, machine=a.output == "json", human=_import_words(result)) if a.models_command == "list": if a.served: if a.install_dir: @@ -544,7 +542,12 @@ def _models(a: argparse.Namespace) -> int: result = models.list_installed(a.install_dir) lines = [f"{item['id']} {item['component_status']} " f"{item.get('name', item.get('error', ''))}" for item in result["models"]] - return _success(result, machine=a.output == "json", human="\n".join(lines) or "No local model run configs") + if not lines: + where = ", ".join(result.get("install_dirs") or [result["install_dir"]]) + lines = [f"No installed models found. Looked in: {where}. " + "If your models are somewhere else, add --install-dir FOLDER (the folder with the strata-*.json files), " + "or run `strata launch` to set one up."] + return _success(result, machine=a.output == "json", human="\n".join(lines)) result = models.inspect(a.model_id, a.install_dir) item = result["model"] return _success(result, machine=a.output == "json", @@ -554,6 +557,24 @@ def _models(a: argparse.Namespace) -> int: f"issues: {len(result['components']['issues'])}") +def _import_words(plan: dict) -> str: + """The plain-words result of `strata models import --plan`: what the files are, what is already set up, what changes (nothing).""" + cand = plan.get("candidate_selection") or {} + lines = [f"{plan['file_count']} GGUF file(s): {plan['file_set_status']}"] + if cand.get("family"): + lines.append(f"These files are Strata's {cand['family']} {cand['model']}.") + done = plan.get("already_set_up") + if done: + have = [name for key, name in (("pack", "prepared model data"), ("mtp", "draft layer"), ("mmproj", "picture reader")) if done.get(key)] + if done.get("config"): + lines.append(f"Already set up: {done['config']}" + (f" (with the {', '.join(have)})" if have else "")) + elif have: + lines.append(f"Prepared files found beside it: {', '.join(have)}.") + lines.append(f"Adoption: {plan['adoption_status']}") + lines.append("No files changed") + return "\n".join(lines) + + def _redact_argv(args: list[str]) -> list[str]: out = [] conceal_next = False diff --git a/strata_cli/launch/app.py b/strata_cli/launch/app.py index 1ae6c96de..bb84e876b 100644 --- a/strata_cli/launch/app.py +++ b/strata_cli/launch/app.py @@ -98,6 +98,31 @@ class Ctx: except Exception: # noqa: BLE001 return None + # ---- where the model is and what it still needs + def model_state(self, data_dir: str | None = None): + """The chosen model's state (installed, what helper files are missing, sizes) for these choices.""" + from strata_cli.launch import modelstate + s = self.settings + return modelstate.assess(self.catalog, s.family, s.model, vision=s.vision, model_path=s.model_path, + data_dir=data_dir if data_dir is not None else s.data_dir, rows=self.rows) + + def default_folder(self) -> Path: + """The folder the Folder step starts with: where the chosen model already is (installed, or the person's own files with their + prepared data), else the data folder every command agrees on.""" + s = self.settings + if s.model_path: + try: + from strata_control import discovery + p = Path(s.model_path).expanduser() + first = p if p.is_file() else next(iter(sorted(p.glob("*.gguf"))), None) + found = discovery.prepared_beside(first)["data_dir"] if first else None + if found: + return Path(found) + except OSError: + pass + return self.catalog.data_dir() + return self.catalog.data_dir_for(s.family, s.model) + def invalidate_rows(self) -> None: self._adv = None @@ -245,10 +270,11 @@ class LaunchApp(App): if choice == "go": self.run_launch(confirmed=True) - self.push_screen(ConfirmScreen(f"Download about {gb:.0f} GB?", - f"This model is not on this computer yet. Strata will download about {gb:.0f} GB now; an interrupted download continues where it stopped." - + room + warn, - [("go", "Download and start"), ("no", "Not now")]), done) + asker = getattr(self.runner, "download_question", None) + title, text = asker() if asker else (f"Download about {gb:.0f} GB?", + f"This model is not on this computer yet. Strata will download about {gb:.0f} GB now; " + "an interrupted download continues where it stopped.") + self.push_screen(ConfirmScreen(title, text + room + warn, [("go", "Download and start"), ("no", "Not now")]), done) return screen = RunScreen() self.run_screen = screen diff --git a/strata_cli/launch/catalog.py b/strata_cli/launch/catalog.py index ebfe0fcb8..bf80ee80f 100644 --- a/strata_cli/launch/catalog.py +++ b/strata_cli/launch/catalog.py @@ -10,6 +10,7 @@ words instead of guessing. from __future__ import annotations import importlib.util +import json import os import sys from dataclasses import dataclass, field @@ -151,15 +152,23 @@ def installer_folder(explicit: str | os.PathLike | None = None) -> Path: return Path(__file__).resolve().parents[2] -def install_root(explicit: str | os.PathLike | None = None) -> Path: - """Where run configs of installed models live: STRATA_INSTALL_DIR, else beside the components folder (what the app uses).""" +def install_places(explicit: str | os.PathLike | None = None, src: str | os.PathLike | None = None) -> list: + """Every folder with run configs, main one first: the SAME discovery `strata models list` and the own-files import use + (strata_control/discovery.py). An explicit folder (the argument, or STRATA_INSTALL_DIR) is the only one.""" + from strata_control import discovery + return discovery.places(here=installer_folder(src), explicit=explicit) + + +def install_root(explicit: str | os.PathLike | None = None, src: str | os.PathLike | None = None) -> Path: + """Where run configs of installed models live: the main place found by the shared discovery, else the folder a new model is + installed into (beside the components folder, what the app uses).""" + from strata_control import discovery if explicit: return Path(explicit) env = os.environ.get("STRATA_INSTALL_DIR") if env: return Path(env) - from strata_control import engine_component - return engine_component.components_dir().parent / "install" + return discovery.find_install(here=installer_folder(src))[0] # Intel Arc cards by PCI id: (name, memory in GB). The kernel does not tell an ordinary user how much memory the card has, so the common @@ -192,7 +201,8 @@ class Catalog: def __init__(self, src: str | os.PathLike | None = None, root: str | os.PathLike | None = None, hardware: dict | None = None): self.src = installer_folder(src) - self.root = install_root(root) + self.places = install_places(root, self.src) + self.root = Path(root) if root else (self.places[0].path if self.places else install_root(None, self.src)) mcp = self.src / "tools" / "strata_mcp.py" if not mcp.is_file() or not (self.src / "setup.py").is_file(): raise CatalogUnavailable( @@ -215,6 +225,8 @@ class Catalog: sys.stdout = real self.mcp = mod self.strata = mod.Strata(self.root, self.src) + self._mcp_data_dir = self.strata.data_dir + self.strata.data_dir = self.data_dir # one answer to "where is the data folder" for every caller, the app's included self._hardware_override = hardware # ---- the computer @@ -281,7 +293,7 @@ class Catalog: about=size["about"], download_gb=float(size["download_gb"]), need_vram_gb=f["vramUseGb"], need_ram_gb=f["ramUseGb"], spill_gb=f["spillGb"], verdict=cat, words=words, experimental=bool(size.get("experimental") or fam.get("experimental")), images=bool(size.get("images")), - installed=bool(size.get("installed")), mem_total=bool(size.get("mem_total")), + installed=bool(size.get("installed")) or self.installed_config(str(size["id"])) is not None, mem_total=bool(size.get("mem_total")), memory_gb=float(size.get("memory_gb") or 0), recommended=(rec.get("family") == fam["family"] and rec.get("model") == size["model"]))) return out @@ -348,19 +360,55 @@ class Catalog: # ---- installed models def installed_configs(self) -> list[dict]: - out = [] + out, seen = [], set() for path in self.strata.configs(): + seen.add(path.name) d = self.strata.describe_config(path) d["path"] = str(path) out.append(d) + for place in self.places: + if place.path == self.root: + continue + from strata_control import discovery + for path in discovery.config_files(place.path): + if path.name in seen: + continue + seen.add(path.name) + d = self.strata.describe_config(path) + d["path"] = str(path) + out.append(d) return out def installed_config(self, tag: str) -> Path | None: - p = self.root / f"strata-{tag.lower()}.json" - return p if p.is_file() else None + name = f"strata-{tag.lower()}.json" + for folder in [self.root, *[p.path for p in self.places if p.path != self.root]]: + p = folder / name + if p.is_file(): + return p + return None + + def data_dir_choice(self) -> tuple[Path, str]: + """(the data folder, where it came from): the person's own setting, the desktop app's folder, a Documents/Strata that holds models, + else the default beside the install. The same rule for every command.""" + from strata_control import discovery + return discovery.data_dir(Path(self._mcp_data_dir())) def data_dir(self) -> Path: - return self.strata.data_dir() + return self.data_dir_choice()[0] + + def data_dir_for(self, family: str | None, model: str | None) -> Path: + """Where this model's files are: the folder its installed run config names, else the data folder.""" + if family and model: + path = self.installed_config(self.tag_of(family, model)) + if path is not None: + try: + from strata_control import discovery + found = discovery.footprint(json.loads(path.read_text(encoding="utf-8-sig")), path).data_dir + if found is not None: + return found + except (OSError, ValueError): + pass + return self.data_dir() def disk_free(self, path: str | os.PathLike) -> float | None: return self.strata.disk_free(Path(path)) diff --git a/strata_cli/launch/modelstate.py b/strata_cli/launch/modelstate.py new file mode 100644 index 000000000..fc4729daa --- /dev/null +++ b/strata_cli/launch/modelstate.py @@ -0,0 +1,163 @@ +"""What a model still needs on this computer, in plain words: the one place the Folder step, the download question and the final summary ask. + +A model that is already installed is "already downloaded". What can still be missing are helper files, each with its own size: the +draft layer (MTP, makes answers faster) and the picture reader. Only when the weights themselves are gone does it say the model +is downloaded again. Nothing here reads more than a run config and file existence. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from pathlib import Path + +from strata_control import discovery +from strata_control.discovery import MMPROJ_GB, MTP_FETCH_GB + + +@dataclass +class Part: + kind: str # weights | mtp | mmproj + label: str # "the draft layer" + gb: float + + +@dataclass +class ModelState: + installed: bool = False # a run config for this model exists and names files that are there + own_files: bool = False # the person's own model files + total_gb: float = 0.0 # the whole model (the catalog's figure) + parts: list[Part] = field(default_factory=list) # what is missing and would be downloaded + folder: Path | None = None # where the model lives now (its data folder), None when not installed + on_disk_gb: float = 0.0 # weights already in the model's folder (a partial download) + notes: list[str] = field(default_factory=list) + + @property + def remaining_gb(self) -> float: + return sum(p.gb for p in self.parts) + + @property + def only_helpers(self) -> bool: + return bool(self.parts) and all(p.kind != "weights" for p in self.parts) + + def _names(self) -> str: + names = [f"{p.label} ({p.gb:.1f} GB)" for p in self.parts] + return names[0] if len(names) == 1 else ", ".join(names[:-1]) + " and " + names[-1] + + def headline(self) -> str: + """One plain sentence for the size line of the Folder step.""" + if self.own_files: + if not self.parts: + return "Your model files are used where they are. Nothing needs to be downloaded." + return f"Your model files are used where they are. Strata still needs to fetch {self._names()}." + if self.installed and not self.parts: + return f"Already downloaded ({self.total_gb:.0f} GB on this computer). Nothing more to download." + if self.installed and self.only_helpers: + first = self.parts[0] + if len(self.parts) == 1: + return f"The model is already downloaded ({self.total_gb:.0f} GB). Only {first.label} is missing ({first.gb:.1f} GB)." + return f"The model is already downloaded ({self.total_gb:.0f} GB). Only {self._names()} are missing." + if self.installed: + return (f"The model's files are no longer all on this computer. About {self.remaining_gb:.0f} GB would be downloaded " + f"({self._names()}).") + if self.on_disk_gb >= 1: + return (f"{self.on_disk_gb:.0f} GB of the model is already in the folder. About {self.remaining_gb:.0f} GB left to download " + f"(the whole model is {self.total_gb:.0f} GB).") + return f"Not downloaded yet: about {self.remaining_gb:.0f} GB to download." + + def question(self) -> tuple[str, str]: + """(title, text) for the "Download?" pop-up before a start that needs a download.""" + gb = self.remaining_gb + if self.only_helpers: + return (f"Download {self._names()}?", + "The model itself is already on this computer and is not downloaded again. Strata needs this extra file to start " + "with the choices you made; an interrupted download continues where it stopped.") + return (f"Download about {gb:.0f} GB?", + f"This model is not on this computer yet. Strata will download about {gb:.0f} GB now; " + "an interrupted download continues where it stopped.") + + +def mmproj_name(catalog, family: str | None) -> str: + try: + _m, families, _c, _s = catalog.tables() + name = (families.get(family or "") or {}).get("mmproj") + if name: + return str(name) + mod = catalog.strata.setup_module() + if mod is not None and getattr(mod, "MMPROJ", None): + return str(mod.MMPROJ) + except Exception: # noqa: BLE001 + pass + return "mmproj-Qwen3.8-Flash-Next-BF16.gguf" + + +def assess(catalog, family: str | None, model: str | None, *, vision: str = "off", model_path: str | None = None, + data_dir: str | Path | None = None, config: Path | None = None, desc: dict | None = None, rows: list | None = None) -> ModelState: + """The state of `family model` for these choices. `config` / `desc` are the installed run config and its description (looked up when + not given).""" + st = ModelState() + row = next((r for r in (rows if rows is not None else catalog.rows()) if r.family == family and r.model == model), None) if (family and model) else None + st.total_gb = float(row.download_gb) if row else 0.0 + wants_pictures = vision != "off" and (row is None or row.images) + data = Path(data_dir) if data_dir else None + if model_path: + st.own_files = True + first = _first_gguf(Path(model_path)) + beside = discovery.prepared_beside(first) if first else {} + st.folder = beside.get("data_dir") + if beside.get("pack") is not None and beside.get("mtp") is None: + st.parts.append(Part("mtp", "the draft layer", MTP_FETCH_GB)) + if wants_pictures and beside.get("pack") is not None and not beside.get("mmproj"): + st.parts.append(Part("mmproj", "the picture reader", MMPROJ_GB)) + return st + if config is None and family and model: + config = catalog.installed_config(catalog.tag_of(family, model)) + cfg = None + if config is not None: + try: + import json + cfg = json.loads(Path(config).read_text(encoding="utf-8-sig")) + except (OSError, ValueError): + cfg = None + if cfg is not None: + fp = discovery.footprint(cfg, Path(config)) + st.installed = True + st.folder = fp.data_dir + gone = [m for m in fp.missing() if m.kind == "weights"] + if gone: + st.installed = False + st.parts.append(Part("weights", "the model", max(st.total_gb, 0.0))) + return st + for m in fp.missing(): + if m.kind == "mtp": + st.parts.append(Part("mtp", "the draft layer", MTP_FETCH_GB)) + elif m.kind == "mmproj" and wants_pictures: + st.parts.append(Part("mmproj", "the picture reader", MMPROJ_GB)) + images = (desc or {}).get("images") if desc is not None else ("cpu" if fp.mmproj else "off") + if wants_pictures and (images in (None, "off")) and not any(p.kind == "mmproj" for p in st.parts): + base = st.folder / "models" if st.folder else data + have = (base / mmproj_name(catalog, family)) if base else None + if have is None or not have.exists(): + st.parts.append(Part("mmproj", "the picture reader", MMPROJ_GB)) + return st + # not installed: whatever of the weights is in the model's folder counts + if family and model: + base = data or Path(catalog.data_dir()) + folder = base / "models" / catalog.tag_of(family, model) + got = 0 + try: + got = sum(f.stat().st_size for f in folder.glob("*.gguf") if not f.name.lower().startswith("mmproj")) + except OSError: + pass + st.on_disk_gb = got / 1e9 + st.parts.append(Part("weights", "the model", max(st.total_gb - st.on_disk_gb, 0.0))) + return st + + +def _first_gguf(path: Path) -> Path | None: + try: + if path.is_file(): + return path + found = sorted(p for p in path.glob("*.gguf") if not p.name.lower().startswith("mmproj")) + return found[0] if found else None + except OSError: + return None diff --git a/strata_cli/launch/runner.py b/strata_cli/launch/runner.py index b0c581567..df8afa4c9 100644 --- a/strata_cli/launch/runner.py +++ b/strata_cli/launch/runner.py @@ -396,7 +396,7 @@ class Runner: args += ["--kv", s.kv] if fam in ("qwen", "coder"): args += ["--experimental-speed-projection", "off"] - data = s.data_dir + data = s.data_dir or self.existing_data_dir() if data: args += ["--data-dir", str(data)] if s.model_path: @@ -427,26 +427,54 @@ class Runner: desc = self.catalog.strata.describe_config(path) return path, desc - def download_gb(self) -> float: - """How much would be downloaded for this launch right now: 0 when the model is installed with these choices (or is the person's own files).""" + def existing_data_dir(self) -> Path | None: + """The data folder to give setup when the person chose none: the folder the model already lives in (its run config names its files), + else the folder the desktop app uses, so a helper file that is missing is fetched next to the rest, never into a second folder. + None: setup's own default is right.""" s = self.s - if s.model_path or not (s.family and s.model): + try: + if s.model_path: + from strata_control import discovery + first = next(iter(sorted(Path(s.model_path).glob("*.gguf"))), None) if Path(s.model_path).is_dir() else Path(s.model_path) + found = discovery.prepared_beside(first)["data_dir"] if first else None + return Path(found) if found else None + if s.family and s.model: + path = self.catalog.installed_config(self.catalog.tag_of(s.family, s.model)) + if path is not None: + from strata_control import discovery + found = discovery.footprint(read_config(path), path).data_dir + if found is not None: + return found + found, source = self.catalog.data_dir_choice() + return found if source != "default" else None + except Exception: # noqa: BLE001 + return None + + def model_state(self): + """What this launch still needs for the model, in numbers and words (strata_cli/launch/modelstate.py).""" + from strata_cli.launch import modelstate + s = self.s + path, desc = (None, None) + if s.family and s.model and not s.model_path: + path, desc = self.installed_state() + return modelstate.assess(self.catalog, s.family, s.model, vision=s.vision, model_path=s.model_path, data_dir=s.data_dir, + config=path, desc=desc) + + def download_gb(self) -> float: + """How much would be downloaded for this launch right now: 0 when everything is on disk (the model is installed with these choices + or is the person's own files). A model whose helper files are missing counts only those files, never the whole model.""" + s = self.s + if not (s.family and s.model) and not s.model_path: return 0.0 try: - path, desc = self.installed_state() - lost = files_present(read_config(path)) if path is not None else [] - if lost and all("mmproj" in Path(x).name.lower() for x in lost): - return 0.9 # only the picture reader's file is gone - if path is not None and desc is not None and not lost: - if desc_matches(desc, s): - return 0.0 - # the model is on disk: what differs is the context / memory format (nothing to download) or the pictures (the 0.9 GB reader file) - return 0.9 if (s.vision != "off" and (desc.get("images") or "off") == "off") else 0.0 - row = next((r for r in self.catalog.rows() if r.family == s.family and r.model == s.model), None) - return float(row.download_gb) if row else 0.0 + return float(self.model_state().remaining_gb) except Exception: # noqa: BLE001 return 0.0 + def download_question(self) -> tuple[str, str]: + """(title, text) of the question before a download starts, naming exactly what is fetched.""" + return self.model_state().question() + def ensure_model(self) -> Path: s, cat = self.s, self.catalog if s.model_path and not (s.family and s.model): diff --git a/strata_cli/launch/screens.py b/strata_cli/launch/screens.py index f090b684e..1391b1c51 100644 --- a/strata_cli/launch/screens.py +++ b/strata_cli/launch/screens.py @@ -588,7 +588,7 @@ class FolderScreen(Base): def on_mount(self) -> None: s = self.ctx.settings f = self.query_one("#folderpath", Input) - f.value = s.data_dir or str(self.ctx.catalog.data_dir()) + f.value = s.data_dir or str(self.ctx.default_folder()) f.focus() self.paint() @@ -601,33 +601,39 @@ class FolderScreen(Base): def on_input_submitted(self, event: Input.Submitted) -> None: self.action_next() - def need_gb(self) -> float: - s = self.ctx.settings - r = next((x for x in self.ctx.rows if (x.family, x.model) == (s.family, s.model)), None) - if r is None: - return 8.0 - return 8.0 + (0 if s.model_path else r.download_gb) + def state(self, value: str | None = None): + return self.ctx.model_state(value or None) + + def need_gb(self, st=None) -> float: + """The room to ask for: what is still to download plus working room (a little when the model is already there).""" + st = st or self.state() + return st.remaining_gb + (2.0 if (st.installed or st.own_files) else 8.0) def refresh_body(self) -> None: try: value = self.query_one("#folderpath", Input).value.strip() except Exception: # noqa: BLE001 return - free = self.ctx.catalog.disk_free(value or self.ctx.catalog.data_dir()) - need = self.need_gb() + free = self.ctx.catalog.disk_free(value or self.ctx.default_folder()) + st = self.state(value) + need = self.need_gb(st) s = self.ctx.settings out = Text() if s.model_path: out.append(f"Your model files stay where they are: {s.model_path}\n", style=FG) out.append("Strata only writes its prepared data here (packs, a small draft layer).\n", style=MUTE) - else: - out.append(f"Needs about {need:.0f} GB (the download plus working room).", style=FG) - out.append("\n") + out.append(st.headline(), style=FG) + out.append("\n") + if st.installed and st.folder is not None: + out.append(f"It is stored in {st.folder}\n", style=MUTE) + if st.parts and (st.installed or st.own_files): + for p in st.parts: + out.append(f" - missing: {p.label}, about {p.gb:.1f} GB\n", style=WARN) if free is not None: warn = free < need out.append(f"Free space in that folder's drive: {free:,.0f} GB", style=WARN if warn else FG) if warn: - out.append(f" Not enough for {need:.0f} GB. You can still go on, but the download is likely to stop; pick a bigger drive.", style=WARN) + out.append(f" Not enough for the {need:.0f} GB needed. You can still go on, but the download is likely to stop; pick a bigger drive.", style=WARN) else: out.append("The free space of that folder could not be read.", style=MUTE) moving = self.models_that_would_move(value) @@ -639,7 +645,7 @@ class FolderScreen(Base): def models_that_would_move(self, value: str) -> str | None: """The folder holding model files today when a different one is typed (setup gathers the model files into the data folder), else None.""" try: - old = Path(self.ctx.catalog.data_dir()) + old = Path(self.ctx.default_folder()) new = Path(value).expanduser() if value else old if os.path.normcase(str(new.resolve())) == os.path.normcase(str(old.resolve())): return None @@ -651,12 +657,15 @@ class FolderScreen(Base): return None def explain(self) -> Text: + if self.state().installed: + return Text("This model is already on this computer, so the folder starts where it is. Press Enter to keep it. " + "Another folder only matters for files that still have to be downloaded.", style=MUTE) return Text("Models are large files. Strata keeps them in one folder; the default is Strata's own data folder. " "Press Enter to keep it, or type another folder (it is created if it does not exist).", style=MUTE) def commit(self) -> None: value = self.query_one("#folderpath", Input).value.strip().strip('"') - default = str(self.ctx.catalog.data_dir()) + default = str(self.ctx.default_folder()) self.ctx.settings.data_dir = None if (not value or value == default) else value diff --git a/strata_control/discovery.py b/strata_control/discovery.py new file mode 100644 index 000000000..e65ef9aa3 --- /dev/null +++ b/strata_control/discovery.py @@ -0,0 +1,345 @@ +"""Where Strata's existing installs and model files are: ONE place that the launch screens, `strata models list`, the own-files import +and `strata launch` all ask. + +Three kinds of place hold a model's run config (`strata-.json`), looked at in this order, the first that has any is the main one: + +* explicit STRATA_INSTALL_DIR or an --install-dir argument (it is the only place looked at); +* here the Terminal's own folder, when it has run configs (a source checkout that ran setup); +* remembered the folders earlier Strata installs wrote into the user's settings file (``installs``); +* app the desktop app's install folder, which the Terminal shares: ``/../install`` (Windows + ``%LOCALAPPDATA%\\Strata\\install``, Linux and Mac ``$XDG_DATA_HOME/strata/install``, default ``~/.local/share/strata/install``). + The app's own settings file (``app-settings.json`` beside it) names its data folder and installs. + +The desktop app itself (it sets STRATA_SETTINGS_FILE) never adopts an older command-line install, so for it only the explicit place, its +own install folder and its own settings file count. Everything here is read-only: nothing is created, copied or moved. + +The data folder (models, packs, mtp) is found the same way: the person's own setting first, then the app's data folder, then a +``Documents/Strata`` folder that holds models, then the default beside the install. A model that is installed tells where it really is +(its run config names its files), which beats all of these. +""" + +from __future__ import annotations + +import json +import os +from dataclasses import dataclass, field +from pathlib import Path + +# What setup still has to fetch for an installed model whose helper files are gone (setup.py: tools/mtp_fetch.py reads the 31 MTP tensors of +# the pinned checkpoint, 5.214 GB, and packs them into a 0.9 GB draft layer; the picture reader is one 0.9 GB file). +MTP_FETCH_GB = 5.2 +MMPROJ_GB = 0.9 + + +# ----------------------------------------------------------------------------------------------- the places + +def _windows(windows: bool | None) -> bool: + return os.name == "nt" if windows is None else windows + + +def _home(home: Path | None) -> Path: + return Path.home() if home is None else Path(home) + + +def components_dir(environ=None, *, windows: bool | None = None, home: Path | None = None) -> Path: + from strata_control import engine_component + return engine_component.components_dir(environ, windows=windows, home=home) + + +def app_install_dir(environ=None, *, windows: bool | None = None, home: Path | None = None) -> Path: + """The install folder the desktop app uses and the Terminal shares (beside the components folder).""" + return components_dir(environ, windows=windows, home=home).parent / "install" + + +def app_settings_file(environ=None, *, windows: bool | None = None, home: Path | None = None) -> Path: + """The desktop app's own settings file (control-runtime.mjs: beside its components folder).""" + return components_dir(environ, windows=windows, home=home).parent / "app-settings.json" + + +def in_app_mode(environ=None) -> bool: + """True inside the desktop app's own processes: they set STRATA_SETTINGS_FILE and never adopt an older install.""" + return bool((os.environ if environ is None else environ).get("STRATA_SETTINGS_FILE")) + + +def documents_dirs(environ=None, *, windows: bool | None = None, home: Path | None = None) -> list[Path]: + """The person's Documents folder(s): where the desktop app keeps ``Strata`` by default (Windows may redirect Documents to OneDrive).""" + environ = os.environ if environ is None else environ + home = _home(home) + out: list[Path] = [] + explicit = environ.get("STRATA_DOCUMENTS_DIR") + if explicit: + out.append(Path(explicit)) + if _windows(windows): + user = Path(environ.get("USERPROFILE") or str(home)) + out += [user / "Documents"] + for key in ("OneDrive", "OneDriveConsumer", "OneDriveCommercial"): + if environ.get(key): + out.append(Path(environ[key]) / "Documents") + else: + xdg = environ.get("XDG_DOCUMENTS_DIR") + if xdg: + out.append(Path(xdg)) + out.append(home / "Documents") + found: list[Path] = [] + for d in out: + if d not in found: + found.append(d) + return found + + +def _read_json(path: Path) -> dict: + try: + with path.open("rb") as stream: + data = json.loads(stream.read(1024 * 1024).decode("utf-8-sig")) + except (OSError, UnicodeError, ValueError): + return {} + return data if isinstance(data, dict) else {} + + +def _installs_in(settings: dict) -> list[Path]: + items = settings.get("installs") + found: list[Path] = [] + for raw in (items if isinstance(items, list) else [])[:20]: + if isinstance(raw, str) and raw: + p = Path(raw).expanduser() + if p.is_absolute() and p not in found: + found.append(p) + return found + + +def own_settings_file(environ=None, *, windows: bool | None = None, home: Path | None = None) -> Path: + """The settings file this process uses: the app's (STRATA_SETTINGS_FILE), else the command-line Strata's (%APPDATA% / ~/.config).""" + environ = os.environ if environ is None else environ + if environ.get("STRATA_SETTINGS_FILE"): + return Path(environ["STRATA_SETTINGS_FILE"]) + home = _home(home) + if _windows(windows): + return Path(environ.get("APPDATA") or str(home / "AppData" / "Roaming")) / "Strata" / "settings.json" + return Path(environ.get("XDG_CONFIG_HOME") or str(home / ".config")) / "strata" / "settings.json" + + +def has_configs(directory: Path) -> bool: + try: + return directory.is_dir() and any(_is_config(p) for p in directory.iterdir()) + except OSError: + return False + + +def _is_config(path: Path) -> bool: + n = path.name + return n.startswith("strata-") and n.endswith(".json") and not n.endswith(".shared-settings.json") and len(n) > len("strata-.json") + + +def config_files(directory: Path) -> list[Path]: + try: + return sorted(p for p in directory.iterdir() if _is_config(p)) + except OSError: + return [] + + +@dataclass +class Place: + path: Path + source: str # explicit | here | remembered | app + configs: int = 0 + + +def places(*, here: Path | None = None, explicit: str | os.PathLike | None = None, environ=None, windows: bool | None = None, + home: Path | None = None) -> list[Place]: + """Every folder that holds run configs, in the order described above. An explicit folder is the only one.""" + environ = os.environ if environ is None else environ + raw = explicit if explicit is not None else environ.get("STRATA_INSTALL_DIR") + if raw is not None: + p = Path(raw).expanduser() + return [Place(p, "explicit", len(config_files(p)))] + cands: list[tuple[Path, str]] = [] + if here is not None: + cands.append((Path(here), "here")) + own = own_settings_file(environ, windows=windows, home=home) + cands += [(p, "remembered") for p in _installs_in(_read_json(own))] + cands += [(app_install_dir(environ, windows=windows, home=home), "app")] + if not in_app_mode(environ): + cands += [(p, "app") for p in _installs_in(_read_json(app_settings_file(environ, windows=windows, home=home)))] + out: list[Place] = [] + seen: set[str] = set() + for path, source in cands: + try: + key = os.path.normcase(str(path.resolve())) + except OSError: + key = os.path.normcase(str(path)) + if key in seen: + continue + seen.add(key) + n = len(config_files(path)) + if n: + out.append(Place(path.resolve() if path.exists() else path, source, n)) + return out + + +def find_install(*, here: Path | None = None, explicit: str | os.PathLike | None = None, environ=None, windows: bool | None = None, + home: Path | None = None) -> tuple[Path, str]: + """(the main install folder, how it was found). With no run config anywhere it is the app's folder, which is where a new model is + installed (``default``).""" + found = places(here=here, explicit=explicit, environ=environ, windows=windows, home=home) + if found: + return found[0].path, found[0].source + environ = os.environ if environ is None else environ + return app_install_dir(environ, windows=windows, home=home), "default" + + +# ----------------------------------------------------------------------------------------------- the data folder + +def data_dir_choices(*, environ=None, windows: bool | None = None, home: Path | None = None) -> list[tuple[Path, str]]: + """Candidate data folders with their source, best first (not checked for existence).""" + environ = os.environ if environ is None else environ + out: list[tuple[Path, str]] = [] + own = _read_json(own_settings_file(environ, windows=windows, home=home)).get("data_dir") + if isinstance(own, str) and own: + out.append((Path(own), "settings")) + if not in_app_mode(environ): + app = _read_json(app_settings_file(environ, windows=windows, home=home)).get("data_dir") + if isinstance(app, str) and app: + out.append((Path(app), "app")) + for d in documents_dirs(environ, windows=windows, home=home): + out.append((d / "Strata", "documents")) + return out + + +def data_dir(default: Path, *, environ=None, windows: bool | None = None, home: Path | None = None) -> tuple[Path, str]: + """(the data folder, source). The person's own setting wins; then the app's folder, then a ``Documents/Strata`` that holds models; + else `default`.""" + choices = data_dir_choices(environ=environ, windows=windows, home=home) + for path, source in choices: + if source == "settings": + return path, source + if source == "app" and path.is_dir(): + return path, source + if source == "documents" and _holds_models(path): + return path, source + return default, "default" + + +def _holds_models(folder: Path) -> bool: + try: + return any((folder / sub).is_dir() and any((folder / sub).iterdir()) for sub in ("Models", "models", "packs")) + except OSError: + return False + + +# ----------------------------------------------------------------------------------------------- what a model has + +def _flag_value(args: list[str], flag: str) -> str | None: + try: + i = args.index(flag) + except ValueError: + return None + return args[i + 1] if i + 1 < len(args) else None + + +@dataclass +class Footprint: + """The files of one installed model, from its run config: what is named, where, and whether it is there.""" + + config: Path | None = None + weights: list[Path] = field(default_factory=list) # the model's GGUF file(s) + pack: Path | None = None # the prepared expert pack (--pack) + mtp: Path | None = None # the draft layer (--mtp) + mmproj: Path | None = None # the picture reader's file + pictures_wanted: bool = False + data_dir: Path | None = None + + def missing(self) -> list["Missing"]: + out: list[Missing] = [] + gone = [p for p in self.weights if not p.exists()] + if gone: + out.append(Missing("weights", "the model's weights", gone[0], None)) + if self.pack is not None and not self.pack.exists(): + out.append(Missing("pack", "the prepared model data", self.pack, 0.0)) + if self.mtp is not None and not _mtp_ready(self.mtp): + out.append(Missing("mtp", "the draft layer (makes answers faster)", self.mtp, MTP_FETCH_GB)) + if self.pictures_wanted and self.mmproj is not None and not self.mmproj.exists(): + out.append(Missing("mmproj", "the picture reader", self.mmproj, MMPROJ_GB)) + return out + + +@dataclass +class Missing: + kind: str # weights | pack | mtp | mmproj + label: str + path: Path + gb: float | None # to download; None: the whole model (the caller knows its size), 0: prepared locally, nothing to download + + +def _mtp_ready(path: Path) -> bool: + """--mtp names the ``rt`` folder; it is usable when its experts file is there.""" + return (path / "experts.bin").exists() or (path.is_file() and path.exists()) + + +def footprint(cfg: dict, config: Path | None = None) -> Footprint: + args = [str(a) for a in cfg.get("args") or []] + fp = Footprint(config=config) + for flag in ("--native", "--model", "--gguf", "--ple-gguf"): + v = _flag_value(args, flag) + if v and v.lower().endswith(".gguf") and Path(v) not in fp.weights and flag != "--ple-gguf": + fp.weights.append(Path(v)) + if not fp.weights: + fp.weights = [Path(a) for a in args if a.lower().endswith(".gguf")][:1] + pack = _flag_value(args, "--pack") + fp.pack = Path(pack) if pack else None + mtp = _flag_value(args, "--mtp") + fp.mtp = Path(mtp) if mtp and not mtp.isdigit() else None + vision = cfg.get("vision") + if isinstance(vision, dict) and vision.get("mmproj"): + fp.mmproj = Path(str(vision["mmproj"])) + fp.pictures_wanted = True + elif "--vision" in args: + fp.pictures_wanted = True + fp.data_dir = data_dir_of(fp) + return fp + + +def data_dir_of(fp: Footprint) -> Path | None: + """The data folder an installed model lives in: the parent of its ``packs`` folder, else the one holding ``models``.""" + if fp.pack is not None: + return fp.pack.parent.parent + if fp.mtp is not None and fp.mtp.name.lower() == "rt": + return fp.mtp.parent.parent + for w in fp.weights: + for parent in w.parents: + if parent.name.lower() == "models": + return parent.parent + return None + + +def prepared_beside(gguf: Path) -> dict: + """What was already prepared for a model file that sits in ``/Models//`` (or ``/models//``): its pack, the draft + layer and the picture reader, found by the layout setup and the desktop app both use. Reads only; absent items are None.""" + gguf = Path(gguf) + folder = gguf.parent + out: dict = {"data_dir": None, "pack": None, "mtp": None, "mmproj": None} + for parent in folder.parents: + if parent.name.lower() == "models": + data = parent.parent + out["data_dir"] = data + tag = folder.name.lower() if folder != parent else "" + if tag and (data / "packs" / tag).is_dir(): + out["pack"] = data / "packs" / tag + if (data / "mtp" / "rt" / "experts.bin").is_file(): + out["mtp"] = data / "mtp" / "rt" + for cand in sorted(parent.glob("mmproj*.gguf")) + sorted(folder.glob("mmproj*.gguf")): + out["mmproj"] = cand + break + break + return out + + +def run_config_for(gguf: Path, install_places: list[Place]) -> Path | None: + """The run config in any known install folder that names this GGUF file (the model is already set up, with its own choices).""" + want = os.path.normcase(str(Path(gguf))) + for place in install_places: + for cfg_path in config_files(place.path): + data = _read_json(cfg_path) + args = [str(a) for a in data.get("args") or []] + if any(os.path.normcase(a) == want for a in args if a.lower().endswith(".gguf")): + return cfg_path + return None diff --git a/strata_control/models.py b/strata_control/models.py index 9dc12a2ad..a00901ea5 100644 --- a/strata_control/models.py +++ b/strata_control/models.py @@ -4,9 +4,10 @@ The setup installer writes strata-*.json beside setup.py. A configured model is not automatically a compatible or loadable model: component presence and native engine compatibility are reported separately. -Finding an existing install: a terminal archive is a new folder with no run configs in it. The folders earlier -Strata installs remembered in the user's settings file (``installs``, written by setup.py next to ``data_dir``) are -searched for run configs, read-only, and the first that has any is used where it is. Nothing is copied or moved: +Finding an existing install: a terminal archive is a new folder with no run configs in it. The places where Strata +installs keep run configs (the Terminal's own folder, the folders earlier installs remembered in the user's settings file, +and the desktop app's install folder, which the Terminal shares) are searched read-only by strata_control/discovery.py, +the same function the launch screens use; the first place that has any is the main one. Nothing is copied or moved: the configs name their engine and model files by absolute path, so the old install and its data folder keep serving the models. """ @@ -20,6 +21,7 @@ import re import stat from pathlib import Path +from strata_control import discovery from strata_control.doctor import inspect as inspect_components from strata_control.gguf_preflight import inspect_metadata from strata_control.profiles import ProfileError @@ -37,33 +39,16 @@ SUPPORTED_SELECTIONS = frozenset({"qwen/Q2_0", "qwen/IQ2_XS", "qwen/IQ3_XXS", "q def settings_path() -> Path: - """The user's Strata settings file: the same place setup.py writes it (%APPDATA%\\Strata\\settings.json, or + """The user's Strata settings file: the same place setup.py writes it (%APPDATA%/Strata/settings.json, or $XDG_CONFIG_HOME/strata/settings.json). The desktop app sets STRATA_SETTINGS_FILE to a file of its own (in its data folder), so it never reads an older command-line install's list of Strata folders.""" - own = os.environ.get("STRATA_SETTINGS_FILE") - if own: - return Path(own) - if os.name == "nt": - return Path(os.environ.get("APPDATA") or Path.home() / "AppData" / "Roaming") / "Strata" / "settings.json" - return Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") / "strata" / "settings.json" + return discovery.own_settings_file() def remembered_installs() -> list[Path]: """Earlier Strata folders recorded in the settings file, newest first. Read-only; a missing or unreadable file is an empty list.""" - try: - with settings_path().open("rb") as stream: - data = json.loads(stream.read(1024 * 1024).decode("utf-8-sig")) - except (OSError, UnicodeError, ValueError): - return [] - items = data.get("installs") if isinstance(data, dict) else None - found: list[Path] = [] - for raw in (items if isinstance(items, list) else [])[:20]: - if isinstance(raw, str) and raw: - path = Path(raw).expanduser() - if path.is_absolute() and path not in found: - found.append(path) - return found + return discovery._installs_in(discovery._read_json(settings_path())) # noqa: SLF001 def _has_configs(directory: Path) -> bool: @@ -74,20 +59,25 @@ def _has_configs(directory: Path) -> bool: def resolve_install(override: str | None = None) -> tuple[Path, str]: - """(the install folder, how it was found): explicit (an argument or STRATA_INSTALL_DIR), here (this folder has run - configs), remembered (an earlier install from the settings file) or default (this folder, with none yet).""" + """(the main install folder, how it was found): explicit (an argument or STRATA_INSTALL_DIR), here (this folder has run + configs), remembered (an earlier install from the settings file), app (the desktop app's install folder, shared with the + Terminal) or default (the app's folder, with no model in it yet).""" raw = override or os.environ.get("STRATA_INSTALL_DIR") if raw is not None: path = Path(raw).expanduser() if not path.is_absolute() or not path.is_dir(): raise ProfileError("install directory must be an existing absolute directory", 5) return path.resolve(), "explicit" - if _has_configs(ROOT): - return ROOT, "here" - for earlier in remembered_installs(): - if earlier != ROOT and _has_configs(earlier): - return earlier.resolve(), "remembered" - return ROOT, "default" + return discovery.find_install(here=ROOT) + + +def all_installs(override: str | None = None) -> list[tuple[Path, str]]: + """Every folder with run configs, main one first: what `strata models list` shows and what the launch screens read.""" + directory, source = resolve_install(override) + if source == "explicit": + return [(directory, source)] + found = [(p.path, p.source) for p in discovery.places(here=ROOT)] + return found or [(directory, source)] def install_dir(override: str | None = None) -> Path: @@ -180,15 +170,28 @@ def _inspect_one(model_id: str, path: Path) -> tuple[dict, dict]: def list_installed(override: str | None = None) -> dict: directory, source = resolve_install(override) + places = all_installs(override) models = [] - for model_id, path in _paths(directory): + for place, place_source in places: try: - summary, _ = _inspect_one(model_id, path) + entries = _paths(place) if place.is_dir() else [] except ProfileError as exc: - summary = {"id": model_id, "source_config": str(path), "component_status": "invalid", - "error": str(exc), "execution_compatibility": "unverified"} - models.append(summary) + models.append({"id": place.name, "install_dir": str(place), "component_status": "invalid", "error": str(exc), + "execution_compatibility": "unverified"}) + continue + for model_id, path in entries: + if any(m.get("id") == model_id for m in models): + continue # the same model in a later place: the main one wins + try: + summary, _ = _inspect_one(model_id, path) + except ProfileError as exc: + summary = {"id": model_id, "source_config": str(path), "component_status": "invalid", + "error": str(exc), "execution_compatibility": "unverified"} + summary["install_dir"] = str(place) + summary["install_dir_source"] = place_source + models.append(summary) return {"scope": "local_run_configs", "install_dir": str(directory), "install_dir_source": source, + "install_dirs": [str(p) for p, _ in places], "models": models, "model_count": len(models), "discovery_limit": MAX_CONFIGS} @@ -196,9 +199,15 @@ def list_installed(override: str | None = None) -> dict: def inspect(model_id: str, override: str | None = None) -> dict: if not isinstance(model_id, str) or not MODEL_ID.fullmatch(model_id): raise ProfileError("model ID must be a setup config name without a path") - directory = install_dir(override) - choices = dict(_paths(directory)) - path = choices.get(model_id) + directory, path = None, None + for place, _src in all_installs(override): + try: + choices = dict(_paths(place)) if place.is_dir() else {} + except ProfileError: + continue + if model_id in choices: + directory, path = place, choices[model_id] + break if path is None: raise ProfileError(f"model {model_id!r} is not configured in this installation", 5) summary, report = _inspect_one(model_id, path) @@ -250,6 +259,40 @@ def _candidate_selection(shards: list[Path]) -> dict | None: return None +def _candidate_by_header(shards: list[Path], tensors: int, payload: int, experts: set[str]) -> dict | None: + """A file set whose NAME setup does not know (renamed, or downloaded elsewhere) but whose header is exactly a known one: the same tensor + count, the same payload size and the same expert formats as a row of the fingerprint table.""" + try: + catalog = json.loads((ROOT / "data" / "gguf_fingerprints.json").read_text(encoding="utf-8")) + except (OSError, UnicodeError, ValueError): + return None + for selection in sorted(SUPPORTED_SELECTIONS): + row = catalog.get(selection) if isinstance(catalog, dict) else None + if not isinstance(row, dict) or row.get("tensors") != tensors or row.get("bytes") != payload: + continue + expected = row.get("expert_bytes") + if not isinstance(expected, dict) or set(expected) != experts: + continue + family, model = selection.split("/", 1) + return {"family": family, "model": model, "basis": "header_size_and_tensors", + "expected_expert_types": sorted(expected), "expected_tensor_count": tensors, + "setup_args_hint": ["--family", family, "--model", model, "--gguf-dir", str(shards[0].parent)]} + return None + + +def _already_set_up(shards: list[Path], candidate: dict | None) -> dict | None: + """Is this model already installed (a run config of any known install names the file) and what was prepared beside it (pack, draft + layer, picture reader)? Read-only; None when the file set is not one Strata knows.""" + if not shards or candidate is None: + return None + first = shards[0] + cfg = discovery.run_config_for(first, discovery.places(here=ROOT)) + beside = discovery.prepared_beside(first) + if cfg is None and not any(beside.get(k) for k in ("pack", "mtp", "mmproj")): + return {"config": None, "data_dir": str(beside["data_dir"]) if beside.get("data_dir") else None, "pack": None, "mtp": None, "mmproj": None} + return {"config": str(cfg) if cfg else None, **{k: (str(v) if v else None) for k, v in beside.items()}} + + UNSUPPORTED_MODEL_TYPE = ("This model type isn't supported by the Strata engine. " "Add it as a Custom model API (llama-server) instead.") # The picture readers Strata ships a pinned copy of (setup downloads and checks one of these next to the model). @@ -363,6 +406,7 @@ def plan_import(raw_path: str) -> dict: metadata = None observed_experts: set[str] = set() observed_tensor_count = 0 + observed_payload = 0 if files and not issues: for item in shards: try: @@ -374,6 +418,7 @@ def plan_import(raw_path: str) -> dict: metadata = inspected observed_experts.update(inspected["expert_tensor_types"]) observed_tensor_count += inspected["tensor_count_scanned"] + observed_payload += int(inspected.get("payload_bytes_required") or 0) if inspected["architecture"] is not None and inspected["architecture"] != "qwen4exp": issues.append(f"{item.name}: unsupported_architecture") if metadata is not None: @@ -390,6 +435,8 @@ def plan_import(raw_path: str) -> dict: elif projectors: messages.extend(p["plain"] for p in projectors) candidate = _candidate_selection(shards) + if candidate is None and metadata is not None and not issues: + candidate = _candidate_by_header(shards, observed_tensor_count, observed_payload, observed_experts) if candidate is None: issues.append("unrecognized_or_unsupported_model_variant") elif metadata is not None: @@ -402,7 +449,7 @@ def plan_import(raw_path: str) -> dict: "observed_expert_types": sorted(observed_experts), "observed_tensor_count": observed_tensor_count, "ignored_auxiliary_files": sorted(ignored_auxiliary), - "projectors": projectors, "messages": messages, + "projectors": projectors, "messages": messages, "already_set_up": _already_set_up(shards, candidate), "reads_pictures": any(p["status"] in ("ok", "other_name") for p in projectors) and not any(i.endswith("unsupported_architecture") for i in issues), "file_set_status": "plausible" if not issues else "invalid_or_incomplete", diff --git a/tests/terminal/test_app_isolation.py b/tests/terminal/test_app_isolation.py index 1c9f5baf5..1534d98b0 100644 --- a/tests/terminal/test_app_isolation.py +++ b/tests/terminal/test_app_isolation.py @@ -31,7 +31,8 @@ class AppIsolation(unittest.TestCase): (base / "xdg" / "strata" / "settings.json").write_text(json.dumps({"installs": [str(self.old)], "data_dir": str(base / "T-data")})) self.app_data = base / "app" / "Strata" self.app_data.mkdir(parents=True) - self.env = {"APPDATA": str(base / "appdata"), "XDG_CONFIG_HOME": str(base / "xdg")} + self.env = {"APPDATA": str(base / "appdata"), "XDG_CONFIG_HOME": str(base / "xdg"), + "LOCALAPPDATA": str(base / "local"), "XDG_DATA_HOME": str(base / "xdg-data"), "STRATA_COMPONENTS_DIR": ""} self.base = base self.terminal = base / "terminal" self.terminal.mkdir() @@ -48,7 +49,10 @@ class AppIsolation(unittest.TestCase): os.environ.pop("STRATA_INSTALL_DIR", None) self.assertEqual(models.settings_path(), self.app_data / "app-settings.json") self.assertEqual(models.remembered_installs(), []) - self.assertEqual(models.resolve_install(), (self.terminal, "default")) + # no older install is adopted: with nothing of its own yet the app gets its own install folder (beside its components folder) + found, how = models.resolve_install() + self.assertEqual(how, "default") + self.assertNotEqual(found, self.old.resolve()) self.assertEqual(strata_setup.settings_path(), self.app_data / "app-settings.json") self.assertEqual(strata_setup.load_settings(), {}) # the installer's own list of "other installs" has no sibling scan and no shared settings diff --git a/tests/terminal/test_discovery.py b/tests/terminal/test_discovery.py new file mode 100644 index 000000000..5aff41bcd --- /dev/null +++ b/tests/terminal/test_discovery.py @@ -0,0 +1,284 @@ +"""Tester report 2026-10-18, T03 / T04 / T05: existing models are found the same way by the launch screens, `strata models list`, the +own-files import and `strata launch` (strata_control/discovery.py), with a mock of the desktop app's folders and a Gyro-S file header +(TQK6 / TQK7 experts, the one-file layout).""" + +from __future__ import annotations + +import io +import json +import os +import shutil +import struct +import tempfile +import unittest +from contextlib import redirect_stdout +from pathlib import Path +from unittest import mock + +from strata_cli.launch import modelstate, settings as S +from strata_cli.launch.catalog import Catalog +from strata_cli.launch.runner import Runner +from strata_control import discovery, models +from tests.terminal.launch_support import HW_12 + +GYRO_S = "Qwen3.8-Flash-Next-Gyro-S-TQ1_0.gguf" +GYRO_M = "Qwen3.8-Flash-Next-Gyro-M-TQ2_0.gguf" +TQ2_T, TQK6, TQK7 = 144, 145, 146 + + +def string(value: str) -> bytes: + raw = value.encode() + return struct.pack(" bytes: + """A tiny qwen4exp GGUF: (name, ggml type id, elements, payload bytes) per tensor, laid out one after another.""" + pairs = [("general.architecture", 8, string("qwen4exp")), ("tokenizer.ggml.model", 8, string("gpt2")), + ("tokenizer.ggml.tokens", 9, struct.pack(" bytes: + return gguf_with([("blk.0.ffn_gate_exps.weight", TQK6, 256, 52), ("blk.0.ffn_down_exps.weight", TQK7, 256, 60)]) + + +def gyro_m_bytes() -> bytes: + return gguf_with([("blk.0.ffn_gate_exps.weight", TQ2_T, 256, 68)]) + + +class GuiLayout(unittest.TestCase): + """The desktop app's folders, as the Windows (and Linux) app lays them out, plus an empty Terminal folder.""" + + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix="strata-discovery-") + self.addCleanup(self.tmp.cleanup) + base = Path(self.tmp.name) + self.base = base + self.env = {"LOCALAPPDATA": str(base / "local"), "XDG_DATA_HOME": str(base / "local"), "APPDATA": str(base / "roaming"), + "XDG_CONFIG_HOME": str(base / "config"), "USERPROFILE": str(base / "home"), "HOME": str(base / "home"), + "XDG_STATE_HOME": str(base / "state"), "XDG_DOCUMENTS_DIR": ""} + patch = mock.patch.dict(os.environ, self.env) + patch.start() + self.addCleanup(patch.stop) + for key in ("STRATA_INSTALL_DIR", "STRATA_SETTINGS_FILE", "STRATA_COMPONENTS_DIR", "STRATA_DOCUMENTS_DIR"): + patch = mock.patch.dict(os.environ) + patch.start() + self.addCleanup(patch.stop) + os.environ.pop(key, None) + self.terminal = base / "terminal" + (self.terminal / "data").mkdir(parents=True) + shutil.copy(Path(models.__file__).resolve().parents[1] / "data" / "gguf_fingerprints.json", self.terminal / "data") + p = mock.patch.object(models, "ROOT", self.terminal) + p.start() + self.addCleanup(p.stop) + self.install = discovery.app_install_dir() + self.docs = base / "home" / "Documents" / "Strata" + gyro = self.docs / "Models" / "gyro-Gyro-S" + gyro.mkdir(parents=True) + self.gguf = gyro / GYRO_S + self.gguf.write_bytes(gyro_s_bytes()) + (gyro / (GYRO_S + ".done")).write_text("done") + (self.docs / "packs" / "gyro-gyro-s" / "tokenizer").mkdir(parents=True) + (self.docs / "packs" / "gyro-gyro-s" / "dense.bin").write_bytes(b"x") + (self.docs / "mtp" / "rt").mkdir(parents=True) + (self.docs / "mtp" / "rt" / "experts.bin").write_bytes(b"x") + self.mmproj = self.docs / "Models" / "mmproj-Qwen3.8-Flash-Next-BF16.gguf" + self.mmproj.write_bytes(b"x") + self.install.mkdir(parents=True) + engine = base / "engine.exe" + engine.write_bytes(b"e") + tok = self.docs / "packs" / "gyro-gyro-s" / "tokenizer" + (tok / "vocab.json").write_text("{}") + self.cfg = {"exe": "component:strata-engine", "tokenizer": str(tok), "model_name": "qwen3.8-flash-next-gyro-gyro-s", + "args": ["--pack", str(self.docs / "packs" / "gyro-gyro-s"), "--native", str(self.gguf), "--ple-gguf", str(self.gguf), + "--mtp", str(self.docs / "mtp" / "rt"), "--max-context", "131072", "--kv", "int8", "--vision"], + "vision": {"exe": str(engine), "mmproj": str(self.mmproj), "model": str(self.gguf), "gpu": False}} + self.config_path = self.install / "strata-gyro-gyro-s.json" + self.config_path.write_text(json.dumps(self.cfg)) + (self.install.parent / "app-settings.json").write_text(json.dumps({"data_dir": str(self.docs), "installs": [str(self.install)]})) + + def catalog(self) -> Catalog: + return Catalog(hardware=HW_12) + + # ---- T04: one answer for list and the wizard + def test_models_list_finds_the_app_install_by_default(self): + result = models.list_installed() + self.assertEqual([m["id"] for m in result["models"]], ["gyro-gyro-s"]) + self.assertEqual(Path(result["install_dir"]).resolve(), self.install.resolve()) + self.assertEqual(result["install_dir_source"], "app") + + def test_list_and_the_wizard_agree_on_the_install(self): + cat = self.catalog() + self.assertEqual(cat.root.resolve(), models.resolve_install()[0].resolve()) + self.assertIsNotNone(cat.installed_config("gyro-gyro-s")) + row = next(r for r in cat.rows() if (r.family, r.model) == ("gyro", "Gyro-S")) + self.assertTrue(row.installed) + + def test_cli_list_names_where_it_looked_when_nothing_is_found(self): + self.config_path.unlink() + out = io.StringIO() + from strata_cli.cli import main + with redirect_stdout(out): + code = main(["models", "list"]) + self.assertEqual(code, 0) + self.assertIn("No installed models found. Looked in:", out.getvalue()) + self.assertIn("--install-dir", out.getvalue()) + + def test_an_explicit_install_dir_is_the_only_place(self): + other = self.base / "other" + other.mkdir() + self.assertEqual(models.list_installed(str(other))["models"], []) + self.assertEqual([p.source for p in discovery.places(explicit=str(other))], ["explicit"]) + + def test_a_remembered_older_install_and_the_app_install_are_both_listed(self): + old = self.base / "T-Strata" + old.mkdir() + (old / "strata-qwen-iq3_xxs.json").write_text(json.dumps({"exe": "e", "tokenizer": "t", "args": []})) + settings = Path(os.environ["APPDATA"]) / "Strata" / "settings.json" + settings.parent.mkdir(parents=True) + settings.write_text(json.dumps({"installs": [str(old)]})) + got = models.list_installed() + self.assertEqual(sorted(m["id"] for m in got["models"]), ["gyro-gyro-s", "qwen-iq3_xxs"]) + self.assertEqual(got["install_dir_source"], "remembered") + + def test_inside_the_app_an_older_install_is_still_not_adopted(self): + old = self.base / "T-Strata" + old.mkdir() + (old / "strata-x.json").write_text(json.dumps({"exe": "e", "tokenizer": "t", "args": []})) + app_settings = self.install.parent / "app-settings.json" + with mock.patch.dict(os.environ, {"STRATA_SETTINGS_FILE": str(app_settings)}): + settings = Path(os.environ["APPDATA"]) / "Strata" / "settings.json" + settings.parent.mkdir(parents=True) + settings.write_text(json.dumps({"installs": [str(old)]})) + self.assertEqual([m["id"] for m in models.list_installed()["models"]], ["gyro-gyro-s"]) + + # ---- T03: the own-files import knows Gyro + def test_import_recognises_a_gyro_s_folder_and_its_prepared_files(self): + plan = models.plan_import(str(self.gguf.parent)) + self.assertEqual(plan["issues"], []) + self.assertEqual((plan["candidate_selection"]["family"], plan["candidate_selection"]["model"]), ("gyro", "Gyro-S")) + self.assertEqual(sorted(plan["observed_expert_types"]), ["TQK6", "TQK7"]) + done = plan["already_set_up"] + self.assertEqual(Path(done["config"]).resolve(), self.config_path.resolve()) + self.assertEqual(Path(done["pack"]).resolve(), (self.docs / "packs" / "gyro-gyro-s").resolve()) + self.assertEqual(Path(done["mtp"]).resolve(), (self.docs / "mtp" / "rt").resolve()) + self.assertEqual(Path(done["data_dir"]).resolve(), self.docs.resolve()) + + def test_import_recognises_the_file_itself(self): + plan = models.plan_import(str(self.gguf)) + self.assertEqual(plan["issues"], []) + self.assertEqual(plan["candidate_selection"]["model"], "Gyro-S") + + def test_import_recognises_gyro_m(self): + folder = self.base / "gyro-m" + folder.mkdir() + (folder / GYRO_M).write_bytes(gyro_m_bytes()) + plan = models.plan_import(str(folder)) + self.assertEqual(plan["issues"], []) + self.assertEqual(plan["candidate_selection"]["model"], "Gyro-M") + + def test_import_still_refuses_gyro_s_with_the_wrong_expert_format(self): + folder = self.base / "wrong" + folder.mkdir() + (folder / GYRO_S).write_bytes(gyro_m_bytes()) # TQ2_T experts under the Gyro-S name + plan = models.plan_import(str(folder)) + self.assertIn("missing_or_unexpected_expert_quantization", plan["issues"]) + + def test_a_renamed_file_is_known_by_its_header(self): + data = gyro_s_bytes() + folder = self.base / "renamed" + folder.mkdir() + (folder / "my-copy.gguf").write_bytes(data) + fake_root = self.base / "fake-root" + (fake_root / "data").mkdir(parents=True) + payload = 52 + 60 + 12 # 32-byte aligned offsets: 64 + 64 -> see below + table = {"gyro/Gyro-S": {"arch": "qwen4exp", "tensors": 2, "bytes": None, "expert_bytes": {"TQK6": 52, "TQK7": 60}, + "files": [GYRO_S], "repo": "x"}} + first = models.plan_import(str(folder / "my-copy.gguf")) + self.assertIsNone(first["candidate_selection"]) # not in the real table under another name + table["gyro/Gyro-S"]["bytes"] = first["metadata"]["payload_bytes_required"] + (fake_root / "data" / "gguf_fingerprints.json").write_text(json.dumps(table)) + with mock.patch.object(models, "ROOT", fake_root): + plan = models.plan_import(str(folder / "my-copy.gguf")) + self.assertEqual(plan["candidate_selection"]["basis"], "header_size_and_tensors") + self.assertEqual(plan["candidate_selection"]["model"], "Gyro-S") + self.assertEqual(payload, 124) + + # ---- T05: where it is, what is missing + def test_the_folder_defaults_to_where_the_model_is(self): + cat = self.catalog() + self.assertEqual(cat.data_dir().resolve(), self.docs.resolve()) + self.assertEqual(cat.data_dir_for("gyro", "Gyro-S").resolve(), self.docs.resolve()) + self.assertEqual(cat.data_dir_choice()[1], "app") + + def test_an_own_setting_beats_the_apps_folder(self): + mine = self.base / "mine" + mine.mkdir() + settings = Path(os.environ["APPDATA"]) / "Strata" / "settings.json" + settings.parent.mkdir(parents=True, exist_ok=True) + settings.write_text(json.dumps({"data_dir": str(mine)})) + self.assertEqual(self.catalog().data_dir().resolve(), mine.resolve()) + + def test_installed_with_everything_there_is_already_downloaded(self): + st = modelstate.assess(self.catalog(), "gyro", "Gyro-S", vision="cpu") + self.assertTrue(st.installed) + self.assertEqual(st.remaining_gb, 0.0) + self.assertIn("Already downloaded", st.headline()) + self.assertIn("Nothing more to download", st.headline()) + + def test_only_the_draft_layer_missing_says_so_with_its_size(self): + (self.docs / "mtp" / "rt" / "experts.bin").unlink() + cat = self.catalog() + st = modelstate.assess(cat, "gyro", "Gyro-S", vision="cpu") + self.assertEqual([p.kind for p in st.parts], ["mtp"]) + self.assertAlmostEqual(st.remaining_gb, discovery.MTP_FETCH_GB) + text = st.headline() + self.assertIn("Only the draft layer is missing (5.2 GB)", text) + self.assertNotIn("download the model", text.lower()) + title, body = st.question() + self.assertIn("draft layer", title) + self.assertIn("not downloaded again", body) + + def test_draft_layer_and_picture_reader_are_itemised(self): + (self.docs / "mtp" / "rt" / "experts.bin").unlink() + self.mmproj.unlink() + st = modelstate.assess(self.catalog(), "gyro", "Gyro-S", vision="cpu") + self.assertEqual(sorted(p.kind for p in st.parts), ["mmproj", "mtp"]) + self.assertIn("the draft layer (5.2 GB) and the picture reader (0.9 GB)", st.headline()) + self.assertAlmostEqual(st.remaining_gb, 5.2 + 0.9) + + def test_pictures_off_never_asks_for_the_reader(self): + self.mmproj.unlink() + st = modelstate.assess(self.catalog(), "gyro", "Gyro-S", vision="off") + self.assertEqual(st.parts, []) + + def test_weights_gone_downloads_the_model_again_and_says_so(self): + self.gguf.unlink() + st = modelstate.assess(self.catalog(), "gyro", "Gyro-S", vision="off") + self.assertFalse(st.installed) + self.assertGreater(st.remaining_gb, 50) + + def test_runner_counts_only_the_missing_helper_and_gives_setup_the_apps_folder(self): + (self.docs / "mtp" / "rt" / "experts.bin").unlink() + cat = self.catalog() + r = Runner(S.LaunchSettings(family="gyro", model="Gyro-S", context=131072, kv="int8", vision="cpu"), catalog=cat) + self.assertAlmostEqual(r.download_gb(), 5.2) + argv = r.setup_args() + self.assertEqual(Path(argv[argv.index("--data-dir") + 1]).resolve(), self.docs.resolve()) + + def test_own_files_use_the_prepared_pack_and_folder(self): + r = Runner(S.LaunchSettings(model_path=str(self.gguf.parent), family="gyro", model="Gyro-S", vision="off"), catalog=self.catalog()) + self.assertEqual(r.download_gb(), 0.0) + self.assertEqual(Path(r.existing_data_dir()).resolve(), self.docs.resolve()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/terminal/test_launch_tui.py b/tests/terminal/test_launch_tui.py index 51ec302ee..d523c1f1d 100644 --- a/tests/terminal/test_launch_tui.py +++ b/tests/terminal/test_launch_tui.py @@ -304,7 +304,7 @@ class FolderStep(Base): await press(pilot, "enter", "enter") t = text_of(app) self.assertIn("Where should the model files be stored?", t) - self.assertIn("Needs about", t) + self.assertIn("Not downloaded yet: about 76 GB", t) self.assertIn("Free space in that folder's drive", t) f = app.screen.query_one("#folderpath") f.value = "" diff --git a/tools/quantscope.py b/tools/quantscope.py index 27af4ae54..de0da2470 100644 --- a/tools/quantscope.py +++ b/tools/quantscope.py @@ -121,6 +121,7 @@ GGML_TYPES = { 23: "IQ4_XS", 24: "I8", 25: "I16", 26: "I32", 27: "I64", 28: "F64", 29: "IQ1_M", 30: "BF16", 34: "TQ1_0", 35: "TQ2_0", 39: "MXFP4", 42: "Q2_0", # ISTA-DASLab GSQ-RCO 2-bit (Strata, llama.cpp fork type) + 143: "PTQ1_0", 144: "TQ2_T", 145: "TQK6", 146: "TQK7", # agentionai Gyro (tools/gguf_reader.py has the block sizes) }