Terminal: one discovery of existing installs and model files for the screens, models list, own-files import and launch; Gyro-S/M GGUFs are known; installed models show what is really missing (T03, T04, T05)

This commit is contained in:
Niko1221 committed 2026-10-18 22:22:31 +02:00
1 parent 1486151d4d
commit 23b08c7620
13 files changed
+1120 -98

No files matched your search

+53 -7
View File
@@ -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"
]
}
+25 -4
View File
@@ -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
+30 -4
View File
@@ -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
+58 -10
View File
@@ -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))
+163
View File
@@ -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
+43 -15
View File
@@ -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):
+24 -15
View File
@@ -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
+345
View File
@@ -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-<tag>.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: ``<components>/../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 ``<data>/Models/<tag>/`` (or ``<data>/models/<tag>/``): 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
+87 -40
View File
@@ -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",
+6 -2
View File
@@ -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
+284
View File
@@ -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("<Q", len(raw)) + raw
def gguf_with(tensors: list[tuple[str, int, int, int]]) -> 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("<IQ", 8, 1) + string("a")),
("tokenizer.ggml.merges", 9, struct.pack("<IQ", 8, 1) + string("a b"))]
header = (b"GGUF" + struct.pack("<IQQ", 3, len(tensors), len(pairs)) +
b"".join(string(k) + struct.pack("<I", kind) + value for k, kind, value in pairs))
offset = 0
for name, type_id, elements, nbytes in tensors:
header += string(name) + struct.pack("<IQIQ", 1, elements, type_id, offset)
offset += -(-nbytes // 32) * 32
header += b"\x00" * (-len(header) % 32)
return header + b"\x00" * offset
def gyro_s_bytes() -> 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()
+1 -1
View File
@@ -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 = ""
+1
View File
@@ -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)
}