Merge bc9c172b3c into 20e7fc0164

routes/cookbook_routes.py

@@ -47,7 +47,7 @@ from routes.cookbook_output import (
 logger = logging.getLogger(__name__)
 
 from routes.cookbook_helpers import (
-    _SESSION_ID_RE, _validate_repo_id, _validate_serve_model_id, _validate_include, _validate_token,
+    _SESSION_ID_RE, _REPO_ID_RE, _validate_repo_id, _validate_serve_model_id, _validate_include, _validate_token,
     _validate_local_dir, _validate_gpus, _shell_path,
     _ps_squote, _bash_squote, _validate_serve_cmd, _parse_serve_phase, OLLAMA_MISSING_HINT,
     _safe_env_prefix, _local_tooling_path_export, _append_serve_preflight_exit_lines,
@@ -370,6 +370,35 @@ def _append_local_ollama_download_command_lines(
     lines.append('if [ -z "$ODYSSEUS_OLLAMA_PULL_CMD" ]; then echo "ERROR: Ollama not found on this server. Install Ollama or start an ollama-rocm/ollama-test container."; exit 127; fi')
 
 
+def _cmdline_references_hf_repo(text: str, repo_id: str) -> bool:
+    """True when text references an HF download of repo_id (exact id, not a prefix)."""
+    if not text or not repo_id:
+        return False
+    boundary = r"(?![A-Za-z0-9._-])"
+    escaped = re.escape(repo_id)
+    patterns = (
+        # hf / hf.exe / "…\hf.exe" / '…\hf.exe' download <repo>
+        # Optional closing quote after .exe — Windows often quotes the executable path.
+        rf"\bhf(?:\.exe)?[\"']?\s+download\s+{escaped}{boundary}",
+        # hf_download.py script arg — quoted/unquoted, optional python -u prefix
+        rf"\bhf_download\.py[\"']?\s+{escaped}{boundary}",
+        # Python huggingface_hub fallback — positional and repo_id= forms.
+        # Require the matching closing quote plus ',' or ')' so org/model
+        # never matches a concurrently downloading org/model-large.
+        rf"\bsnapshot_download\s*\(\s*([\"']){escaped}\1\s*(?:,|\))",
+        rf"\bsnapshot_download\s*\(\s*repo_id\s*=\s*([\"']){escaped}\1\s*(?:,|\))",
+    )
+    return any(re.search(p, text, re.IGNORECASE) for p in patterns)
+
+
+def _coerce_ssh_port(v: str | None) -> str | None:
+    """Non-throwing ssh port check; returns normalized port or None."""
+    try:
+        return validate_ssh_port(None if v in (None, "") else str(v))
+    except HTTPException:
+        return None
+
+
 def setup_cookbook_routes() -> APIRouter:
     router = APIRouter(tags=["cookbook"])
     _cookbook_state_path = Path(COOKBOOK_STATE_FILE)
@@ -1023,6 +1052,473 @@ def setup_cookbook_routes() -> APIRouter:
         pid_path.write_text(str(proc.pid), encoding="utf-8")
         return {"pid": proc.pid, "log_path": str(log_path)}
 
+    def _windows_download_process_scan_ps() -> str:
+        return (
+            "Get-CimInstance Win32_Process -Filter "
+            "\"Name='python.exe' OR Name='hf.exe'\" | "
+            "ForEach-Object { \"$($_.ProcessId)`t$($_.CommandLine)\" }"
+        )
+
+    def _parse_windows_download_scan(stdout: str, repo_id: str) -> int | None:
+        for line in (stdout or "").splitlines():
+            pid_s, _, cmdline = line.partition("\t")
+            if _cmdline_references_hf_repo(cmdline, repo_id):
+                try:
+                    return int(pid_s.strip())
+                except ValueError:
+                    continue
+        return None
+
+    def _scan_windows_download_processes_checked(
+        repo_id: str,
+    ) -> tuple[bool, int | None]:
+        """Return (scan_succeeded, downloader_pid) for a local Windows repo."""
+        try:
+            probe = subprocess.run(
+                ["powershell", "-NoProfile", "-Command", _windows_download_process_scan_ps()],
+                capture_output=True,
+                text=True,
+                timeout=15,
+                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
+            )
+        except Exception:
+            return False, None
+        if probe.returncode != 0:
+            return False, None
+        out = probe.stdout or ""
+        return True, _parse_windows_download_scan(out, repo_id)
+
+    def _scan_windows_download_processes(repo_id: str) -> int | None:
+        """PID of any live process downloading repo_id (Windows, best-effort).
+
+        Catches downloaders that cookbook session files no longer track —
+        e.g. a stop issued from a stale browser tab running pre-tree-kill JS
+        deleted the .pid file but left the hf/python children running. Those
+        orphans hold the HF cache file locks and deadlock any new download of
+        the same repo.
+        """
+        _scan_ok, pid = _scan_windows_download_processes_checked(repo_id)
+        return pid
+
+    def _ssh_powershell_argv(remote: str, ssh_port: str | None, ps: str) -> list[str]:
+        ssh_args = [
+            "ssh",
+            "-o", "ConnectTimeout=5",
+            "-o", "StrictHostKeyChecking=no",
+        ]
+        safe_port = _coerce_ssh_port(ssh_port)
+        if safe_port and safe_port != "22":
+            ssh_args.extend(["-p", safe_port])
+        ssh_args.extend([remote, "powershell", "-NoProfile", "-Command", ps])
+        return ssh_args
+
+    async def _ssh_powershell(remote: str, ssh_port: str | None, ps: str) -> tuple[int, str]:
+        """Run inline PowerShell on a remote Windows host via ssh argv (no local shell)."""
+        try:
+            proc = await asyncio.create_subprocess_exec(
+                *_ssh_powershell_argv(remote, ssh_port, ps),
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            stdout, _stderr = await asyncio.wait_for(proc.communicate(), timeout=20)
+        except Exception:
+            return 1, ""
+        return proc.returncode or 0, (stdout or b"").decode("utf-8", errors="replace")
+
+    async def _scan_remote_windows_download_processes(
+        remote: str, ssh_port: str | None, repo_id: str
+    ) -> int | None:
+        """PID of a live remote-Windows downloader for repo_id (best-effort)."""
+        _rc, out = await _ssh_powershell(remote, ssh_port, _windows_download_process_scan_ps())
+        return _parse_windows_download_scan(out, repo_id)
+
+    def _local_ps1_sessions_for_repo(repo_id: str) -> list[str]:
+        """Session ids whose local _run.ps1 references an HF download of repo_id."""
+        try:
+            runners = sorted(TMUX_LOG_DIR.glob("cookbook-*_run.ps1"))
+        except OSError:
+            return []
+        sids: list[str] = []
+        for path in runners:
+            sid = path.name.removesuffix("_run.ps1")
+            if not _SESSION_ID_RE.match(sid):
+                continue
+            try:
+                text = path.read_text(encoding="utf-8")
+            except OSError:
+                continue
+            if _cmdline_references_hf_repo(text, repo_id):
+                sids.append(sid)
+        return sids
+    def _server_platform_for_host(remote_host: str | None) -> str:
+        """Platform string from cookbook server profiles for a remote host."""
+        if not remote_host or not _cookbook_state_path.exists():
+            return ""
+        try:
+            state = json.loads(_cookbook_state_path.read_text(encoding="utf-8"))
+        except Exception:
+            return ""
+        env_state = state.get("env") if isinstance(state, dict) else {}
+        servers = env_state.get("servers") if isinstance(env_state, dict) else []
+        if not isinstance(servers, list):
+            return ""
+        for server in servers:
+            if isinstance(server, dict) and (server.get("host") or "").strip() == remote_host:
+                return (server.get("platform") or "").strip().lower()
+        return ""
+
+    def _resolve_windows_platform(remote_host: str | None, platform: str | None) -> str:
+        plat = (platform or "").strip().lower()
+        if remote_host:
+            configured = _server_platform_for_host(remote_host)
+            return configured or plat
+        return plat or ("windows" if IS_WINDOWS else "")
+
+    def _unlink_session_artifacts(session_id: str) -> None:
+        for pattern in (f"{session_id}.*", f"{session_id}_run.*"):
+            for path in TMUX_LOG_DIR.glob(pattern):
+                # Keep the per-session .stop marker — the bash retry loop
+                # checks it between attempts.
+                if path.suffix == ".stop":
+                    continue
+                try:
+                    path.unlink()
+                except OSError:
+                    pass
+
+    _STOPPED_REPOS_PATH = TMUX_LOG_DIR / "cookbook-stopped-repos.json"
+
+    def _read_stopped_repos() -> set[str]:
+        try:
+            text = _STOPPED_REPOS_PATH.read_text(encoding="utf-8").strip()
+            if not text:
+                return set()
+            # Legacy JSON list from earlier builds — still readable on upgrade.
+            if text.startswith("["):
+                data = json.loads(text)
+                if isinstance(data, list):
+                    return {str(x) for x in data if x}
+            return {line.strip() for line in text.splitlines() if line.strip()}
+        except Exception:
+            pass
+        return set()
+
+    def _write_stopped_repos(repos: set[str]) -> None:
+        TMUX_LOG_DIR.mkdir(parents=True, exist_ok=True)
+        _STOPPED_REPOS_PATH.write_text(
+            "\n".join(sorted(repos)) + ("\n" if repos else ""),
+            encoding="utf-8",
+        )
+
+    def _mark_download_stopped(repo_id: str) -> None:
+        if not repo_id:
+            return
+        repos = _read_stopped_repos()
+        repos.add(repo_id)
+        _write_stopped_repos(repos)
+
+    def _clear_download_stopped(repo_id: str) -> None:
+        if not repo_id:
+            return
+        repos = _read_stopped_repos()
+        if repo_id in repos:
+            repos.remove(repo_id)
+            _write_stopped_repos(repos)
+
+    def _session_stop_file(session_id: str) -> Path:
+        return TMUX_LOG_DIR / f"{session_id}.stop"
+
+    def _write_session_stop_marker(session_id: str, repo_id: str | None = None) -> None:
+        TMUX_LOG_DIR.mkdir(parents=True, exist_ok=True)
+        try:
+            _session_stop_file(session_id).write_text("1", encoding="utf-8")
+        except OSError:
+            pass
+        if repo_id:
+            _mark_download_stopped(repo_id)
+
+    def _bash_download_stop_guard(session_id: str, repo_id: str | None = None) -> list[str]:
+        """Bash lines that honour a user stop between download attempts."""
+        stop = shlex.quote(_session_stop_file(session_id).as_posix())
+        lines = [
+            f"_ODYSSEUS_STOP_FILE={stop}",
+            f"_ODYSSEUS_STOPPED_REPOS={shlex.quote(_STOPPED_REPOS_PATH.as_posix())}",
+            "trap 'echo \"\"; echo \"DOWNLOAD_STOPPED\"; exit 130' INT TERM",
+        ]
+        if repo_id:
+            lines.append(f"_ODYSSEUS_REPO={shlex.quote(repo_id)}")
+        return lines
+
+    def _bash_download_attempt_guard() -> str:
+        return (
+            '  if [ -f "$_ODYSSEUS_STOP_FILE" ]; then '
+            'echo ""; echo "DOWNLOAD_STOPPED"; exit 130; fi; '
+            'if [ -n "${_ODYSSEUS_REPO:-}" ] && [ -f "${_ODYSSEUS_STOPPED_REPOS:-}" ] '
+            '&& grep -Fxq "$_ODYSSEUS_REPO" "$_ODYSSEUS_STOPPED_REPOS" 2>/dev/null; then '
+            'echo ""; echo "DOWNLOAD_STOPPED"; exit 130; fi'
+        )
+
+    def _find_live_local_download(repo_id: str) -> dict | None:
+        """Live LOCAL download of this repo, if any (honours user-stop markers)."""
+        if repo_id in _read_stopped_repos():
+            live = _probe_live_local_download(repo_id)
+            if live:
+                if live.get("session_id"):
+                    _stop_local_windows_session(live["session_id"], repo_id)
+                elif live.get("orphan_pid") and pid_alive(live["orphan_pid"]):
+                    kill_process_tree(live["orphan_pid"])
+            return None
+        return _probe_live_local_download(repo_id)
+
+    def _probe_live_local_download(repo_id: str) -> dict | None:
+        """Probe for a live LOCAL download without honouring user-stop markers."""
+        try:
+            sids = sorted({
+                p.stem.removesuffix("_run")
+                for p in TMUX_LOG_DIR.glob("cookbook-*.sh")
+            })
+        except OSError:
+            return None
+        for sid in sids:
+            if not _SESSION_ID_RE.match(sid):
+                continue
+            script = TMUX_LOG_DIR / (f"{sid}_run.sh" if IS_WINDOWS else f"{sid}.sh")
+            try:
+                script_text = script.read_text(encoding="utf-8")
+            except OSError:
+                continue
+            if not _cmdline_references_hf_repo(script_text, repo_id):
+                continue
+            if IS_WINDOWS:
+                try:
+                    pid = int((TMUX_LOG_DIR / f"{sid}.pid").read_text(encoding="utf-8").strip())
+                except (OSError, ValueError):
+                    continue
+                if pid_alive(pid):
+                    return {"session_id": sid}
+            else:
+                try:
+                    probe = subprocess.run(
+                        ["tmux", "has-session", "-t", sid],
+                        capture_output=True,
+                        timeout=5,
+                    )
+                except Exception:
+                    continue
+                if probe.returncode == 0:
+                    return {"session_id": sid}
+        if IS_WINDOWS:
+            orphan_pid = _scan_windows_download_processes(repo_id)
+            if orphan_pid:
+                return {"orphan_pid": orphan_pid}
+        return None
+    def _scan_windows_session_pids(session_id: str) -> list[int] | None:
+        """PIDs whose command line references this cookbook session's wrappers."""
+        if not session_id:
+            return []
+        sid = session_id.replace("'", "''")
+        ps = (
+            "Get-CimInstance Win32_Process | "
+            f"Where-Object {{ $_.CommandLine -and $_.CommandLine -like '*{sid}*' }} | "
+            "ForEach-Object { $_.ProcessId }"
+        )
+        try:
+            probe = subprocess.run(
+                ["powershell", "-NoProfile", "-Command", ps],
+                capture_output=True,
+                text=True,
+                timeout=15,
+                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
+            )
+        except Exception:
+            return None
+        if probe.returncode != 0:
+            return None
+        out = probe.stdout or ""
+        pids: list[int] = []
+        for line in out.splitlines():
+            try:
+                pids.append(int(line.strip()))
+            except ValueError:
+                continue
+        return pids
+
+    def _kill_local_windows_pid(pid: int) -> bool:
+        """Kill a Windows process tree and verify that its root exited."""
+        try:
+            result = subprocess.run(
+                ["taskkill", "/F", "/T", "/PID", str(pid)],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                timeout=15,
+                creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0),
+            )
+        except Exception:
+            return False
+        if result.returncode != 0:
+            return False
+        for _ in range(10):
+            if not pid_alive(pid):
+                return True
+            time.sleep(0.05)
+        return not pid_alive(pid)
+
+    def _stop_local_windows_session(session_id: str, repo_id: str | None = None) -> dict:
+        """Kill a local Windows detached cookbook session and orphan downloaders."""
+        pid_path = TMUX_LOG_DIR / f"{session_id}.pid"
+        found_process = False
+        detail: list[str] = []
+        try:
+            pid = int(pid_path.read_text(encoding="utf-8").strip())
+            if pid_alive(pid):
+                found_process = True
+                if not _kill_local_windows_pid(pid):
+                    return {
+                        "ok": False,
+                        "stopped": False,
+                        "error": f"failed to stop local Windows process tree pid {pid}",
+                        "detail": "recovery artifacts preserved",
+                    }
+                detail.append(f"killed pid {pid}")
+        except (OSError, ValueError):
+            pass
+        for _ in range(3):
+            scanned_pids = _scan_windows_session_pids(session_id)
+            if scanned_pids is None:
+                return {
+                    "ok": False,
+                    "stopped": False,
+                    "error": "failed to scan local Windows session processes",
+                    "detail": "recovery artifacts preserved",
+                }
+            session_pids = sorted({p for p in scanned_pids if pid_alive(p)})
+            if not session_pids:
+                break
+            for pid in session_pids:
+                found_process = True
+                if not _kill_local_windows_pid(pid):
+                    return {
+                        "ok": False,
+                        "stopped": False,
+                        "error": f"failed to stop local Windows session process pid {pid}",
+                        "detail": "recovery artifacts preserved",
+                    }
+                detail.append(f"killed session process pid {pid}")
+        if repo_id:
+            for _ in range(3):
+                scan_ok, orphan = _scan_windows_download_processes_checked(repo_id)
+                if not scan_ok:
+                    return {
+                        "ok": False,
+                        "stopped": False,
+                        "error": "failed to scan local Windows download processes",
+                        "detail": "recovery artifacts preserved",
+                    }
+                if not orphan or not pid_alive(orphan):
+                    break
+                found_process = True
+                if not _kill_local_windows_pid(orphan):
+                    return {
+                        "ok": False,
+                        "stopped": False,
+                        "error": f"failed to stop local Windows orphan downloader pid {orphan}",
+                        "detail": "recovery artifacts preserved",
+                    }
+                detail.append(f"killed orphan downloader pid {orphan}")
+        # Already-dead sessions are success (same as tmux "session not found"):
+        # scans succeeded and nothing live remains. Only kill/scan failures fail.
+        _write_session_stop_marker(session_id, repo_id)
+        _unlink_session_artifacts(session_id)
+        return {
+            "ok": True,
+            "stopped": found_process,
+            "detail": "; ".join(detail) or "session already gone",
+        }
+
+    async def _stop_remote_windows_session(
+        session_id: str,
+        remote: str,
+        ssh_port: str | None = None,
+        repo_id: str | None = None,
+    ) -> dict:
+        """PR1: remote Windows stop is out of scope — fail closed."""
+        return {
+            "ok": False,
+            "stopped": False,
+            "error": "remote Windows stop-session is not supported in this build",
+            "detail": "recovery artifacts preserved",
+        }
+
+    def _tmux_stop_succeeded(returncode: int, stderr: bytes | str = b"") -> bool:
+        if returncode == 0:
+            return True
+        err = (
+            stderr.decode("utf-8", errors="replace")
+            if isinstance(stderr, bytes)
+            else str(stderr)
+        ).lower()
+        return any(
+            s in err
+            for s in ("no server running", "can't find session", "session not found")
+        )
+
+    async def _stop_cookbook_session_impl(
+        session_id: str,
+        remote_host: str = "",
+        ssh_port: str | None = None,
+        platform: str = "",
+        repo_id: str | None = None,
+    ) -> dict:
+        if not _SESSION_ID_RE.match(session_id):
+            return {"ok": False, "error": "invalid session_id"}
+        remote = (remote_host or "").strip()
+        sport = ssh_port or ""
+        is_win = _resolve_windows_platform(remote or None, platform) == "windows"
+        if remote:
+            if is_win:
+                return await _stop_remote_windows_session(session_id, remote, sport, repo_id)
+            _write_session_stop_marker(session_id, repo_id)
+            sid = shlex.quote(session_id)
+            ssh_args = [
+                "ssh",
+                "-o", "ConnectTimeout=5",
+                "-o", "StrictHostKeyChecking=no",
+            ]
+            safe_port = _coerce_ssh_port(sport)
+            if safe_port and safe_port != "22":
+                ssh_args.extend(["-p", safe_port])
+            ssh_args.append(remote)
+            ssh_args.append(
+                f"tmux has-session -t {sid} 2>/dev/null || exit 0; "
+                f"tmux send-keys -t {sid} C-c 2>/dev/null; "
+                f"sleep 2; tmux kill-session -t {sid}"
+            )
+            proc = await asyncio.create_subprocess_exec(
+                *ssh_args,
+                stdout=asyncio.subprocess.PIPE,
+                stderr=asyncio.subprocess.PIPE,
+            )
+            _stdout, stderr = await proc.communicate()
+            ok = _tmux_stop_succeeded(proc.returncode, stderr)
+            return {"ok": ok, "exit_code": proc.returncode}
+        if IS_WINDOWS or is_win:
+            return await asyncio.to_thread(_stop_local_windows_session, session_id, repo_id)
+        _write_session_stop_marker(session_id, repo_id)
+        sid = shlex.quote(session_id)
+        cmd = (
+            f"tmux has-session -t {sid} 2>/dev/null || exit 0; "
+            f"tmux send-keys -t {sid} C-c 2>/dev/null; "
+            f"sleep 2; tmux kill-session -t {sid}"
+        )
+        proc = await asyncio.create_subprocess_shell(
+            cmd,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        _stdout, stderr = await proc.communicate()
+        ok = _tmux_stop_succeeded(proc.returncode, stderr)
+        return {"ok": ok, "exit_code": proc.returncode}
+
     @router.post("/api/model/download")
     async def model_download(request: Request, req: ModelDownloadRequest):
         """Download a HuggingFace model in a tmux session.
@@ -1045,6 +1541,46 @@ def setup_cookbook_routes() -> APIRouter:
         req.local_dir = _validate_local_dir(req.local_dir)
         req.hf_token = "" if is_ollama_download else (req.hf_token or _load_stored_hf_token())
         _validate_token(req.hf_token)
+
+        if not is_ollama_download:
+            # Explicit launch from the UI/agent — clear any prior user-stop
+            # marker so Retry works, while background auto-reattach stays off.
+            _clear_download_stopped(req.repo_id)
+        # Concurrent downloads of the same repo deadlock on the HF cache's
+        # per-file locks ("Still waiting to acquire lock..."). If a live local
+        # session is already downloading this repo, reattach the UI to it instead
+        # of launching a duplicate. Remote POSIX hosts are covered by the
+        # frontend's tmux has-session zombie probe; Ollama serializes pulls in
+        # its own daemon.
+        if not is_ollama_download and not (req.remote_host or "").strip():
+            live = await asyncio.to_thread(_find_live_local_download, req.repo_id)
+            if live and live.get("session_id"):
+                live_sid = live["session_id"]
+                logger.info(
+                    f"Download of {req.repo_id} already running locally in {live_sid}; reattaching"
+                )
+                return {
+                    "ok": True,
+                    "session_id": live_sid,
+                    "remote": "local",
+                    "reused": True,
+                }
+            if live and live.get("orphan_pid"):
+                pid = live["orphan_pid"]
+                logger.warning(
+                    f"Download of {req.repo_id} blocked: untracked downloader process "
+                    f"pid={pid} is live locally"
+                )
+                return {
+                    "ok": False,
+                    "error": (
+                        f"Another process (pid {pid}) is already downloading {req.repo_id} "
+                        "on local outside cookbook tracking — likely left over from an "
+                        "earlier stop. Wait for it to finish, or end its process tree (e.g. "
+                        f"taskkill /F /T /PID {pid}) and retry; the download resumes from cache."
+                    ),
+                    "session_id": "",
+                }
         if req.remote_host and not req.env_prefix:
             req.env_prefix = _server_env_prefix_for_download(req.remote_host)
         TMUX_LOG_DIR.mkdir(parents=True, exist_ok=True)
@@ -1260,12 +1796,14 @@ def setup_cookbook_routes() -> APIRouter:
                 # download's "not authorized" failure can be told apart from a missing
                 # token (the token is masked — we only print applied / not-set).
                 runner_lines.append(_HF_TOKEN_STATUS_SNIPPET)
-            # Wrap the download in a retry loop. Large HF/Ollama transfers can
-            # hit transient network failures; both backends resume cached partials.
+            # Retry transient failures, but honour explicit user stops between
+            # attempts so a stopped download cannot silently resume.
             mw = 4 if req.disable_hf_transfer else 8
+            runner_lines.extend(_bash_download_stop_guard(session_id, req.repo_id))
             runner_lines.append('_max_retries=10; _attempt=0; _ec=0')
             runner_lines.append('while [ $_attempt -lt $_max_retries ]; do')
             runner_lines.append('  _attempt=$((_attempt+1))')
+            runner_lines.append(_bash_download_attempt_guard())
             if is_ollama_download:
                 runner_lines.append('  eval "$ODYSSEUS_OLLAMA_PULL_CMD" < /dev/null')
             else:
@@ -1275,6 +1813,7 @@ def setup_cookbook_routes() -> APIRouter:
             runner_lines.append('  if [ $_attempt -lt $_max_retries ]; then')
             runner_lines.append('    echo ""; echo "Download attempt $_attempt failed (exit $_ec) — retrying in 30s..."')
             runner_lines.append('    sleep 30')
+            runner_lines.append(_bash_download_attempt_guard())
             runner_lines.append('  fi')
             runner_lines.append('done')
             runner_lines.append('if [ $_ec -eq 0 ]; then echo ""; echo "DOWNLOAD_OK"; else echo ""; echo "DOWNLOAD_FAILED (exit $_ec after $_attempt attempts)"; fi')
@@ -1305,17 +1844,21 @@ def setup_cookbook_routes() -> APIRouter:
             # "not authorized" failure apart from a missing token.
             if not is_ollama_download:
                 lines.append(_HF_TOKEN_STATUS_SNIPPET)
-            # Retry loop — same rationale as the remote-bash path. Issue #2722.
+            # Retry transient failures, but honour explicit user stops between
+            # attempts so a stopped download cannot silently resume.
             _hf_invoke = 'eval "$ODYSSEUS_OLLAMA_PULL_CMD" < /dev/null' if is_ollama_download else (hf_cmd if IS_WINDOWS else f"{hf_cmd} < /dev/null")
+            lines.extend(_bash_download_stop_guard(session_id, req.repo_id))
             lines.append('_max_retries=10; _attempt=0; _ec=0')
             lines.append('while [ $_attempt -lt $_max_retries ]; do')
             lines.append('  _attempt=$((_attempt+1))')
+            lines.append(_bash_download_attempt_guard())
             lines.append(f'  {_hf_invoke}')
             lines.append('  _ec=$?')
             lines.append('  if [ $_ec -eq 0 ]; then break; fi')
             lines.append('  if [ $_attempt -lt $_max_retries ]; then')
             lines.append('    echo ""; echo "Download attempt $_attempt failed (exit $_ec) — retrying in 30s..."')
             lines.append('    sleep 30')
+            lines.append(_bash_download_attempt_guard())
             lines.append('  fi')
             lines.append('done')
             lines.append('if [ $_ec -eq 0 ]; then echo ""; echo "DOWNLOAD_OK"; else echo ""; echo "DOWNLOAD_FAILED (exit $_ec after $_attempt attempts)"; fi')
@@ -3269,6 +3812,46 @@ def setup_cookbook_routes() -> APIRouter:
 
         return {"ok": False, "error": nvidia_error or "No GPU memory probe available", "gpus": []}
 
+    class StopSessionRequest(BaseModel):
+        session_id: str
+        remote_host: str | None = None
+        ssh_port: str | None = None
+        platform: str | None = None
+        repo_id: str | None = None
+        # download | serve | dependency | … — gates download-only stop side effects
+        task_type: str | None = None
+
+    @router.post("/api/cookbook/stop-session")
+    async def stop_cookbook_session(request: Request, req: StopSessionRequest):
+        """Stop a cookbook download/serve session (local or remote, all platforms).
+
+        Centralizes kill logic server-side so the UI does not have to guess
+        whether a local task is tmux-backed or a Windows detached process tree.
+        """
+        require_admin(request)
+        validate_remote_host(req.remote_host)
+        sport = validate_ssh_port(req.ssh_port)
+        repo_id_raw = (req.repo_id or "").strip()
+        task_type = (req.task_type or "").strip().lower()
+        # repo_id is optional download-only metadata for stopped-repo markers and
+        # orphan downloader cleanup. Never validate it before kill — dependency
+        # rows store pip labels such as llama-cpp-python[server] in payload.repo_id.
+        # Serve tasks also carry HF-shaped repo_id; only honour it for downloads.
+        repo_id_for_stop = (
+            repo_id_raw
+            if task_type == "download" and repo_id_raw and _REPO_ID_RE.match(repo_id_raw)
+            else None
+        )
+        platform = _resolve_windows_platform(req.remote_host, req.platform)
+        return await _stop_cookbook_session_impl(
+            req.session_id.strip(),
+            remote_host=req.remote_host or "",
+            ssh_port=sport,
+            platform=platform,
+            repo_id=repo_id_for_stop,
+        )
+
+
     class KillPidRequest(BaseModel):
         pid: int
         host: str | None = None

src/agent_loop.py

@@ -12,7 +12,7 @@ import json
 import re
 import time
 import logging
-from typing import Any, AsyncGenerator, List, Dict, Optional, Set
+from typing import Any, AsyncGenerator, Dict, List, Optional, Set
 from urllib.parse import urlparse
 
 from src.llm_core import (

src/tools/system.py

@@ -557,6 +557,7 @@ _APP_API_BLOCKLIST_METHOD_PATH = (
     ("POST",   "/api/cookbook/packages/install"),
     ("POST",   "/api/cookbook/rebuild-engine"),
     ("POST",   "/api/cookbook/kill-pid"),
+    ("POST",   "/api/cookbook/stop-session"),
     # Use the named tools (download_model / serve_model) — they handle
     # host-name resolution, per-host env_prefix, AND register the task
     # in cookbook state so it shows in the UI + list_downloads. Hitting
@@ -681,6 +682,8 @@ async def do_app_api(content: str, owner: Optional[str] = None) -> Dict:
             return {"error": "Don't POST /api/cookbook/rebuild-engine via app_api — engine rebuild mutates local or remote host state. Use the dedicated Cookbook UI/flow instead.", "exit_code": 1}
         if "/api/cookbook/kill-pid" in path:
             return {"error": "Don't POST /api/cookbook/kill-pid via app_api — process signalling is host control. Use the dedicated Cookbook stop/diagnostic flow instead.", "exit_code": 1}
+        if "/api/cookbook/stop-session" in path:
+            return {"error": "Don't POST /api/cookbook/stop-session via app_api — session stop is host process control. Use stop_served_model, cancel_download, or the Cookbook UI stop flow instead.", "exit_code": 1}
         if "/api/model/download" in path:
             return {"error": "Don't POST /api/model/download directly — use the `download_model` tool (it resolves the server name, sets the venv env_prefix, and registers the task so it shows in the UI).", "exit_code": 1}
         if "/api/model/serve" in path:

static/js/cookbookRunning.js

@@ -11,6 +11,8 @@ import { computeProgressSignal } from './cookbookProgressSignal.js';
 import { portOf, nextFreePort } from './cookbookPorts.js';
 import { topPortalZ } from './toolWindowZOrder.js';
 
+const _RECONNECT_STATUSES = ['running', 'ready', 'loading', 'warming', 'starting'];
+
 // Human-friendly badge label for a task's internal status. Avoids surfacing
 // the word "error" in the sidebar — a server the user stopped or one that
 // quit cleanly reads as "stopped", not "error".
@@ -52,6 +54,14 @@ function _taskBadge(task) {
   return { text: _statusLabel(task.status, task.type), cls: 'cookbook-task-' + task.status };
 }
 
+function _applyTaskBadge(badge, task, progressText) {
+  if (!badge) return;
+  const view = progressText != null ? { ...task, progress: progressText } : task;
+  const bdg = _taskBadge(view);
+  badge.textContent = bdg.text;
+  badge.className = 'cookbook-task-status' + (bdg.cls ? ` ${bdg.cls}` : '');
+}
+
 function _ggufDisplayPartFromPath(path) {
   const parts = String(path || '').split('/').filter(Boolean);
   const file = parts[parts.length - 1] || '';
@@ -119,7 +129,7 @@ function _downloadOutputLooksActive(task) {
   if (!task || task.type !== 'download') return false;
   const out = task.output || '';
   if (!out) return false;
-  if (out.includes('DOWNLOAD_OK') || out.includes('DOWNLOAD_FAILED')) return false;
+  if (out.includes('DOWNLOAD_OK') || out.includes('DOWNLOAD_FAILED') || out.includes('DOWNLOAD_STOPPED')) return false;
   // An active shard line: filename + a colon + a percentage that isn't 100%.
   // We catch any in-flight shard or "Downloading 'X' to ..." line (no %).
   return /model-\d+-of-\d+\.[a-z]+:\s+(?!100%)\d+%/i.test(out)
@@ -996,6 +1006,117 @@ function _animateOutThenRemove(el, sessionId) {
   setTimeout(() => _removeTask(sessionId), 360);
 }
 
+function _applyStoppedTaskCard(el, task) {
+  const badge = el.querySelector('.cookbook-task-status');
+  if (badge) {
+    badge.textContent = _statusLabel('stopped', task.type);
+    badge.className = 'cookbook-task-status cookbook-task-stopped';
+  }
+  el.dataset.status = 'stopped';
+  const pre = el.querySelector('.cookbook-output-pre');
+  const output = pre?.textContent || task.output || '';
+  const stamped = task.type === 'download' && !output.includes('DOWNLOAD_STOPPED')
+    ? (output.trimEnd() ? output.trimEnd() + '\n\n' : '') + 'DOWNLOAD_STOPPED'
+    : output;
+  if (pre && stamped !== output) pre.textContent = stamped;
+  _updateTask(task.sessionId, { status: 'stopped', _userStopped: true, output: stamped || output });
+  const wave = el.querySelector('.cookbook-task-wave');
+  if (wave) wave.style.display = 'none';
+  const uptime = el.querySelector('.cookbook-task-uptime');
+  if (uptime) {
+    uptime.style.display = 'none';
+    if (el._uptimeInterval) {
+      clearInterval(el._uptimeInterval);
+      el._uptimeInterval = null;
+    }
+  }
+  const startNow = el.querySelector('.cookbook-task-start-now');
+  if (startNow) startNow.style.display = 'none';
+  const check = el.querySelector('.cookbook-task-check');
+  if (check) check.style.display = '';
+}
+
+async function _executeTaskStop(el, task) {
+  if (el._abort) el._abort.abort();
+  const badge = el.querySelector('.cookbook-task-status');
+  if (badge) {
+    badge.textContent = 'stopping...';
+    badge.className = 'cookbook-task-status cookbook-task-stopping';
+  }
+  _updateTask(task.sessionId, { _userStopped: true });
+  const outputText = el.querySelector('.cookbook-output-pre')?.textContent || task.output || '';
+  if (task.type === 'serve' && task.payload) {
+    _removeEndpointByUrl(_endpointUrlForTask(task, outputText));
+  }
+  const ollamaUnload = _ollamaUnloadCommand(task, outputText);
+  if (ollamaUnload) {
+    try {
+      await fetch('/api/shell/exec', {
+        method: 'POST', credentials: 'same-origin',
+        headers: { 'Content-Type': 'application/json' },
+        body: JSON.stringify({ command: ollamaUnload }),
+      });
+    } catch {}
+  }
+  try {
+    const result = await _stopCookbookSession(task);
+    return !!(result && result.ok);
+  } catch {
+    return false;
+  }
+}
+
+async function _onTaskStop(el, task, { removeAfter = false } = {}) {
+  const priorStatus = task.status;
+  const wasActiveServe = task.type === 'serve' && _RECONNECT_STATUSES.includes(priorStatus);
+  const wasActiveDownload = task.type === 'download' && priorStatus === 'running';
+  const badge = el.querySelector('.cookbook-task-status');
+  const stopOk = await _executeTaskStop(el, task);
+  if (!stopOk) {
+    try { uiModule.showToast('Stop failed — download may still be running in the background', 'error'); } catch (_) {}
+    if (wasActiveServe || wasActiveDownload) {
+      if (badge) _applyTaskBadge(badge, task);
+      el.dataset.status = priorStatus;
+      _updateTask(task.sessionId, { _userStopped: false, status: priorStatus });
+      _reconnectTask(el, task);
+    }
+    return false;
+  }
+  if (removeAfter) {
+    _animateOutThenRemove(el, task.sessionId);
+  } else {
+    _applyStoppedTaskCard(el, task);
+  }
+  return true;
+}
+
+async function _stopCookbookSession(task) {
+  const taskType = task?.type || '';
+  const body = {
+    session_id: task.sessionId,
+    remote_host: task.remoteHost || '',
+    ssh_port: _getPort(task) || '',
+    platform: task.platform || _getPlatform(task) || '',
+    task_type: taskType,
+  };
+  if (taskType === 'download') {
+    const repoId = task?.payload?.repo_id || task?.payload?.repoId || '';
+    if (repoId) body.repo_id = repoId;
+  }
+  try {
+    const r = await fetch('/api/cookbook/stop-session', {
+      method: 'POST',
+      credentials: 'same-origin',
+      headers: { 'Content-Type': 'application/json' },
+      body: JSON.stringify(body),
+    });
+    if (!r.ok) return { ok: false };
+    return await r.json();
+  } catch {
+    return { ok: false };
+  }
+}
+
 // ── tmux / Windows session commands ──
 
 function _taskRemoteHost(task) {
@@ -1007,7 +1128,9 @@ function _remoteTmuxPrefix() {
 }
 
 export function _tmuxCmd(task, tmuxArgs) {
-  if (_isWindows(task)) {
+  const localWin = !_taskRemoteHost(task)
+    && (_isWindows(task) || _isWindows('local'));
+  if (_isWindows(task) || localWin) {
     return _winSessionCmd(task, tmuxArgs);
   }
   const host = _taskRemoteHost(task);
@@ -1061,8 +1184,8 @@ function _winSessionStopTreePs(task) {
   const sid = task.sessionId;
   const stopTree = `function Stop-Tree([int]$Id) { Get-CimInstance Win32_Process -Filter ('ParentProcessId = ' + $Id) -ErrorAction SilentlyContinue | ForEach-Object { Stop-Tree ([int]$_.ProcessId) }; Stop-Process -Id $Id -Force -ErrorAction SilentlyContinue }`;
   return host
-    ? `${stopTree}; $p = Get-Content '${sd}\\${sid}.pid' -ErrorAction SilentlyContinue; if ($p -match '^\\d+$') { Stop-Tree ([int]$p) }; Remove-Item '${sd}\\${sid}.*' -Force -ErrorAction SilentlyContinue`
-    : `${stopTree}; $p = Get-Content (Join-Path $env:TEMP 'odysseus-tmux\\${sid}.pid') -ErrorAction SilentlyContinue; if ($p -match '^\\d+$') { Stop-Tree ([int]$p) }; Remove-Item (Join-Path $env:TEMP 'odysseus-tmux\\${sid}.*') -Force -ErrorAction SilentlyContinue`;
+    ? `${stopTree}; $p = Get-Content '${sd}\\${sid}.pid' -ErrorAction SilentlyContinue; if ($p -match '^\\d+$') { Stop-Tree ([int]$p) }`
+    : `${stopTree}; $p = Get-Content (Join-Path $env:TEMP 'odysseus-tmux\\${sid}.pid') -ErrorAction SilentlyContinue; if ($p -match '^\\d+$') { Stop-Tree ([int]$p) }`;
 }
 
 export function _tmuxGracefulKill(task) {
@@ -2866,53 +2989,32 @@ export function _renderRunningTab() {
       _reconnectTask(el, task);
     });
 
-    // Wire stop
+    // Wire stop — server-side stop-session; rollback on failure for live tasks.
     el.querySelector('.cookbook-task-action-stop').addEventListener('click', async () => {
-      // Abort the reconnect loop before sending kill so that a DOWNLOAD_FAILED
-      // marker written by the shell wrapper (on SIGINT/non-zero exit) cannot
-      // trigger an auto-retry after a manual stop.
-      if (el._abort) el._abort.abort();
-      const badge = el.querySelector('.cookbook-task-status');
-      if (badge) { badge.textContent = 'stopping...'; badge.className = 'cookbook-task-status cookbook-task-stopping'; }
-      el.dataset.status = 'stopped';
-      _updateTask(task.sessionId, { _userStopped: true });
-      const outputText = el.querySelector('.cookbook-output-pre')?.textContent || task.output || '';
-      // Drop the model endpoint so the picker stops listing it.
-      if (task.type === 'serve' && task.payload) {
-        _removeEndpointByUrl(_endpointUrlForTask(task, outputText));
-      }
-      const ollamaUnload = _ollamaUnloadCommand(task, outputText);
-      if (ollamaUnload) {
-        try {
-          await fetch('/api/shell/exec', {
-            method: 'POST', credentials: 'same-origin',
-            headers: { 'Content-Type': 'application/json' },
-            body: JSON.stringify({ command: ollamaUnload }),
-          });
-        } catch {}
-      }
-      // Gracefully stop (C-c, then kill the session) so it's fully down...
-      try {
-        await fetch('/api/shell/exec', {
-          method: 'POST', credentials: 'same-origin',
-          headers: { 'Content-Type': 'application/json' },
-          body: JSON.stringify({ command: _tmuxGracefulKill(task) }),
-        });
-      } catch {}
-      // ...then smoothly fade/slide the card out and auto-remove it — no manual
-      // ⋮ → Remove needed.
-      _animateOutThenRemove(el, task.sessionId);
+      const liveTask = _loadTasks().find(t => t.sessionId === task.sessionId) || task;
+      await _onTaskStop(el, liveTask, { removeAfter: false });
     });
 
-    // Wire kill — awaits the SSH/tmux kill and verifies the session is
-    // actually gone before removing the row. Previously fire-and-forget,
-    // which meant a failed kill (wrong remoteHost, SSH error, tmux server
-    // already exited) silently left the live serve running while the
-    // row disappeared from the UI.
+    // Wire kill — running tasks: stop then dismiss; finished/stopped: remove only.
     el.querySelector('.cookbook-task-action-kill').addEventListener('click', async () => {
-      const outputText = el.querySelector('.cookbook-output-pre')?.textContent || task.output || '';
-      const isLive = task.type === 'serve' && ['running', 'ready', 'loading', 'warming', 'starting'].includes(task.status || '');
-      const ollamaUnload = _ollamaUnloadCommand(task, outputText);
+      const liveTask = _loadTasks().find(t => t.sessionId === task.sessionId) || task;
+      const liveStatus = liveTask.status || el.dataset.status || task.status;
+      const _isLiveServe = liveTask.type === 'serve' && ['running', 'ready', 'loading', 'warming', 'starting'].includes(liveStatus || '');
+      const _isActive = _isLiveServe || (liveStatus === 'running' && liveTask.type === 'download');
+      if (_isActive) {
+        await _onTaskStop(el, liveTask, { removeAfter: true });
+        return;
+      }
+      // Inactive download rows are UI history only — the session is already dead.
+      // stop-session with repo_id would mark the model user-stopped and kill any
+      // newer Restart of the same repo (duplicate-card cleanup scenario).
+      if (liveTask.type === 'download') {
+        _animateOutThenRemove(el, liveTask.sessionId);
+        return;
+      }
+      const outputText = el.querySelector('.cookbook-output-pre')?.textContent || liveTask.output || '';
+      const isLive = liveTask.type === 'serve' && ['running', 'ready', 'loading', 'warming', 'starting'].includes(liveStatus || '');
+      const ollamaUnload = _ollamaUnloadCommand(liveTask, outputText);
       if (ollamaUnload) {
         try {
           await fetch('/api/shell/exec', {
@@ -2924,41 +3026,30 @@ export function _renderRunningTab() {
       }
       let killOk = true;
       try {
-        const r = await fetch('/api/shell/exec', {
-          method: 'POST', credentials: 'same-origin',
-          headers: { 'Content-Type': 'application/json' },
-          body: JSON.stringify({ command: _tmuxGracefulKill(task) }),
-        });
-        if (r.ok) {
-          const out = await r.json();
-          // Don't trust exit_code alone — tmux kill returns 0 even when
-          // there was nothing to kill. Verify the session is actually gone.
-          if (task.sessionId && isLive) {
-            try {
-              const probe = await fetch('/api/shell/exec', {
-                method: 'POST', credentials: 'same-origin',
-                headers: { 'Content-Type': 'application/json' },
-                body: JSON.stringify({ command: _tmuxCmd(task, `has-session -t ${task.sessionId}`) }),
-              });
-              if (probe.ok) {
-                const pj = await probe.json();
-                // has-session exits 0 when session STILL exists; non-zero = gone.
-                if ((pj.exit_code || 0) === 0) killOk = false;
-              }
-            } catch (_) { /* probe best-effort; trust kill */ }
-          }
-        } else {
-          killOk = false;
+        const result = await _stopCookbookSession(liveTask);
+        killOk = !!(result && result.ok);
+        if (killOk && liveTask.sessionId && isLive && !_isWindows(liveTask)) {
+          try {
+            const probe = await fetch('/api/shell/exec', {
+              method: 'POST', credentials: 'same-origin',
+              headers: { 'Content-Type': 'application/json' },
+              body: JSON.stringify({ command: _tmuxCmd(liveTask, `has-session -t ${liveTask.sessionId}`) }),
+            });
+            if (probe.ok) {
+              const pj = await probe.json();
+              if ((pj.exit_code || 0) === 0) killOk = false;
+            }
+          } catch (_) { /* probe best-effort; trust stop-session */ }
         }
       } catch (_) { killOk = false; }
       if (!killOk) {
         try { uiModule.showToast('Kill failed — session may still be running. Check `tmux ls` on the server.', 'error'); } catch (_) {}
-        return;  // leave the row so the user can retry
+        return;
       }
-      if (task.type === 'serve' && task.payload) {
-        const endpointUrl = _endpointUrlForTask(task, outputText);
+      if (liveTask.type === 'serve' && liveTask.payload) {
+        const endpointUrl = _endpointUrlForTask(liveTask, outputText);
         _removeEndpointByUrl(endpointUrl);
-        const modelName = task.payload.model || task.name || '';
+        const modelName = liveTask.payload.model || liveTask.name || '';
         if (modelName) {
           fetch('/api/model-endpoints', { credentials: 'same-origin' })
             .then(r => r.json())
@@ -2968,7 +3059,7 @@ export function _renderRunningTab() {
             }).catch(() => {});
         }
       }
-      _animateOutThenRemove(el, task.sessionId);
+      _animateOutThenRemove(el, liveTask.sessionId);
     });
 
     // Wire retry
@@ -3481,6 +3572,13 @@ async function _reconnectTask(el, task) {
               badge.textContent = 'finishing';
               badge.className = 'cookbook-task-status cookbook-task-running';
             }
+            if (snapshot.includes('DOWNLOAD_STOPPED')) {
+              badge.textContent = _statusLabel('stopped', task.type);
+              badge.className = 'cookbook-task-status cookbook-task-stopped';
+              _updateTask(task.sessionId, { status: 'stopped', _userStopped: true });
+              el.dataset.status = 'stopped';
+              break;
+            }
             if (snapshot.includes('DOWNLOAD_FAILED')) {
               // The wrapper prints DOWNLOAD_FAILED but exits 0, and per-file
               // "Download complete"/"100%" lines make it look successful — so
@@ -3491,7 +3589,7 @@ async function _reconnectTask(el, task) {
               const _accessDenied = /Access to model.*is restricted|gated repo|GatedRepoError|401 Unauthorized|403 Forbidden|not in the authorized list|awaiting a review|must (?:be authenticated|have access)/i.test(snapshot);
               const _dlKey = task.payload?.repo_id || task.name;
               const _dlN = _dlRetryCount.get(_dlKey) || 0;
-              if (!controller.signal.aborted && !_accessDenied && task.type === 'download' && task.payload && _dlN < _DL_MAX_AUTO_RETRY) {
+              if (!controller.signal.aborted && !task._userStopped && !_accessDenied && task.type === 'download' && task.payload && _dlN < _DL_MAX_AUTO_RETRY) {
                 // Auto-retry: kill the dead session and re-launch (resumes from
                 // the cached .incomplete files) after a short delay.
                 _dlRetryCount.set(_dlKey, _dlN + 1);

tests/test_cookbook_download_restart_js.py

@@ -0,0 +1,52 @@
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+RUNNING_JS = ROOT / "static" / "js" / "cookbookRunning.js"
+
+
+def test_inactive_download_remove_skips_stop_session():
+    source = RUNNING_JS.read_text(encoding="utf-8")
+    idx = source.index("Inactive download rows are UI history only")
+    block = source[idx:idx + 500]
+    assert "liveTask.type === 'download'" in block
+    assert "_animateOutThenRemove(el, liveTask.sessionId)" in block
+    assert "_stopCookbookSession(task)" not in block
+
+
+def test_menu_kill_handler_uses_refreshed_task_not_stale_closure():
+    source = RUNNING_JS.read_text(encoding="utf-8")
+    idx = source.index(".cookbook-task-action-kill').addEventListener('click'")
+    block = source[idx:idx + 900]
+    assert "const liveTask = _loadTasks().find(t => t.sessionId === task.sessionId)" in block
+    assert "const liveStatus = liveTask.status || el.dataset.status" in block
+    assert "await _onTaskStop(el, liveTask" in block
+
+
+def test_download_stopped_is_terminal_for_active_output():
+    source = RUNNING_JS.read_text(encoding="utf-8")
+    idx = source.index("function _downloadOutputLooksActive")
+    block = source[idx:idx + 450]
+    assert "DOWNLOAD_STOPPED" in block
+
+
+def test_failed_stop_rolls_back_live_serve_and_download_statuses():
+    source = RUNNING_JS.read_text(encoding="utf-8")
+    idx = source.index("async function _onTaskStop")
+    block = source[idx:idx + 1000]
+    assert "_RECONNECT_STATUSES.includes(priorStatus)" in block
+    assert "task.type === 'download' && priorStatus === 'running'" in block
+    failed = block[block.index("if (!stopOk)"):block.index("if (removeAfter)")]
+    assert "el.dataset.status = priorStatus" in block
+    assert "_userStopped: false, status: priorStatus" in failed
+    assert "_applyStoppedTaskCard" not in failed
+
+
+def test_stop_cookbook_session_sends_repo_id_only_for_downloads():
+    source = RUNNING_JS.read_text(encoding="utf-8")
+    idx = source.index("async function _stopCookbookSession")
+    block = source[idx:idx + 900]
+    assert "task_type: taskType" in block
+    assert "if (taskType === 'download')" in block
+    assert "body.repo_id = repoId" in block
+    assert block.index("if (taskType === 'download')") < block.index("body.repo_id = repoId")

tests/test_cookbook_helpers.py

@@ -724,9 +724,13 @@ def test_local_windows_download_pid_tracks_inner_bash_and_stop_kills_tree():
     running_src = (Path(__file__).resolve().parents[1] / "static" / "js" / "cookbookRunning.js").read_text(encoding="utf-8")
 
     assert 'printf \'%s\\\\n\' \\"$$\\" > {pp}' in routes_src
+    assert "/api/cookbook/stop-session" in routes_src
+    assert "_scan_windows_session_pids" in routes_src
+    assert "_cmdline_references_hf_repo" in routes_src
+    assert "grep -Fxq" in routes_src
     assert "function Stop-Tree([int]$Id)" in running_src
-    assert "('ParentProcessId = ' + $Id)" in running_src
-    assert "Stop-Tree ([int]$p)" in running_src
+    assert "_stopCookbookSession" in running_src
+    assert "_winSessionStopTreePs" in running_src
 
 
 def test_llama_cpp_rebuild_cmd_runs_clean_on_a_fresh_home(tmp_path):

tests/test_cookbook_local_windows_stop.py

@@ -0,0 +1,217 @@
+"""Local-Windows cookbook stop-session / orphan guard tests (PR1 scope)."""
+from __future__ import annotations
+
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+from starlette.requests import Request
+
+import routes.cookbook_routes as cookbook_routes
+from routes.cookbook_routes import _cmdline_references_hf_repo, _coerce_ssh_port
+
+
+def _route_endpoint(path: str, method: str):
+    router = cookbook_routes.setup_cookbook_routes()
+    return next(
+        route.endpoint
+        for route in router.routes
+        if route.path == path and method in route.methods
+    )
+
+
+def _admin_request(path: str) -> Request:
+    return Request(
+        {
+            "type": "http",
+            "method": "POST",
+            "path": path,
+            "headers": [],
+            "state": {},
+        }
+    )
+
+
+def test_cmdline_references_hf_repo_exact_match_not_prefix():
+    short = "org/model"
+    long_repo = "org/model-large"
+    assert _cmdline_references_hf_repo(f"python hf_download.py {short}", short)
+    assert _cmdline_references_hf_repo(f"hf download {short} --local-dir /tmp", short)
+    assert _cmdline_references_hf_repo(
+        f"snapshot_download('{short}', local_dir='/tmp')", short
+    )
+    assert not _cmdline_references_hf_repo(f"python hf_download.py {long_repo}", short)
+    assert not _cmdline_references_hf_repo(f"hf download {long_repo}", short)
+
+
+def test_coerce_ssh_port_rejects_out_of_range():
+    assert _coerce_ssh_port("2222") == "2222"
+    assert _coerce_ssh_port("0") is None
+    assert _coerce_ssh_port("65536") is None
+
+
+def test_local_windows_stop_session_wiring_source():
+    routes_src = Path(cookbook_routes.__file__).read_text(encoding="utf-8")
+    running_src = (Path(__file__).resolve().parents[1] / "static" / "js" / "cookbookRunning.js").read_text(
+        encoding="utf-8"
+    )
+    assert "/api/cookbook/stop-session" in routes_src
+    assert "_scan_windows_session_pids" in routes_src
+    assert "_cmdline_references_hf_repo" in routes_src
+    assert "grep -Fxq" in routes_src
+    assert "_stopCookbookSession" in running_src
+    assert "async function _onTaskStop" in running_src
+    assert "_userStopped: false, status: priorStatus" in running_src
+
+
+@pytest.mark.asyncio
+async def test_stop_session_dependency_pip_label_still_kills(monkeypatch):
+    import asyncio
+
+    endpoint = _route_endpoint("/api/cookbook/stop-session", "POST")
+    kill_ran: list[bool] = []
+
+    class FakeProc:
+        returncode = 0
+
+        async def communicate(self):
+            return b"", b""
+
+    async def fake_subprocess_shell(*args, **kwargs):
+        kill_ran.append(True)
+        return FakeProc()
+
+    monkeypatch.setattr(cookbook_routes, "IS_WINDOWS", False)
+    monkeypatch.setattr(asyncio, "create_subprocess_shell", fake_subprocess_shell)
+    monkeypatch.setattr(cookbook_routes, "require_admin", lambda request: None)
+
+    req = SimpleNamespace(
+        session_id="cookbook-deadbeef",
+        remote_host=None,
+        ssh_port=None,
+        platform=None,
+        repo_id="llama-cpp-python[server]",
+        task_type="dependency",
+    )
+    result = await endpoint(_admin_request("/api/cookbook/stop-session"), req)
+    assert result["ok"] is True
+    assert kill_ran
+
+
+@pytest.mark.asyncio
+async def test_stop_session_serve_skips_download_repo_side_effects(monkeypatch):
+    monkeypatch.setattr(cookbook_routes, "require_admin", lambda request: None)
+    endpoint = _route_endpoint("/api/cookbook/stop-session", "POST")
+    seen: list = []
+
+    async def fake_impl(session_id, remote_host="", ssh_port=None, platform="", repo_id=None):
+        seen.append({"repo_id": repo_id, "platform": platform})
+        return {"ok": True}
+
+    for name, cell in zip(endpoint.__code__.co_freevars, endpoint.__closure__ or ()):
+        if name == "_stop_cookbook_session_impl":
+            cell.cell_contents = fake_impl
+            break
+
+    request = _admin_request("/api/cookbook/stop-session")
+    req_serve = SimpleNamespace(
+        session_id="serve-deadbeef",
+        remote_host=None,
+        ssh_port=None,
+        platform=None,
+        repo_id="org/model",
+        task_type="serve",
+    )
+    result = await endpoint(request, req_serve)
+    assert result["ok"] is True
+    assert seen and seen[0]["repo_id"] is None
+
+    req_dl = SimpleNamespace(
+        session_id="cookbook-deadbeef",
+        remote_host=None,
+        ssh_port=None,
+        platform=None,
+        repo_id="org/model",
+        task_type="download",
+    )
+    await endpoint(request, req_dl)
+    assert seen[-1]["repo_id"] == "org/model"
+
+
+def test_local_download_guard_present_without_remote_windows():
+    src = Path(cookbook_routes.__file__).read_text(encoding="utf-8")
+    assert "_find_live_local_download" in src
+    assert "_clear_download_stopped" in src
+    assert "_find_live_remote_windows_download" not in src
+
+
+@pytest.mark.asyncio
+async def test_local_windows_stop_failure_preserves_recovery_state(monkeypatch, tmp_path):
+    monkeypatch.setattr(cookbook_routes, "TMUX_LOG_DIR", tmp_path)
+    monkeypatch.setattr(
+        cookbook_routes, "COOKBOOK_STATE_FILE", str(tmp_path / "state.json")
+    )
+    monkeypatch.setattr(cookbook_routes, "require_admin", lambda request: None)
+    monkeypatch.setattr(cookbook_routes, "pid_alive", lambda pid: True)
+    runner = tmp_path / "cookbook-deadbeef_run.ps1"
+    pid_file = tmp_path / "cookbook-deadbeef.pid"
+    runner.write_text("hf download org/model", encoding="utf-8")
+    pid_file.write_text("1234", encoding="utf-8")
+
+    def failed_taskkill(args, **kwargs):
+        assert args[0] == "taskkill"
+        return SimpleNamespace(returncode=1, stdout="")
+
+    monkeypatch.setattr(cookbook_routes.subprocess, "run", failed_taskkill)
+    endpoint = _route_endpoint("/api/cookbook/stop-session", "POST")
+    req = SimpleNamespace(
+        session_id="cookbook-deadbeef",
+        remote_host=None,
+        ssh_port=None,
+        platform="windows",
+        repo_id="org/model",
+        task_type="download",
+    )
+
+    result = await endpoint(_admin_request("/api/cookbook/stop-session"), req)
+
+    assert result["ok"] is False
+    assert result["stopped"] is False
+    assert "failed to stop local Windows process tree" in result["error"]
+    assert runner.exists()
+    assert pid_file.exists()
+    assert not (tmp_path / "cookbook-deadbeef.stop").exists()
+    assert not (tmp_path / "cookbook-stopped-repos.json").exists()
+
+
+@pytest.mark.asyncio
+async def test_local_windows_already_gone_stop_is_success(monkeypatch, tmp_path):
+    monkeypatch.setattr(cookbook_routes, "TMUX_LOG_DIR", tmp_path)
+    monkeypatch.setattr(
+        cookbook_routes, "COOKBOOK_STATE_FILE", str(tmp_path / "state.json")
+    )
+    monkeypatch.setattr(cookbook_routes, "require_admin", lambda request: None)
+    monkeypatch.setattr(cookbook_routes, "pid_alive", lambda pid: False)
+    runner = tmp_path / "cookbook-deadbeef_run.sh"
+    runner.write_text("hf download org/model", encoding="utf-8")
+
+    def empty_scan(args, **kwargs):
+        return SimpleNamespace(returncode=0, stdout="")
+
+    monkeypatch.setattr(cookbook_routes.subprocess, "run", empty_scan)
+    endpoint = _route_endpoint("/api/cookbook/stop-session", "POST")
+    req = SimpleNamespace(
+        session_id="cookbook-deadbeef",
+        remote_host=None,
+        ssh_port=None,
+        platform="windows",
+        repo_id="org/model",
+        task_type="download",
+    )
+
+    result = await endpoint(_admin_request("/api/cookbook/stop-session"), req)
+
+    assert result["ok"] is True
+    assert result["stopped"] is False
+    assert not runner.exists()
+    assert (tmp_path / "cookbook-deadbeef.stop").exists()