Niko1221/Strata
1
0 Fork 0
mirror of https://github.com/Niko1221/Strata.git synced 2026-11-07 14:21:06 -10:00
Code Issues Packages Projects Releases 1 Wiki Activity

DeepSeek engine updates move to the explicit update; a start only repairs a stale path (replaces the start-time feed check)

Browse Source 
Starting never reaches the feed again (test_no_acquisition passes unchanged). What a start does now, locally: a saved profile whose DeepSeek engine
folder an update removed is pointed at the newest installed package of its own kind (strata_control/dsv41_paths.py, called from runtime.start), which
fixes the 'engine could not start: file not found' after a strata launch updated the engine beside the app.

The explicit update (strata update, the app's Update engine, updater.run_update with the engine) now also updates an installed DeepSeek engine
(updater.update_dsv41): the feed's newer package of the same kind is downloaded, checked (signature, every file), unpacked next to the old one,
started once (--version), and only then every saved run config is pointed at it; older folders go, except one a running engine uses. update --check
reports it; the check kept after any update or check also feeds engine_component.cached_offers / status()['update_available'], which the app reads
without the network to show one card. strata launch's own upgrade (dsv41_engine.py) is left as it was.

Tests: tests/terminal/test_dsv41_update_path.py (the old keep_current tests, adapted, plus a start that makes no request).
This commit is contained in:
Niko1221 committed 2026-10-31 23:15:10 +01:00
1 parent d4d0c30da5
commit 94edf723ca
8 files changed
+639 -451
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

strata_cli/update_cmd.py
+14
View File
@@ -132,6 +132,17 @@ def _line(name: str, result: dict) -> str:
if status == "failed":
return f"{name:<8}{before or 'none'} NOT changed: {result.get('error')}"
return f"{name:<8}{before or 'none'} {result.get('note') or status}"
if result.get("component") == "dsv41":
if status == "none-installed":
return f"{name:<8}no DeepSeek engine installed"
before, after = result.get("installed_before"), result.get("installed_after")
if status == "updated":
return f"{name:<8}{before} -> {after} updated (tested: it starts). {result.get('note') or ''}".rstrip()
if status == "current":
return f"{name:<8}{before} up to date"
if status == "failed":
return f"{name:<8}{before or '?'} NOT changed: {result.get('error')}"
return f"{name:<8}{before or '?'} {result.get('note') or status}"
if result.get("component") == "control":
if status == "updated":
return f"{name:<8}{result.get('installed_before') or '?'} -> {result.get('installed_after')} updated. The next `strata` command already runs it; the app switches on its next start"
@@ -174,6 +185,9 @@ def _print_check(report: dict) -> None:
print(" Control not part of this feed")
else:
print(f" Control {ctl['installed'] or '?'}: {ctl['status']}" + (f" ({ctl['offered']})" if ctl["offered"] else ""))
ds = report.get("dsv41") or {}
if ds.get("status") not in (None, "none-installed"):
print(f" DeepSeek {ds.get('installed')}: " + (f"update available -> {ds['offered']} ({_mib(ds['download_bytes'])})" if ds["status"] == "available" else ds["status"]))
if app["status"] != "not-in-feed":
print(f" App {app['installed'] or 'unknown'} -> {app['offered']}: {app['status']}")
if report.get("serving"):
strata_control/dsv41_engine.py
+5 -241
View File
@@ -7,11 +7,8 @@ there is no test start here (setup's own 16-token question follows the pack).
"""
from __future__ import annotations
import contextlib
import json
import os
import shutil
import time
from pathlib import Path
from typing import Callable
@@ -174,8 +171,8 @@ def install(dest: Path, source: str, *, hardware: dict | None = None, emit: Call
shutil.rmtree(stage_root, ignore_errors=True)
if os.name != "nt":
(target / exe_name()).chmod(0o755)
for old in dest.iterdir(): # one generation: the older folders of this engine go (one a running engine uses stays)
if old.is_dir() and old.name != target.name and not in_use(old):
for old in dest.iterdir(): # one generation: the older folders of this engine go
if old.is_dir() and old.name != target.name:
shutil.rmtree(old, ignore_errors=True)
emit(f"DeepSeek engine {pkg.version} installed: {target}")
shutil.rmtree(work, ignore_errors=True) # (on a failure the partial download stays in `work`: the next try resumes it)
@@ -228,17 +225,7 @@ def _test_start(exe: Path, timeout: float = 120.0) -> tuple[bool, str]:
def upgrade(dest: Path, source: str, *, installed: Path | None = None, hardware: dict | None = None,
emit: Callable[[str], None] = print, cancel=None, same_kind: bool = False) -> Path | None:
"""`_upgrade` (below), one at a time: a second caller (a `strata launch`, an app start, the background fetch) while one runs gets None at once, so two
of them never write the same download and work folders. `same_kind`: only a package of the installed one's own platform, backend and variant."""
with _fetch_lock(Path(dest).parent) as got:
if not got:
return None
return _upgrade(dest, source, installed=installed, hardware=hardware, emit=emit, cancel=cancel, same_kind=same_kind)
def _upgrade(dest: Path, source: str, *, installed: Path | None = None, hardware: dict | None = None,
emit: Callable[[str], None] = print, cancel=None, same_kind: bool = False) -> Path | None:
emit: Callable[[str], None] = print, cancel=None) -> Path | None:
"""The installed DeepSeek engine package in `dest` (engine-dsv41/) is older than the signed feed's `dsv41` member for this PC: fetch the newer one,
check it as `install` does (signature, hashes, every file), put it NEXT TO the old one, start it once (`--version`), and only then remove the older
folders (one generation). Returns the new strata-dsv41, or None: nothing newer, the feed cannot be read (silent: a launch never waits on the
@@ -272,8 +259,6 @@ def _upgrade(dest: Path, source: str, *, installed: Path | None = None, hardware
pkg, _reasons = feedmod.select_engine(list(feed.dsv41), hw)
if pkg is None or version_key(pkg.version) <= have:
return None
if same_kind and kind_of(updater.folder_name(pkg)) != kind_of(have_name):
return None # (a package for another backend is not an update of this one)
target = None
try:
archive = updater._download(ctx, "dsv41", pkg.url, pkg.size, pkg.sha256, "the DeepSeek engine", feed)
@@ -295,8 +280,8 @@ def _upgrade(dest: Path, source: str, *, installed: Path | None = None, hardware
shutil.rmtree(target, ignore_errors=True) # (never the installed one)
emit(f"A newer DeepSeek engine is available but could not be installed: {str(exc).rstrip('.')}; still using {shown}.")
return None
for old in dest.iterdir(): # switched: the older generations go (one a running engine uses stays)
if old.is_dir() and old.name != target.name and version_key(old.name) < version_key(target.name) and not in_use(old):
for old in dest.iterdir(): # switched: the older generations go
if old.is_dir() and old.name != target.name and version_key(old.name) < version_key(target.name):
shutil.rmtree(old, ignore_errors=True)
emit(f"DeepSeek engine updated from {shown} to {pkg.version}.")
shutil.rmtree(work, ignore_errors=True)
@@ -304,224 +289,3 @@ def _upgrade(dest: Path, source: str, *, installed: Path | None = None, hardware
except Exception as exc: # noqa: BLE001 - an upgrade check must never stop a launch
emit(f"A newer DeepSeek engine may be available but could not be checked ({type(exc).__name__}); still using {shown}.")
return None
# ---------------------------------------------------------------------------------------------- keeping the engine current for the app
def kind_of(folder_name: str) -> str:
"""The platform-backend-variant part of a package folder's name ("0.1.55.1-windows-x86_64-hip-rdna3" -> "windows-x86_64-hip-rdna3")."""
return str(folder_name).split("-", 1)[1] if "-" in str(folder_name) else ""
def in_use(folder: str | os.PathLike) -> bool:
"""Is a program of this package folder running (an app's engine, a `strata launch` server)? A folder in use is never removed: on Windows the
removal would take its libraries from under the running engine. psutil when it is there (any process whose program lies in the folder), and
on Windows the program itself cannot be opened for writing while it runs."""
folder = Path(folder)
try:
import psutil
except ImportError:
psutil = None
if psutil is not None:
here = os.path.normcase(os.path.abspath(folder)) + os.sep
for p in psutil.process_iter(["exe"]):
try:
if os.path.normcase(os.path.abspath(p.info.get("exe") or "")).startswith(here):
return True
except (psutil.Error, OSError, ValueError):
continue
exe = folder / exe_name()
if os.name == "nt" and exe.is_file():
try:
with open(exe, "r+b"):
pass
except PermissionError:
return True
except OSError:
return False
return False
@contextlib.contextmanager
def _fetch_lock(base: Path):
"""One DeepSeek engine fetch at a time (a lock file beside engine-dsv41/; yields False when a live one holds it). A lock older than 45 minutes is stale."""
path = Path(base) / (DIR + ".lock")
got = False
try:
for _ in range(2):
try:
fd = os.open(path, os.O_CREAT | os.O_EXCL | os.O_WRONLY)
os.write(fd, str(os.getpid()).encode())
os.close(fd)
got = True
break
except FileExistsError:
try:
if time.time() - path.stat().st_mtime < 45 * 60:
break
path.unlink()
except OSError:
break
except OSError:
break
yield got
finally:
if got:
try:
path.unlink()
except OSError:
pass
CHECK_FILE = "feed-check.json"
CHECK_EVERY = 6 * 3600 # a feed check that worked is not repeated for six hours
RETRY_EVERY = 30 * 60 # one that failed (offline, feed unreadable) is not repeated for half an hour
FETCH_WAIT = 90.0 # seconds a start waits for the fetch; after that it goes on with the engine it has and the fetch finishes in the background
def _read_check(dest: Path) -> dict:
try:
data = json.loads((Path(dest) / CHECK_FILE).read_text(encoding="utf-8"))
return data if isinstance(data, dict) else {}
except (OSError, ValueError):
return {}
def _write_check(dest: Path, **fields) -> None:
try:
data = {**_read_check(dest), **fields}
(Path(dest) / CHECK_FILE).write_text(json.dumps(data), encoding="utf-8")
except OSError:
pass
def fetch_due(dest: Path, now: float | None = None) -> bool:
"""Is a feed check due? Never checked, the last good one more than CHECK_EVERY ago, or the last failed one more than RETRY_EVERY ago."""
now = time.time() if now is None else now
check = _read_check(dest)
at = check.get("at")
if not isinstance(at, (int, float)):
return True
return now - at >= (CHECK_EVERY if check.get("ok") else RETRY_EVERY)
def repoint(config: Path, exe: Path) -> bool:
"""Make the run config `config` name `exe` (a strata-dsv41 in an engine-dsv41 folder) and its library folders, nothing else; True when it changed."""
config = Path(config)
cfg = json.loads(config.read_text(encoding="utf-8-sig"))
if not isinstance(cfg, dict) or not isinstance(cfg.get("exe"), str):
return False
libs = [d for d in (cfg.get("lib_dirs") or []) if DIR not in str(d)]
wanted = {"exe": str(exe), "lib_dirs": [*libs, *lib_dirs(exe)]}
if all(cfg.get(k) == v for k, v in wanted.items()):
return False
cfg.update(wanted)
from strata_control import profiles
profiles._atomic(config, json.dumps(cfg, indent=2).encode("utf-8"))
return True
def _engine_root(exe) -> Path | None:
"""The engine-dsv41/ folder a plain run-config exe lies in, or None (a component reference, another engine, an engine of your own)."""
if not isinstance(exe, str) or not exe or exe.startswith("@") or os.environ.get("STRATA_DSV41_EXE"):
return None
root = Path(exe).parent.parent
return root if root.name == DIR and Path(exe).name == exe_name() else None
def _newest_of_kind(root: Path, kind: str) -> Path | None:
try:
same = [d for d in root.iterdir() if d.is_dir() and (d / exe_name()).is_file() and kind_of(d.name) == kind]
except OSError:
return None
same.sort(key=lambda d: (version_key(d.name), d.name))
return same[-1] / exe_name() if same else None
def _spawn_fetch(root: Path, source: str, installed: Path):
import subprocess
import sys
flags = (subprocess.CREATE_NEW_PROCESS_GROUP | subprocess.CREATE_NO_WINDOW) if os.name == "nt" else 0
work = root.parent / (DIR + ".work")
work.mkdir(parents=True, exist_ok=True)
with open(work / "fetch.log", "ab") as out:
return subprocess.Popen([sys.executable, "-m", "strata_control.dsv41_engine", "--fetch", str(root), "--source", source, "--installed", str(installed)],
cwd=str(Path(__file__).resolve().parents[1]), stdin=subprocess.DEVNULL, stdout=out, stderr=subprocess.STDOUT,
creationflags=flags, start_new_session=(os.name != "nt"))
def keep_current(config: str | os.PathLike, *, source: str | None = None, log: Callable[[str], None] = lambda m: None,
wait: float = FETCH_WAIT, configs: list | None = None) -> Path | None:
"""Called before the app starts a saved profile (strata_control.runtime.start); never raises, never holds a start for more than `wait` seconds.
The run config `config` names a DeepSeek engine in engine-dsv41/. (1) When the feed check is due (fetch_due: once in six hours, half an hour
after a failure) a newer package is fetched by a separate background process (python -m strata_control.dsv41_engine --fetch); the start waits for
it `wait` seconds, then goes on with the engine it has and the fetch ends in the background: the next start uses it. (2) When a newer package of
the same kind is installed (that fetch, a `strata launch` that updated it, or the folder the config names is gone), `config` and `configs` (the
other saved profiles' run configs) are pointed at it: local, no network. (3) Older package folders that no running engine uses go.
Returns the engine this start will use, or None when this config is not one of ours to look after."""
try:
cfg = json.loads(Path(config).read_text(encoding="utf-8-sig"))
root = _engine_root(cfg.get("exe") if isinstance(cfg, dict) else None)
if root is None or not root.is_dir():
return None
mine = Path(cfg["exe"])
kind = kind_of(mine.parent.name)
if mine.is_file() and fetch_due(root):
from strata_cli import update_cmd
import argparse
src = source or update_cmd._feed_source(argparse.Namespace(feed=None), root.parent)
_write_check(root, at=time.time(), ok=False) # (a second start in these seconds does not start a second fetch)
child = _spawn_fetch(root, src, mine)
try:
child.wait(timeout=wait)
except Exception: # noqa: BLE001 - subprocess.TimeoutExpired
log(f"A newer DeepSeek engine is downloading in the background ({wait:.0f} s passed): this start uses {mine.parent.name.split('-')[0]}, the next one the new engine.")
best = _newest_of_kind(root, kind)
if best is not None and best != mine and (not mine.is_file() or version_key(best.parent.name) > version_key(mine.parent.name)):
for other in [config, *(configs or [])]:
try:
o = json.loads(Path(other).read_text(encoding="utf-8-sig"))
if isinstance(o, dict) and _engine_root(o.get("exe")) == root and Path(o["exe"]) != best and (
not Path(o["exe"]).is_file() or version_key(Path(o["exe"]).parent.name) < version_key(best.parent.name)):
repoint(Path(other), best)
except (OSError, ValueError):
continue
log(f"DeepSeek engine {best.parent.name.split('-')[0]} is used from now on (it replaced {mine.parent.name.split('-')[0]}).")
mine = best
for old in root.iterdir():
if old.is_dir() and old != mine.parent and kind_of(old.name) == kind and version_key(old.name) < version_key(mine.parent.name) and not in_use(old):
shutil.rmtree(old, ignore_errors=True)
return mine
except Exception as exc: # noqa: BLE001 - a start never fails because of this check
log(f"The DeepSeek engine check was skipped ({type(exc).__name__}); the installed engine is used.")
return None
def _fetch_main(argv: list[str]) -> int:
import argparse
ap = argparse.ArgumentParser(prog="strata_control.dsv41_engine")
ap.add_argument("--fetch", required=True)
ap.add_argument("--source", required=True)
ap.add_argument("--installed", required=True)
a = ap.parse_args(argv)
root = Path(a.fetch)
lines: list[str] = []
ok = False
try:
if not source_reachable(a.source, 3.0):
lines.append("the update feed could not be reached")
else:
upgrade(root, a.source, installed=Path(a.installed), emit=lines.append, same_kind=True)
ok = not any("could not" in x for x in lines)
except Exception as exc: # noqa: BLE001
lines.append(f"{type(exc).__name__}: {exc}")
_write_check(root, at=time.time(), ok=ok)
for line in lines:
print(line, flush=True)
return 0 if ok else 1
if __name__ == "__main__":
import sys
raise SystemExit(_fetch_main(sys.argv[1:]))
strata_control/dsv41_paths.py
+220
View File
@@ -0,0 +1,220 @@
"""Where the DeepSeek engine package (strata-dsv41) is installed and which saved run configs name it: local only, no network, no download.
A saved profile's run config names the DeepSeek engine by a plain path into ``engine-dsv41/<version>-<platform>-<backend>[-<variant>]/``. When an
engine update (the explicit `strata update`, the app's "Update engine", or `strata launch`) replaces that package, the old folder goes and the run
configs would name a program that is not there. This module is what both sides share:
* the start path (strata_control/runtime.py) only REPAIRS: a run config whose folder is gone is pointed at the newest installed package of its own
kind. It never looks at the feed. (Starting and serving never download anything: tests/terminal/test_no_acquisition.py.)
* the explicit update step (strata_control/updater.py update_dsv41) installs the newer package next to the old one, then points every saved run
config at it with `repoint_older`, and removes older folders with `prune`, never one a running engine uses.
"""
from __future__ import annotations
import json
import os
import shutil
import subprocess
from pathlib import Path
NAME = "strata-dsv41"
DIR = "engine-dsv41"
def exe_name() -> str:
return NAME + (".exe" if os.name == "nt" else "")
def version_key(folder_name: str) -> tuple:
"""The version at the front of a package folder's name ("0.1.55.1-windows-x86_64-hip-rdna3" -> (0, 1, 55, 1))."""
head = str(folder_name).split("-")[0]
return tuple(int(x) for x in head.split(".") if x.isdigit())
def kind_of(folder_name: str) -> str:
"""The platform-backend-variant part of a package folder's name ("0.1.55.1-windows-x86_64-hip-rdna3" -> "windows-x86_64-hip-rdna3")."""
return str(folder_name).split("-", 1)[1] if "-" in str(folder_name) else ""
def engine_root(exe) -> Path | None:
"""The engine-dsv41 folder a plain run-config `exe` lies in, or None: a component reference, another engine, or an engine of your own
(STRATA_DSV41_EXE)."""
if not isinstance(exe, str) or not exe or exe.startswith("@") or os.environ.get("STRATA_DSV41_EXE"):
return None
root = Path(exe).parent.parent
return root if root.name == DIR and Path(exe).name == exe_name() else None
def installed(root: Path, kind: str | None = None) -> list[Path]:
"""The strata-dsv41 programs installed in `root` (engine-dsv41/), oldest first; only packages of `kind` when it is given."""
try:
folders = [d for d in Path(root).iterdir() if d.is_dir() and (d / exe_name()).is_file() and (kind is None or kind_of(d.name) == kind)]
except OSError:
return []
folders.sort(key=lambda d: (version_key(d.name), d.name))
return [d / exe_name() for d in folders]
def lib_dirs(exe) -> list[str]:
"""The folders that must be on the library path when this engine starts (its own folder; a Windows HIP package also rocm/bin)."""
from strata_control import engine_component as ec
try:
return ec.package_lib_dirs(Path(exe).parent)
except OSError:
return [str(Path(exe).parent)]
def in_use(folder) -> bool:
"""Is a program of this package folder running (an app's engine, a `strata launch` server)? A folder in use is never removed: on Windows the
removal would take its libraries from under the running engine. psutil when it is there (any process whose program lies in the folder), and
on Windows the program itself cannot be opened for writing while it runs."""
folder = Path(folder)
try:
import psutil
except ImportError:
psutil = None
if psutil is not None:
here = os.path.normcase(os.path.abspath(folder)) + os.sep
for p in psutil.process_iter(["exe"]):
try:
if os.path.normcase(os.path.abspath(p.info.get("exe") or "")).startswith(here):
return True
except (psutil.Error, OSError, ValueError):
continue
exe = folder / exe_name()
if os.name == "nt" and exe.is_file():
try:
with open(exe, "r+b"):
pass
except PermissionError:
return True
except OSError:
return False
return False
def prune(root: Path, keep: Path) -> list[str]:
"""Remove the older package folders of `keep`'s kind in `root` that no running engine uses; returns the names kept because they are in use."""
busy = []
kind = kind_of(keep.parent.name)
try:
folders = [d for d in Path(root).iterdir() if d.is_dir()]
except OSError:
return busy
for old in folders:
if old != keep.parent and kind_of(old.name) == kind and version_key(old.name) < version_key(keep.parent.name):
if in_use(old):
busy.append(old.name)
else:
shutil.rmtree(old, ignore_errors=True)
return busy
def repoint(config, exe) -> bool:
"""Make the run config `config` name `exe` (a strata-dsv41 in an engine-dsv41 folder) and that package's library folders, nothing else
changed; True when the file changed."""
config = Path(config)
cfg = json.loads(config.read_text(encoding="utf-8-sig"))
if not isinstance(cfg, dict) or not isinstance(cfg.get("exe"), str):
return False
libs = [d for d in (cfg.get("lib_dirs") or []) if DIR not in str(d)]
wanted = {"exe": str(exe), "lib_dirs": [*libs, *lib_dirs(exe)]}
if all(cfg.get(k) == v for k, v in wanted.items()):
return False
cfg.update(wanted)
from strata_control import profiles
profiles._atomic(config, json.dumps(cfg, indent=2).encode("utf-8"))
return True
def saved_configs() -> list[Path]:
"""The run config of every saved profile (app-1, app-2, launch-...): the files a start reads."""
from strata_control import profiles
out = []
try:
for p in profiles.list_profiles():
if isinstance(p.get("source_config"), str):
out.append(Path(p["source_config"]))
except Exception: # noqa: BLE001 - informational
pass
return out
def _read(path) -> dict | None:
try:
cfg = json.loads(Path(path).read_text(encoding="utf-8-sig"))
except (OSError, ValueError):
return None
return cfg if isinstance(cfg, dict) else None
def roots_of(configs) -> list[Path]:
"""The engine-dsv41 folders the run configs `configs` name (each once)."""
roots: list[Path] = []
for c in configs:
cfg = _read(c)
root = engine_root(cfg.get("exe")) if cfg else None
if root is not None and root not in roots:
roots.append(root)
return roots
def repair_missing(configs) -> list[str]:
"""For a START: every run config in `configs` whose package folder is gone is pointed at the newest installed package of its own kind (nothing
else changes, nothing is fetched). Returns one plain line for each config it changed."""
notes = []
for c in configs:
cfg = _read(c)
exe = cfg.get("exe") if cfg else None
root = engine_root(exe)
if root is None or Path(exe).is_file() or not root.is_dir():
continue
found = installed(root, kind_of(Path(exe).parent.name))
if not found:
continue
try:
if repoint(c, found[-1]):
notes.append(f"The DeepSeek engine {Path(exe).parent.name.split('-')[0]} was replaced by an update; {Path(c).name} now uses {found[-1].parent.name.split('-')[0]}.")
except (OSError, ValueError):
continue
return notes
def repoint_older(configs, new_exe) -> int:
"""After an update installed `new_exe`: every run config in `configs` that names an older package of the same kind in the same folder (present or
gone) is pointed at it. Returns how many files changed."""
new_exe = Path(new_exe)
root, kind = new_exe.parent.parent, kind_of(new_exe.parent.name)
changed = 0
for c in configs:
cfg = _read(c)
exe = cfg.get("exe") if cfg else None
if engine_root(exe) != root or Path(exe) == new_exe or kind_of(Path(exe).parent.name) != kind:
continue
if version_key(Path(exe).parent.name) >= version_key(new_exe.parent.name):
continue
try:
changed += 1 if repoint(c, new_exe) else 0
except (OSError, ValueError):
continue
return changed
def test_start(exe, timeout: float = 120.0) -> tuple[bool, str]:
"""`strata-dsv41 --version` runs from its own folder (the package's library folders first on the path) and names itself: it needs no graphics
card and no model."""
env = dict(os.environ)
dirs = [d for d in lib_dirs(exe) if Path(d).is_dir()]
var = "PATH" if os.name == "nt" else "LD_LIBRARY_PATH"
if dirs:
env[var] = os.pathsep.join(dirs + ([env[var]] if env.get(var) else []))
try:
done = subprocess.run([str(exe), "--version"], capture_output=True, text=True, errors="replace", timeout=timeout, cwd=str(Path(exe).parent), env=env,
creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0))
except (OSError, subprocess.SubprocessError) as exc:
return False, f"it could not be started ({exc})"
first = ((done.stdout or "").strip().splitlines() or [""])[0]
if done.returncode != 0 or not first.startswith(NAME):
tail = ((done.stderr or "").strip().splitlines() or [first or f"exit {done.returncode}"])[-1]
return False, f"it did not start correctly ({tail[:160]})"
return True, first
strata_control/engine_component.py
+34 -7
View File
@@ -205,24 +205,50 @@ def cached_notice(root: Path | None = None) -> str | None:
The notice belongs to the versions the check was made for: it is dropped when Control or the installed engine is not the one
recorded then (an app install, an update, a rollback), and an engine that is not newer than the installed one is never offered."""
offers = cached_offers(root)
parts = ([f"engine {offers['engine']}"] if offers.get("engine") else []) + ([f"DeepSeek engine {offers['dsv41']}"] if offers.get("dsv41") else []) \
+ ([f"app {offers['app']}"] if offers.get("app") else [])
return f"update available ({', '.join(parts)}): run `strata update`" if parts else None
def _installed_dsv41_version(profile_configs=None) -> str | None:
"""The newest DeepSeek engine package the saved profiles name (a local read), or None."""
from strata_control import dsv41_paths
newest = None
for r in dsv41_paths.roots_of(profile_configs if profile_configs is not None else dsv41_paths.saved_configs()):
for exe in dsv41_paths.installed(r):
v = exe.parent.name.split("-")[0]
if newest is None or _version_key(v) > _version_key(newest):
newest = v
return newest
def cached_offers(root: Path | None = None) -> dict:
"""What the last explicit check found newer, as {"engine": version|None, "dsv41": version|None, "app": version|None}; local reads only (the cached
check, the installed packages), never the network. The same rules as `cached_notice`: it belongs to the Control and the packages it was made for,
and a package that is not newer than the installed one is never offered."""
base = root or components_dir()
none = {"engine": None, "dsv41": None, "app": None}
try:
last = (json.loads((base / UPDATE_STATE).read_text(encoding="utf-8-sig")).get("last_check") or {})
except (OSError, ValueError, AttributeError):
return None
return none
if not isinstance(last, dict):
return None
return none
control = runtime_version()
if "control" in last and last["control"] != control:
return None
engine, app = last.get("engine"), last.get("app")
return none
engine, app, dsv41 = last.get("engine"), last.get("app"), last.get("dsv41")
installed = newest_installed_version(base)
if "engine_installed" in last and last["engine_installed"] != installed:
engine = None
if engine and installed and _version_key(str(engine)) <= _version_key(installed):
engine = None
parts = ([f"engine {engine}"] if engine else []) + ([f"app {app}"] if app else [])
return f"update available ({', '.join(parts)}): run `strata update`" if parts else None
if dsv41:
have = _installed_dsv41_version()
if have is None or "dsv41_installed" in last and last["dsv41_installed"] != have or _version_key(str(dsv41)) <= _version_key(have):
dsv41 = None
return {"engine": engine or None, "dsv41": dsv41 or None, "app": app or None}
def previous_folder(root: Path | None = None) -> str | None:
@@ -1028,7 +1054,8 @@ def status(root: Path | None = None, *, backend: str | None = None, full: bool =
"backend": backend or detected, "backend_source": "requested" if backend else how,
"installed": [item.to_dict() for item in found], "problems": problems,
"engine": None, "usable": False, "notice": None, "acquires": False,
"pinned": pinned_folder(base), "previous": previous_folder(base), "update_notice": cached_notice(base)}
"pinned": pinned_folder(base), "previous": previous_folder(base), "update_notice": cached_notice(base),
"update_available": cached_offers(base)}
try:
chosen = resolve(base, backend=backend, full=full)
report["engine"] = chosen.to_dict()
strata_control/runtime.py
+15 -21
View File
@@ -515,28 +515,24 @@ def _check_start_candidate(name: str, expected_profile_revision: str | None,
return plan, digest
def _keep_dsv41_current(name: str) -> float:
"""Before a start: a saved profile that runs the DeepSeek engine from an engine-dsv41/ folder gets the newest installed package of its kind, and the
signed feed is asked for a newer one at most every six hours (strata_control.dsv41_engine.keep_current: never fails a start, waits for a download at
most 90 s and then goes on with the installed engine). Returns the seconds it took. One line of what it did goes to the profile's host log."""
began = time.monotonic()
def _repair_dsv41_paths(name: str) -> None:
"""Before a start: a saved profile that runs the DeepSeek engine from an engine-dsv41/ folder an update has replaced is pointed at the newest
installed package of its own kind (every saved profile's run config, so the next start of another one is repaired too). Local only: it never
looks at the feed and fetches nothing - engine updates are the explicit `strata update` / the app's "Update engine". Never fails a start;
one line of what it did goes to the profile's host log."""
try:
from strata_control import dsv41_engine
from strata_control import dsv41_paths
profile, _ = profiles.load(name)
others = [p["source_config"] for p in profiles.list_profiles() if p.get("source_config") and p["id"] != name]
_, _, log_path = _paths(name)
def log(message: str) -> None:
try:
if log_path.parent.is_dir():
with open(log_path, "ab") as stream:
stream.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S')} engine: {message}\n".encode("utf-8"))
except OSError:
pass
dsv41_engine.keep_current(profile["source_config"], configs=others, log=log)
configs = [Path(profile["source_config"]), *[c for c in dsv41_paths.saved_configs() if c != Path(profile["source_config"])]]
notes = dsv41_paths.repair_missing(configs)
if notes:
_, _, log_path = _paths(name)
if log_path.parent.is_dir():
with open(log_path, "ab") as stream:
for note in notes:
stream.write(f"{time.strftime('%Y-%m-%dT%H:%M:%S')} engine: {note}\n".encode("utf-8"))
except Exception: # noqa: BLE001 - a start never fails because of this
pass
return time.monotonic() - began
MAX_START_WAIT = 1800 # seconds a start may wait for the model: a 250 GB DeepSeek set on a slow disk needs more than 10 minutes (#3949)
@@ -570,9 +566,7 @@ def start(name: str, *, wait: float = 300, shutdown_timeout: float = 30,
if expected_source_sha256 is not None and (not isinstance(expected_source_sha256, str) or
not re.fullmatch(r"[0-9a-f]{64}", expected_source_sha256)):
raise ProfileError("expected source digest must be 64 lowercase hex characters")
spent = _keep_dsv41_current(name)
if spent > 1.0 and wait:
wait = max(5.0, wait - spent) # (what the engine check took is not taken from the model's own wait for the caller's timeout)
_repair_dsv41_paths(name)
plan, _ = _check_start_candidate(name, expected_profile_revision, expected_source_sha256)
_engine_preflight(plan)
_, _, log_path = _paths(name)
strata_control/updater.py
+125 -8
View File
@@ -42,7 +42,7 @@ from dataclasses import dataclass, field
from pathlib import Path
from typing import Callable
from strata_control import acquisition, components, engine_component as ec, feed as feedmod, longpath, preset_marks, signing, transfer
from strata_control import acquisition, components, dsv41_paths, engine_component as ec, feed as feedmod, longpath, preset_marks, signing, transfer
from strata_control.targets import host_platform
from strata_control import atomic as _atomic_mod
@@ -636,17 +636,25 @@ def check(ctx: Context) -> dict:
report = {"feed": {"source": ctx.source, "sequence": feed.sequence, "channel": feed.channel, "key_id": feed.key_id,
"generated_at": feed.generated_at, "trusted_by": "test key file" if signing.extra_keys_in_effect(ctx.environ) else "built-in keys"},
"engine": decide_engine(feed, ctx, state), "control": decide_control(feed, ctx, state),
"app": decide_app(feed, ctx), "serving": _serving_or_empty(ctx), "hardware": ctx.hardware}
state.data["last_check"] = {
"at": (ctx.now or dt.datetime.now(dt.timezone.utc)).isoformat(),
"engine": report["engine"]["offered"]["version"] if report["engine"]["status"] in ("available", "install") and report["engine"]["offered"] else None,
"app": report["app"]["offered"] if report["app"]["status"] == "available" else None,
# the versions this result is for, so a later Control, engine or app change drops the cached notice
"control": ec.runtime_version(), "engine_installed": ec.newest_installed_version(ctx.components)}
"app": decide_app(feed, ctx), "dsv41": decide_dsv41(feed, ctx), "serving": _serving_or_empty(ctx), "hardware": ctx.hardware}
state.data["last_check"] = _last_check(ctx, report)
state.save()
return report
def _last_check(ctx: Context, report: dict, control_version: str | None = None) -> dict:
"""What the feed offers that is newer than what is installed, kept for the notice a start or the app shows without touching the network
(engine_component.cached_offers). ``control_version``: the Control an update has just installed (the notice belongs to it, not to the one running)."""
return {
"at": (ctx.now or dt.datetime.now(dt.timezone.utc)).isoformat(),
"engine": report["engine"]["offered"]["version"] if report["engine"]["status"] in ("available", "install") and report["engine"]["offered"] else None,
"app": report["app"]["offered"] if report["app"]["status"] == "available" else None,
"dsv41": report["dsv41"]["offered"] if report["dsv41"]["status"] == "available" else None,
"dsv41_installed": report["dsv41"]["installed"],
# the versions this result is for, so a later Control, engine or app change drops the cached notice
"control": control_version or ec.runtime_version(), "engine_installed": ec.newest_installed_version(ctx.components)}
def new_stage(root: Path, prefix: str = "") -> Path:
r"""A fresh, short staging folder: ``<components>\.s\<4 hex>``. Old leftovers (a crashed update; the update lock is held, so
nothing else uses them) are swept first, along with the longer folder older versions used."""
@@ -1334,6 +1342,106 @@ def _failed_mark_is_old(mark: dict, ctx: Context, hours: float = 24.0) -> bool:
return False
def _dsv41_installs(ctx: Context) -> tuple[list[Path], list[Path]]:
"""(the engine-dsv41 folders in use, the saved run configs that may name a package in them). A saved profile's run config says where the
DeepSeek engine is; the install folder the app or Terminal runs from (STRATA_INSTALL_DIR) is looked at too, with its model run configs."""
configs = dsv41_paths.saved_configs()
install = ctx.environ.get("STRATA_INSTALL_DIR")
if install and Path(install).is_dir():
configs += sorted(Path(install).glob("strata-*.json"))
roots = dsv41_paths.roots_of(configs)
for extra in ([Path(install) / dsv41_paths.DIR] if install else []) + [ctx.components.parent / dsv41_paths.DIR]:
if extra.is_dir() and extra not in roots and dsv41_paths.installed(extra):
roots.append(extra)
return roots, configs
def _dsv41_offer(feed: feedmod.Feed, ctx: Context, newest: Path) -> tuple[feedmod.EnginePackage | None, list[str]]:
"""The feed's DeepSeek package of the SAME kind (platform, backend, variant) as the installed `newest`, if it fits this PC."""
kind = dsv41_paths.kind_of(newest.parent.name)
same = [p for p in feed.dsv41 if dsv41_paths.kind_of(folder_name(p)) == kind]
if not same:
return None, []
parts = kind.split("-")
hardware = {**ctx.hardware, "backend": parts[2] if len(parts) > 2 else ctx.hardware.get("backend"), "backend_source": "requested"}
return feedmod.select_engine(same, hardware)
def decide_dsv41(feed: feedmod.Feed, ctx: Context) -> dict:
"""What an update would do for the installed DeepSeek engine: nothing is downloaded. ``status``: none-installed, current, available, none-for-this-pc."""
roots, _ = _dsv41_installs(ctx)
report = {"component": "dsv41", "status": "none-installed", "installed": None, "offered": None, "download_bytes": 0}
for root in roots:
found = dsv41_paths.installed(root)
if not found:
continue
newest = found[-1]
report.update(status="current", installed=newest.parent.name.split("-")[0], root=str(root))
pkg, reasons = _dsv41_offer(feed, ctx, newest)
if pkg is None:
report.update(status="none-for-this-pc", reasons=reasons)
else:
report.update(offered=pkg.version, download_bytes=pkg.size)
if feedmod.version_key(pkg.version) > feedmod.version_key(report["installed"]):
report["status"] = "available"
if report["status"] == "available":
break
return report
def update_dsv41(ctx: Context, feed: feedmod.Feed, state: State) -> dict:
"""The explicit update of an installed DeepSeek engine package (the app's "Update engine", `strata update`): the signed feed's newer package of
the same kind is downloaded, checked (signature and every file), unpacked NEXT TO the old one, started once (`--version`), and only then every saved
run config is pointed at it. Older package folders go, except one a running engine uses (it goes with a later update). Nothing installed: nothing
happens. The engine that is serving is not touched, so nothing waits for it. Starting never does any of this (strata_control/dsv41_paths.py)."""
roots, configs = _dsv41_installs(ctx)
result = {"component": "dsv41", "status": "none-installed", "changed": False}
for root in roots:
found = dsv41_paths.installed(root)
if not found:
continue
newest = found[-1]
before = newest.parent.name.split("-")[0]
result.update(status="current", installed_before=before)
pkg, reasons = _dsv41_offer(feed, ctx, newest)
if pkg is None:
result.update(status="none-for-this-pc", note="; ".join(reasons[:2]) or "the feed has no DeepSeek engine for this PC")
continue
if feedmod.version_key(pkg.version) <= feedmod.version_key(before):
result["note"] = "The DeepSeek engine is up to date."
continue
_free_space_check(root, pkg.size + pkg.installed_bytes, ctx, pkg.sha256, "dsv41")
archive = _download(ctx, "dsv41", pkg.url, pkg.size, pkg.sha256, f"the DeepSeek engine {pkg.version}", feed)
stage_root = None
try:
stage_root, _staged = stage_engine(ctx, feed, pkg, archive)
folder = stage_root / "engine" / folder_name(pkg)
target = root / folder.name
root.mkdir(parents=True, exist_ok=True)
if target.exists():
shutil.rmtree(longpath.long_path(target), ignore_errors=True)
_move_dir(folder, target)
if os.name != "nt":
(target / dsv41_paths.exe_name()).chmod(0o755)
ctx.say("test", f"Test-starting the DeepSeek engine {pkg.version} before switching to it", component="dsv41")
ok, why = dsv41_paths.test_start(target / dsv41_paths.exe_name())
if not ok:
shutil.rmtree(longpath.long_path(target), ignore_errors=True) # (never the installed one)
raise TestStartFailed(f"DeepSeek engine {pkg.version}", why)
new_exe = target / dsv41_paths.exe_name()
moved = dsv41_paths.repoint_older(configs, new_exe)
kept = dsv41_paths.prune(root, new_exe)
archive.unlink(missing_ok=True)
result.update(status="updated", changed=True, installed_after=pkg.version, profiles_repointed=moved, kept_in_use=kept,
note="Models you start from now on use the new DeepSeek engine." + (" The one that is running now keeps the old engine until it is stopped." if kept else ""))
state.history({"at": (ctx.now or dt.datetime.now(dt.timezone.utc)).isoformat(), "action": "update-dsv41", "from": before, "to": pkg.version, "ok": True})
return result
finally:
if stage_root is not None:
shutil.rmtree(longpath.long_path(stage_root), ignore_errors=True)
return result
def update_control(ctx: Context, feed: feedmod.Feed, state: State) -> dict:
"""Stage a newer Control bundle beside the running one, run it once (`--version`), and only then point the app and the
terminal at it (``components/control/active.json``). The running Control is never modified."""
@@ -1473,6 +1581,7 @@ def run_update(ctx: Context, *, engine: bool = True, control: bool = True, app:
feed = open_feed(ctx, state)
state.save()
steps = ([("engine", lambda: update_engine(ctx, feed, state))] if engine else []) + \
([("dsv41", lambda: update_dsv41(ctx, feed, state))] if engine else []) + \
([("control", lambda: update_control(ctx, feed, state))] if control else []) + \
([("app", lambda: update_app(ctx, feed))] if app else [])
for name, step in steps:
@@ -1484,6 +1593,14 @@ def run_update(ctx: Context, *, engine: bool = True, control: bool = True, app:
failure = failure or exc
if any(r.get("component") == "engine" and r.get("status") == "updated" for r in results):
state.data.get("last_check", {}).pop("engine", None)
if any(r.get("component") == "dsv41" and r.get("status") == "updated" for r in results):
state.data.get("last_check", {}).pop("dsv41", None)
try: # the feed was read for this update: what it offers that is still newer is kept for the app's one card
now_control = next((r.get("installed_after") for r in results if r.get("component") == "control" and r.get("status") == "updated"), None)
state.data["last_check"] = _last_check(ctx, {"engine": decide_engine(feed, ctx, state), "app": decide_app(feed, ctx),
"dsv41": decide_dsv41(feed, ctx)}, now_control)
except Exception: # noqa: BLE001 - only a hint for the app
pass
state.save()
return {"feed": {"source": ctx.source, "sequence": feed.sequence, "channel": feed.channel, "key_id": feed.key_id},
"results": results, "exit_code": failure.exit_code if failure else 0}
tests/terminal/test_dsv41_keep_current.py
-174
View File
@@ -1,174 +0,0 @@
"""The app starts a saved profile without going through `strata launch`: before `runtime start` the DeepSeek engine in that profile's run config is
kept current (strata_control.dsv41_engine.keep_current). A newer package is fetched by a background process (the start waits for it a little, then goes
on), the run configs of every saved profile are pointed at the newest installed package, a folder a running engine uses is never removed, and nothing
here ever fails a start."""
from __future__ import annotations
import json
import subprocess
import sys
import time
import unittest
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from strata_control import dsv41_engine # noqa: E402
from tests.terminal import test_dsv41_upgrade as _up # noqa: E402
EXE = dsv41_engine.exe_name()
class FakeChild:
"""The background fetch: runs the upgrade in this process when waited for (the test box has no GPU to detect), or never ends."""
def __init__(self, work=None, hang=False):
self.work, self.hang = work, hang
def wait(self, timeout=None):
if self.hang:
raise subprocess.TimeoutExpired("fetch", timeout)
if self.work:
self.work()
return 0
class KeepCurrent(_up.Upgrade):
def setUp(self):
super().setUp()
self.cfg = self.root / "run" / "app-1.json"
self.cfg2 = self.root / "run" / "app-2.json"
self.cfg.parent.mkdir()
for c in (self.cfg, self.cfg2):
self.write_cfg(c, self.old)
self.lines = []
def write_cfg(self, path, exe):
path.write_text(json.dumps({"exe": str(exe), "args": ["--pack", "p"], "tokenizer": "t", "lib_dirs": ["C:/other", str(Path(exe).parent)]}), encoding="utf-8")
def exe_of(self, path):
return Path(json.loads(Path(path).read_text(encoding="utf-8"))["exe"])
def keep(self, child=None, **kw):
spawn = mock.patch.object(dsv41_engine, "_spawn_fetch", return_value=child or FakeChild())
with spawn as sp, mock.patch.object(dsv41_engine, "in_use", return_value=kw.pop("busy", False)):
out = dsv41_engine.keep_current(self.cfg, source=self.url, log=self.lines.append, configs=[self.cfg2], wait=5, **kw)
return out, sp
def fetch_now(self):
return lambda: self.upgrade()
def test_a_due_check_fetches_waits_and_every_saved_profile_points_at_the_new_engine(self):
self.publish("0.1.49.1")
out, sp = self.keep(FakeChild(self.fetch_now()))
sp.assert_called_once()
new = dsv41_engine.find(self.dest)
self.assertEqual(out, new)
self.assertIn("0.1.49.1", str(new))
self.assertEqual(self.exe_of(self.cfg), new)
self.assertEqual(self.exe_of(self.cfg2), new, "the other saved profile must not be left on a folder that goes")
libs = json.loads(self.cfg.read_text(encoding="utf-8"))["lib_dirs"]
self.assertEqual(libs[0], "C:/other")
self.assertIn(str(new.parent), libs)
self.assertEqual(self.folders(), ["0.1.49.1"])
self.assertIn("used from now on", " ".join(self.lines))
def test_a_fresh_check_is_not_repeated_and_costs_no_network(self):
self.publish("0.1.49.1")
dsv41_engine._write_check(self.dest, at=time.time() - 60, ok=True)
out, sp = self.keep()
sp.assert_not_called()
self.assertEqual(out, self.old)
self.assertEqual(self.exe_of(self.cfg), self.old)
def test_the_check_is_due_after_six_hours_or_half_an_hour_after_a_failure(self):
now = time.time()
self.assertTrue(dsv41_engine.fetch_due(self.dest, now)) # never checked
dsv41_engine._write_check(self.dest, at=now - 3600, ok=True)
self.assertFalse(dsv41_engine.fetch_due(self.dest, now))
self.assertTrue(dsv41_engine.fetch_due(self.dest, now + 6 * 3600))
dsv41_engine._write_check(self.dest, at=now - 600, ok=False)
self.assertFalse(dsv41_engine.fetch_due(self.dest, now))
self.assertTrue(dsv41_engine.fetch_due(self.dest, now + 1800))
def test_a_failed_or_offline_fetch_keeps_the_installed_engine_and_the_next_start_does_not_try_again_at_once(self):
out, sp = self.keep(FakeChild()) # (the child ends and installs nothing)
self.assertEqual(out, self.old)
self.assertEqual(self.exe_of(self.cfg), self.old)
out, sp2 = self.keep()
sp2.assert_not_called() # (_spawn_fetch wrote the attempt before it ran)
self.assertEqual(out, self.old)
def test_a_slow_download_does_not_hold_the_start_and_the_next_one_uses_the_new_engine(self):
self.publish("0.1.49.1")
out, _ = self.keep(FakeChild(hang=True))
self.assertEqual(out, self.old)
self.assertEqual(self.exe_of(self.cfg), self.old)
self.assertIn("downloading in the background", " ".join(self.lines))
self.upgrade() # (the background fetch ends)
out, sp = self.keep()
sp.assert_not_called()
self.assertEqual(self.exe_of(self.cfg), dsv41_engine.find(self.dest))
self.assertIn("0.1.49.1", str(out))
def test_a_folder_a_running_engine_uses_is_not_removed(self):
self.publish("0.1.49.1")
with mock.patch.object(dsv41_engine, "in_use", return_value=True):
self.assertIsNotNone(self.upgrade())
self.assertEqual(self.folders(), ["0.1.49", "0.1.49.1"])
out, _ = self.keep(busy=True)
self.assertEqual(self.folders(), ["0.1.49", "0.1.49.1"], "still in use: kept")
out, _ = self.keep(busy=False)
self.assertEqual(self.folders(), ["0.1.49.1"], "free now: removed")
def test_a_config_whose_folder_was_removed_by_a_launch_is_pointed_at_the_installed_one(self):
self.publish("0.1.49.1")
self.assertIsNotNone(self.upgrade()) # `strata launch` did this; the app's configs still name 0.1.49
self.assertFalse(self.old.exists())
dsv41_engine._write_check(self.dest, at=time.time(), ok=True)
out, sp = self.keep()
sp.assert_not_called()
self.assertEqual(self.exe_of(self.cfg), dsv41_engine.find(self.dest))
self.assertEqual(self.exe_of(self.cfg2), dsv41_engine.find(self.dest))
def test_an_engine_of_your_own_or_a_component_reference_is_left_alone(self):
for exe in ("@component:strata-engine", str(self.root / "mine" / EXE)):
c = self.root / "run" / "own.json"
c.write_text(json.dumps({"exe": exe, "args": []}), encoding="utf-8")
with mock.patch.object(dsv41_engine, "_spawn_fetch") as sp:
self.assertIsNone(dsv41_engine.keep_current(c, source=self.url, log=self.lines.append))
sp.assert_not_called()
self.assertEqual(json.loads(c.read_text(encoding="utf-8"))["exe"], exe)
with mock.patch.dict("os.environ", {"STRATA_DSV41_EXE": str(self.old)}), mock.patch.object(dsv41_engine, "_spawn_fetch") as sp:
self.assertIsNone(dsv41_engine.keep_current(self.cfg, source=self.url, log=self.lines.append))
sp.assert_not_called()
def test_nothing_in_the_check_can_fail_a_start(self):
with mock.patch.object(dsv41_engine, "_spawn_fetch", side_effect=OSError("no process")):
out = dsv41_engine.keep_current(self.cfg, source=self.url, log=self.lines.append, configs=[self.cfg2], wait=1)
self.assertIsNone(out)
self.assertIn("skipped", " ".join(self.lines))
self.cfg.write_text("{not json", encoding="utf-8")
self.assertIsNone(dsv41_engine.keep_current(self.cfg, source=self.url, log=self.lines.append))
def test_a_package_of_another_backend_is_not_an_update_of_this_one(self):
self.publish("0.1.49.1")
with mock.patch.object(dsv41_engine, "kind_of", side_effect=lambda n: "other" if n.startswith("0.1.49.1") else "cuda"):
with mock.patch.object(dsv41_engine, "_test_start", return_value=(True, "strata-dsv41 0.1.49.1")):
self.assertIsNone(dsv41_engine.upgrade(self.dest, self.url, installed=self.old, hardware=self.hw, emit=self.lines.append, same_kind=True))
self.assertEqual(self.folders(), ["0.1.49"])
def test_two_fetches_do_not_run_at_once(self):
self.publish("0.1.49.1")
with dsv41_engine._fetch_lock(self.dest.parent) as first:
self.assertTrue(first)
self.assertIsNone(self.upgrade()) # a second caller gets None at once
self.assertIsNotNone(self.upgrade())
for _name in dir(_up.Upgrade): # (the parent's own tests are not repeated here)
if _name.startswith("test_") and _name not in KeepCurrent.__dict__:
setattr(KeepCurrent, _name, None)
if __name__ == "__main__":
unittest.main()
tests/terminal/test_dsv41_update_path.py
+226
View File
@@ -0,0 +1,226 @@
"""An installed DeepSeek engine is updated by the explicit update (`strata update`, the app's "Update engine"), never by a start.
* update_dsv41: the signed feed's newer package of the SAME kind is downloaded, checked, put next to the old one, started once, and only then every saved
run config is pointed at it; older folders go, except one a running engine uses; nothing installed means nothing happens.
* The start (strata_control/runtime.py) only repairs: a run config whose folder an update removed is pointed at the newest installed package, locally.
It makes no request of any kind.
* The check keeps what is newer (engine_component.cached_offers) for the app's one card; `strata launch`'s own upgrade is not touched here.
"""
from __future__ import annotations
import json
import os
import socket
import sys
import time
import unittest
import urllib.request
from pathlib import Path
from unittest import mock
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
from strata_control import dsv41_paths, engine_component as ec, runtime, signing, updater # noqa: E402
from tests.terminal import test_dsv41_upgrade as _up # noqa: E402
EXE = dsv41_paths.exe_name()
class UpdatePath(_up.Upgrade):
def setUp(self):
super().setUp()
self.install_dir = self.root / "data"
self.cfg1 = self.install_dir / "strata-a.json"
self.cfg2 = self.root / "run" / "app-1.json"
self.cfg2.parent.mkdir()
for c in (self.cfg1, self.cfg2):
self.write_cfg(c, self.old)
self.configs = [self.cfg1, self.cfg2]
for p in (mock.patch.object(dsv41_paths, "saved_configs", return_value=[self.cfg2]),
mock.patch.object(dsv41_paths, "test_start", return_value=(True, "strata-dsv41 0.1.49.1 build x")),
mock.patch.object(dsv41_paths, "in_use", return_value=False)):
p.start()
self.addCleanup(p.stop)
self.lines = []
def write_cfg(self, path, exe):
Path(path).write_text(json.dumps({"exe": str(exe), "args": ["--pack", "p"], "tokenizer": "t",
"lib_dirs": ["C:/other", str(Path(exe).parent)]}), encoding="utf-8")
def exe_of(self, path):
return Path(json.loads(Path(path).read_text(encoding="utf-8"))["exe"])
def ctx(self):
return updater.Context(components=self.root / "c", source=self.url, trusted=signing.load_trusted_keys(), hardware=self.hw,
emit=lambda e: self.lines.append(e.get("message")), environ={**os.environ, "STRATA_INSTALL_DIR": str(self.install_dir)})
def run_step(self):
ctx = self.ctx()
state = updater.State(ctx.components)
feed = updater.open_feed(ctx, state)
return ctx, state, feed
# ---- the explicit update ------------------------------------------------------------------------------------------
def test_the_update_installs_next_to_the_old_points_every_saved_config_at_it_and_removes_the_old(self):
self.publish("0.1.49.1")
ctx, state, feed = self.run_step()
r = updater.update_dsv41(ctx, feed, state)
self.assertEqual((r["status"], r["installed_before"], r["installed_after"]), ("updated", "0.1.49", "0.1.49.1"), r)
new = dsv41_paths.installed(self.dest)[-1]
self.assertEqual(self.folders(), ["0.1.49.1"])
self.assertEqual(self.exe_of(self.cfg1), new)
self.assertEqual(self.exe_of(self.cfg2), new, "every saved run config, not one")
libs = json.loads(self.cfg1.read_text(encoding="utf-8"))["lib_dirs"]
self.assertEqual(libs[0], "C:/other")
self.assertIn(str(new.parent), libs)
self.assertEqual(r["profiles_repointed"], 2)
self.assertEqual(self.tree(self.eng), self.before, "the Qwen engine's files must not change")
def test_the_old_folder_is_there_until_the_new_one_has_started(self):
self.publish("0.1.49.1")
ctx, state, feed = self.run_step()
seen = {}
def start(exe):
seen["folders"] = self.folders()
seen["cfg"] = self.exe_of(self.cfg1)
return True, "strata-dsv41 0.1.49.1"
with mock.patch.object(dsv41_paths, "test_start", side_effect=start):
updater.update_dsv41(ctx, feed, state)
self.assertEqual(seen["folders"], ["0.1.49", "0.1.49.1"])
self.assertEqual(seen["cfg"], self.old, "no config is touched before the test start passed")
def test_a_package_that_does_not_start_changes_nothing(self):
self.publish("0.1.49.1")
ctx, state, feed = self.run_step()
with mock.patch.object(dsv41_paths, "test_start", return_value=(False, "it did not start correctly (x)")):
with self.assertRaises(updater.UpdateError):
updater.update_dsv41(ctx, feed, state)
self.assertEqual(self.folders(), ["0.1.49"])
self.assertEqual(self.exe_of(self.cfg1), self.old)
self.assertEqual(self.exe_of(self.cfg2), self.old)
def test_a_folder_a_running_engine_uses_is_not_removed_and_the_configs_still_move(self):
self.publish("0.1.49.1")
ctx, state, feed = self.run_step()
with mock.patch.object(dsv41_paths, "in_use", return_value=True):
r = updater.update_dsv41(ctx, feed, state)
self.assertEqual(r["status"], "updated")
self.assertEqual(self.folders(), ["0.1.49", "0.1.49.1"])
self.assertTrue(r["kept_in_use"])
self.assertIn("0.1.49.1", str(self.exe_of(self.cfg2)))
# a later update (nothing newer) leaves it; the start-side repair never removes anything
self.assertEqual(dsv41_paths.prune(self.dest, dsv41_paths.installed(self.dest)[-1]), [])
self.assertEqual(self.folders(), ["0.1.49.1"])
def test_nothing_newer_or_nothing_installed_changes_nothing(self):
ctx, state, feed = self.run_step()
r = updater.update_dsv41(ctx, feed, state)
self.assertEqual((r["status"], r["changed"]), ("current", False))
self.assertEqual(self.folders(), ["0.1.49"])
with mock.patch.object(dsv41_paths, "saved_configs", return_value=[]), mock.patch.dict(os.environ, {}):
ctx = updater.Context(components=self.root / "c2", source=self.url, trusted=signing.load_trusted_keys(), hardware=self.hw, environ={})
r = updater.update_dsv41(ctx, feed, state)
self.assertEqual((r["status"], r["changed"]), ("none-installed", False))
def test_a_package_of_another_backend_is_not_an_update_of_this_one(self):
self.publish("0.1.49.1")
ctx, state, feed = self.run_step()
with mock.patch.object(dsv41_paths, "kind_of", side_effect=lambda n: "windows-x86_64-hip-rdna3" if n.startswith("0.1.49-") else "x"):
r = updater.update_dsv41(ctx, feed, state)
self.assertNotEqual(r["status"], "updated")
self.assertEqual(self.folders(), ["0.1.49"])
def test_the_update_step_is_part_of_run_update_with_the_engine(self):
self.publish("0.1.49.1")
ctx = self.ctx()
with mock.patch.object(updater, "update_engine", return_value={"component": "engine", "status": "current", "changed": False}), \
mock.patch.object(updater, "decide_engine", return_value={"status": "current", "offered": None, "installed": None}), \
mock.patch.object(updater, "decide_app", return_value={"status": "not-in-feed", "offered": None}):
out = updater.run_update(ctx, engine=True, control=False)
names = [r["component"] for r in out["results"]]
self.assertEqual(names, ["engine", "dsv41"], out)
self.assertEqual(out["results"][1]["status"], "updated")
with mock.patch.object(updater, "decide_engine", return_value={"status": "current", "offered": None, "installed": None}), \
mock.patch.object(updater, "decide_app", return_value={"status": "not-in-feed", "offered": None}):
out = updater.run_update(ctx, engine=False, control=False)
self.assertEqual(out["results"], [], "--no-engine leaves the DeepSeek engine alone too")
# ---- what the check keeps for the app's one card --------------------------------------------------------------------
def test_the_check_says_a_newer_deepseek_engine_is_there_and_the_app_can_read_it_without_the_network(self):
self.publish("0.1.49.1")
ctx = self.ctx()
with mock.patch.object(updater, "decide_engine", return_value={"status": "current", "offered": None, "installed": None}), \
mock.patch.object(updater, "decide_control", return_value={"status": "not-in-feed", "installed": None, "offered": None}), \
mock.patch.object(updater, "decide_app", return_value={"status": "not-in-feed", "offered": None}):
report = updater.check(ctx)
self.assertEqual((report["dsv41"]["status"], report["dsv41"]["installed"], report["dsv41"]["offered"]), ("available", "0.1.49", "0.1.49.1"))
with mock.patch.object(urllib.request, "urlopen", side_effect=AssertionError("network")), mock.patch.object(socket, "create_connection", side_effect=AssertionError("network")):
offers = ec.cached_offers(ctx.components)
self.assertEqual(offers["dsv41"], "0.1.49.1")
self.assertIn("DeepSeek engine 0.1.49.1", ec.cached_notice(ctx.components))
self.assertEqual(ec.status(ctx.components)["update_available"]["dsv41"], "0.1.49.1")
state = updater.State(ctx.components)
feed = updater.open_feed(ctx, state)
self.assertEqual(updater.update_dsv41(ctx, feed, state)["status"], "updated")
self.assertIsNone(ec.cached_offers(ctx.components)["dsv41"], "installed: nothing to offer any more")
# ---- the start ----------------------------------------------------------------------------------------------------------
def test_a_start_repairs_a_config_whose_folder_an_update_removed_and_touches_the_network_not_at_all(self):
self.publish("0.1.49.1")
ctx, state, feed = self.run_step()
self.assertEqual(updater.update_dsv41(ctx, feed, state)["status"], "updated")
new = dsv41_paths.installed(self.dest)[-1]
self.write_cfg(self.cfg2, self.old) # (a config written again from a model's older file: its folder is gone)
self.assertFalse(self.old.exists())
log = self.root / "run" / "app-1.host.log"
profile = {"source_config": str(self.cfg2)}
with mock.patch.object(urllib.request, "urlopen", side_effect=AssertionError("a start reached the network")), \
mock.patch.object(socket, "create_connection", side_effect=AssertionError("a start opened a connection")), \
mock.patch("subprocess.Popen", side_effect=AssertionError("a start ran a program")), \
mock.patch.object(runtime.profiles, "load", return_value=(profile, "r")), \
mock.patch.object(runtime, "_paths", return_value=(None, None, log)):
runtime._repair_dsv41_paths("app-1")
self.assertEqual(self.exe_of(self.cfg2), new)
self.assertIn("now uses 0.1.49.1", log.read_text(encoding="utf-8"))
def test_a_start_leaves_a_config_whose_engine_is_there_and_the_feed_unread(self):
self.publish("0.1.49.1")
before = self.cfg2.read_bytes()
with mock.patch.object(urllib.request, "urlopen", side_effect=AssertionError("network")), \
mock.patch.object(runtime.profiles, "load", return_value=({"source_config": str(self.cfg2)}, "r")):
runtime._repair_dsv41_paths("app-1")
self.assertEqual(self.cfg2.read_bytes(), before)
self.assertEqual(self.folders(), ["0.1.49"], "a newer package in the feed is not fetched by a start")
def test_the_start_side_modules_do_not_reach_the_updater_the_feed_or_the_launch_upgrade(self):
import ast
for name in ("runtime.py", "dsv41_paths.py"):
tree = ast.parse((Path(__file__).resolve().parents[2] / "strata_control" / name).read_text(encoding="utf-8"))
imported = set()
for node in ast.walk(tree):
if isinstance(node, ast.Import):
imported.update(a.name for a in node.names)
elif isinstance(node, ast.ImportFrom) and node.module:
imported.add(node.module)
imported.update(f"{node.module}.{a.name}" for a in node.names)
for bad in ("updater", "dsv41_engine", "feed", "transfer", "components", "urllib.request", "http.client"):
self.assertFalse([i for i in imported if i == bad or i.endswith("." + bad)], f"{name} imports {bad}")
def test_a_config_that_is_not_ours_is_left_alone(self):
for exe in ("@component:strata-engine", str(self.root / "mine" / EXE)):
c = self.root / "run" / "own.json"
c.write_text(json.dumps({"exe": exe, "args": []}), encoding="utf-8")
self.assertEqual(dsv41_paths.repair_missing([c]), [])
self.assertEqual(json.loads(c.read_text(encoding="utf-8"))["exe"], exe)
c = self.root / "run" / "broken.json"
c.write_text("{not json", encoding="utf-8")
self.assertEqual(dsv41_paths.repair_missing([c, self.root / "run" / "missing.json"]), [])
for _name in dir(_up.Upgrade): # (the parent's own tests are not repeated here)
if _name.startswith("test_") and _name not in UpdatePath.__dict__:
setattr(UpdatePath, _name, None)
if __name__ == "__main__":
unittest.main()