From 4827a3c8bbe08dc8306c11027f673e9256623b25 Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 16:05:59 -0400 Subject: [PATCH 1/8] chore(tasks): plan Windows auto-update lifecycle redesign --- tasks/todo.md | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/tasks/todo.md b/tasks/todo.md index 876d13d6..24508b54 100644 --- a/tasks/todo.md +++ b/tasks/todo.md @@ -348,3 +348,27 @@ Done: `mythos-enhancements` PR #118 merged (d51ef649). Completed-work logs through 2026-06-11 (agent robustness arc, statusline v2, review-safety hardening, telemetry sync, CodeRabbit triage) were trimmed on repush of PR #118 — see git history of this file for the full record. + +### Windows auto-update lifecycle redesign (2026-07-17) + +Branch: `fix/windows-auto-update-lifecycle` (off origin/main @ c5c0bc92) + +Root cause (3-lane scout, confirmed): silent startup auto-update on Windows runs the Inno +installer inline mid-session (`ui/shell/update.py:1239` -> `_run_native_installer` -> +`/SILENT /CLOSEAPPLICATIONS` + `sys.exit(0)`); shell swallows SystemExit, installer's 15s +WaitForLauncherExit times out, Restart Manager force-closes the session. No Windows +stage-and-promote path (macOS/Linux have one via atexit). + +Scope (user): FULL redesign. Producer: Codex (GPT-5.6 Sol) via claude-architect MCP pipeline. + +- [ ] Delegation A (Python core): phased update engine (check -> download+verify+stage with + atomic manifest -> apply at safe boundary only); typed intent replaces check_only bool; + config enum policy off|notify|download|apply_on_exit (default download, legacy bool + migration); background task never installs/exits mid-session; restart-to-apply notice; + pre-start apply hook in app.py before PythinkerCLI.create(); tests + changelog. + -> verify: pipeline ruff/pyright/pytest + post-integration full package gate. +- [ ] Delegation B (after A): packages/windows-installer/installer.iss tuning + multi-session + concurrency guard (Session.acquire_ownership is fcntl no-op on Windows, session.py:53). +- [ ] Full gate + tests_e2e snapshots, then PR. + +Out of scope (log): broader Windows session locking beyond updater guard; install.ps1 mirrors. From 8659d23d79fe0729c8cf4ece0c3d76d9b04ba4cb Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 18:34:21 -0400 Subject: [PATCH 2/8] feat(update): stage auto-updates instead of installing mid-session On Windows the silent startup auto-updater launched the Inno Setup installer inline with /CLOSEAPPLICATIONS, letting its Restart Manager force-close the running pythinker.exe and kill the active session. Redesign the update lifecycle around a typed UpdateIntent (CHECK / STAGE_FOR_RESTART / INSTALL / INSTALL_AND_EXIT) so background callers are type-unable to request an in-session install: - Background startup updates download, sha256-verify, and stage the Windows installer with an atomically written manifest; they can no longer spawn installers, run package-manager upgrades, or raise SystemExit (contained in _run_silent_update_job; the done-callback stays as defense in depth). - A pre-session bootstrap in the CLI entry applies a verified staged update before any config/session/runtime construction and fails closed (discard + continue) on any invalid or stale stage; apply re-verifies the digest and version and guards against a concurrently superseded manifest. - In-shell /update stages on Windows (restart-to-apply notice); the standalone `pythinker update` CLI keeps its install-and-exit behavior as an explicit foreground operation. - config auto_update becomes a policy enum off|notify|download| apply_on_exit (default download) with legacy bool compatibility (true->download, false->notify) including PYTHINKER_AUTO_UPDATE; PYTHINKER_CLI_NO_AUTO_UPDATE stays the highest-precedence kill switch. /update auto, the settings panel, and pythinker info are mode-aware (info JSON auto_update_config is now a string). - The post-install smoke check is skipped for the Windows staged path: it would run the old executable and falsely certify the stage, which is instead digest-verified at staging and again at apply. --- CHANGELOG.md | 10 + src/pythinker_code/cli/__init__.py | 11 + src/pythinker_code/cli/info.py | 10 +- src/pythinker_code/cli/update.py | 7 +- src/pythinker_code/config.py | 61 +++- src/pythinker_code/ui/shell/__init__.py | 68 +++- .../ui/shell/selectors/settings.py | 17 +- src/pythinker_code/ui/shell/slash.py | 112 +++--- src/pythinker_code/ui/shell/update.py | 295 ++++++++++++++-- .../ui/shell/update_orchestrator.py | 66 +++- src/pythinker_code/update_policy.py | 44 ++- tasks/todo.md | 31 +- tests/cli/test_info.py | 8 +- tests/core/test_config.py | 35 +- tests/ui/test_update_staging.py | 320 ++++++++++++++++++ .../ui_and_conv/test_native_update_parity.py | 12 +- tests/ui_and_conv/test_settings_selector.py | 22 +- tests/ui_and_conv/test_shell_update.py | 91 +++-- tests/ui_and_conv/test_silent_auto_update.py | 66 +++- tests/ui_and_conv/test_update_auto_slash.py | 68 ++-- tests/ui_and_conv/test_update_orchestrator.py | 54 ++- 21 files changed, 1155 insertions(+), 253 deletions(-) create mode 100644 tests/ui/test_update_staging.py diff --git a/CHANGELOG.md b/CHANGELOG.md index dd1ca4a1..1eb8ed98 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,16 @@ GitHub Releases page; `0.8.0` is the new starting line. ## Unreleased +- **Updates never interrupt a running session.** On Windows, the background + auto-updater previously launched the installer mid-session, force-closing the + active Pythinker session. Updates are now downloaded and staged with a verified + manifest, surfaced as a "restart to apply" notice, and applied before the next + session starts (or at clean exit with the new `apply_on_exit` policy). The + `auto_update` config becomes a policy enum — `off`, `notify`, `download` + (default), `apply_on_exit` — with legacy booleans still accepted + (`true` → `download`, `false` → `notify`); `pythinker info` now reports the + mode string, and `/update auto` accepts the new mode names. + ## 0.59.0 (2026-07-17) - **Reviewer subagents now receive deterministic Git scopes.** Structured automatic, diff --git a/src/pythinker_code/cli/__init__.py b/src/pythinker_code/cli/__init__.py index 66838a41..97bd875a 100644 --- a/src/pythinker_code/cli/__init__.py +++ b/src/pythinker_code/cli/__init__.py @@ -859,6 +859,17 @@ def _emit_fatal_error(message: str) -> None: param_hint="--session", ) + # Apply a previously staged Windows update before any session, runtime, or + # agent is constructed. Interactive shell launches only: print/ACP/wire + # callers are scripted flows where exiting to run an installer would break + # the invoker. Fail closed inside: an invalid stage is discarded and normal + # startup continues. + if ui == "shell" and prompt is None: + from pythinker_code.ui.shell.update import apply_staged_update_before_start + + if apply_staged_update_before_start(): + raise typer.Exit(0) + config: Config | Path | None = None if config_string is not None: config_string = config_string.strip() diff --git a/src/pythinker_code/cli/info.py b/src/pythinker_code/cli/info.py index 58317af2..ae4885fc 100644 --- a/src/pythinker_code/cli/info.py +++ b/src/pythinker_code/cli/info.py @@ -14,12 +14,12 @@ class InfoData(TypedDict): wire_protocol_version: str python_version: str auto_update: bool | None - auto_update_config: bool | None + auto_update_config: str | None auto_update_override: str | None -def _auto_update_info() -> tuple[bool | None, bool | None, str | None]: - """Return ``(effective_enabled, config_value, override_reason)``. +def _auto_update_info() -> tuple[bool | None, str | None, str | None]: + """Return ``(effective_enabled, config_mode, override_reason)``. Every element is ``None`` when the status cannot be resolved. The whole block is guarded so an unreadable config or any other failure never turns @@ -38,7 +38,7 @@ def _auto_update_info() -> tuple[bool | None, bool | None, str | None]: # has no config file yet rather than creating one as a side effect. config_exists = get_config_file(create=False).expanduser().exists() config = load_config() if config_exists else Config() - return auto_update_enabled(config), config.auto_update, override + return auto_update_enabled(config), config.auto_update.value, override except (OSError, ValueError, ImportError) as exc: # Read-only diagnostic: never abort `info`, but log the degraded path # instead of silently masking a real config/policy failure. ConfigError @@ -73,7 +73,7 @@ def _auto_update_line(info: InfoData) -> str: if effective is None: return "auto-update: unknown" state = "enabled" if effective else "disabled" - detail = f"config auto_update={'true' if info['auto_update_config'] else 'false'}" + detail = f"config auto_update={info['auto_update_config'] or 'unknown'}" override = info["auto_update_override"] if override: detail += f"; {override}" diff --git a/src/pythinker_code/cli/update.py b/src/pythinker_code/cli/update.py index 366ec36b..45a41093 100644 --- a/src/pythinker_code/cli/update.py +++ b/src/pythinker_code/cli/update.py @@ -23,10 +23,13 @@ def update( if ctx.invoked_subcommand is not None: return - from pythinker_code.ui.shell.update import UpdateResult + from pythinker_code.ui.shell.update import UpdateIntent, UpdateResult from pythinker_code.ui.shell.update_orchestrator import run_update_job - result = asyncio.run(run_update_job(print_output=True, check_only=check_only, source="cli")) + # The standalone CLI is its own foreground process: exiting to hand off to + # the platform installer is expected, unlike in-shell updates which stage. + intent = UpdateIntent.CHECK if check_only else UpdateIntent.INSTALL_AND_EXIT + result = asyncio.run(run_update_job(print_output=True, intent=intent, source="cli")) if result in (UpdateResult.FAILED, UpdateResult.UNSUPPORTED): raise typer.Exit(1) diff --git a/src/pythinker_code/config.py b/src/pythinker_code/config.py index dd52c126..27825334 100644 --- a/src/pythinker_code/config.py +++ b/src/pythinker_code/config.py @@ -3,6 +3,7 @@ import contextlib import json import os +from enum import StrEnum from pathlib import Path from types import UnionType from typing import Any, Literal, Self, Union, cast, get_args, get_origin @@ -1115,6 +1116,47 @@ class PluginsConfig(BaseModel): ) +class AutoUpdateMode(StrEnum): + """Startup auto-update policy. + + ``OFF`` schedules nothing at startup; ``NOTIFY`` only refreshes the passive + update notice; ``DOWNLOAD`` downloads and stages the new release in the + background so a restart applies it; ``APPLY_ON_EXIT`` additionally launches + the staged installer when the process exits cleanly. No mode ever installs + or restarts while an interactive session is running. + """ + + OFF = "off" + NOTIFY = "notify" + DOWNLOAD = "download" + APPLY_ON_EXIT = "apply_on_exit" + + +# Legacy boolean spellings accepted for backward compatibility with the old +# `auto_update: bool` config field and PYTHINKER_AUTO_UPDATE env values. +_AUTO_UPDATE_LEGACY_TRUE = frozenset({"true", "1", "yes"}) +_AUTO_UPDATE_LEGACY_FALSE = frozenset({"false", "0", "no"}) + + +def coerce_auto_update_mode(value: object) -> object: + """Map legacy boolean auto_update values onto the policy enum. + + ``true`` keeps its old meaning of "update automatically in the background" + (now download-and-stage); ``false`` maps to ``notify`` because the old + disabled state still surfaced the passive update notice. + """ + if isinstance(value, bool): + return AutoUpdateMode.DOWNLOAD if value else AutoUpdateMode.NOTIFY + if isinstance(value, str): + normalized = value.strip().lower() + if normalized in _AUTO_UPDATE_LEGACY_TRUE: + return AutoUpdateMode.DOWNLOAD + if normalized in _AUTO_UPDATE_LEGACY_FALSE: + return AutoUpdateMode.NOTIFY + return normalized + return value + + class Config(BaseModel): """Main configuration structure.""" @@ -1223,10 +1265,23 @@ class Config(BaseModel): "Supported on macOS, Linux, and Windows. Default: false." ), ) - auto_update: bool = Field( - default=True, - description="Automatically install new releases in the background at startup.", + auto_update: AutoUpdateMode = Field( + default=AutoUpdateMode.DOWNLOAD, + description=( + "Startup auto-update policy: 'off' (no startup update task), 'notify' " + "(show update notices only), 'download' (download and stage new releases " + "in the background; a restart applies them), or 'apply_on_exit' (also " + "launch the staged installer after the session exits). Updates are never " + "applied while a session is running. Legacy booleans are accepted: " + "true → download, false → notify." + ), ) + + @field_validator("auto_update", mode="before") + @classmethod + def _coerce_auto_update(cls, value: object) -> object: + return coerce_auto_update_mode(value) + models: dict[str, LLMModel] = Field(default_factory=dict, description="List of LLM models") providers: dict[str, LLMProvider] = Field( default_factory=dict, description="List of LLM providers" diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index a346ce30..71524a74 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -71,6 +71,7 @@ from pythinker_code.ui.shell.slash import registry as shell_slash_registry from pythinker_code.ui.shell.update import ( MANAGED_CHANNEL_MARKER, + UpdateIntent, UpdateResult, _detect_upgrade_command, # pyright: ignore[reportPrivateUsage] _mark_auto_update_check_attempt, # pyright: ignore[reportPrivateUsage] @@ -78,7 +79,9 @@ consume_whats_new, format_managed_channel_notice, pending_update_notice, + read_windows_staged_update, refresh_update_cache_if_due, + register_windows_staged_apply_on_exit, welcome_update_target, ) from pythinker_code.ui.shell.update_orchestrator import ( @@ -94,7 +97,7 @@ from pythinker_code.ui.terminal_capabilities import ascii_glyphs_enabled, motion_disabled from pythinker_code.ui.theme import BRAND, BrandToken, tui_rich_style from pythinker_code.ui.theme import get_tui_tokens as _get_tui_tokens -from pythinker_code.update_policy import auto_update_enabled +from pythinker_code.update_policy import resolve_auto_update_mode from pythinker_code.utils.aioqueue import QueueShutDown from pythinker_code.utils.envvar import get_env_bool from pythinker_code.utils.logging import logger @@ -2145,7 +2148,12 @@ async def _auto_update(self) -> None: self._refresh_update_notice_line() async def _silent_auto_update(self) -> None: - """Install a newer release silently in the background at startup.""" + """Download and stage a newer release in the background at startup. + + This never installs mid-session: the Windows installer / native binary + is staged and applied at the next launch (or at clean exit under the + ``apply_on_exit`` policy), which the persistent restart notice reflects. + """ if not _should_auto_check_for_updates(): return @@ -2156,16 +2164,36 @@ async def _silent_auto_update(self) -> None: if result is not None and result is not UpdateResult.FAILED: _mark_auto_update_check_attempt() if result is UpdateResult.UPDATED: + self._maybe_arm_windows_apply_on_exit() self._surface_installed_update_notice() elif result is UpdateResult.UPDATE_AVAILABLE: self._surface_managed_channel_notice() # FAILED / UP_TO_DATE / UNSUPPORTED / None → silent (recorded in the job log). + def _maybe_arm_windows_apply_on_exit(self) -> None: + from pythinker_code.config import AutoUpdateMode + + if not isinstance(self.soul, PythinkerSoul): + return + mode = resolve_auto_update_mode(self.soul.runtime.config) + if mode is not AutoUpdateMode.APPLY_ON_EXIT: + return + if read_windows_staged_update() is None: + return + register_windows_staged_apply_on_exit() + async def _run_silent_update_job(self) -> UpdateResult | None: try: - return await run_update_job(print_output=False, check_only=False, source="startup-auto") + return await run_update_job( + print_output=False, intent=UpdateIntent.STAGE_FOR_RESTART, source="startup-auto" + ) except SystemExit: - raise + # STAGE_FOR_RESTART must never exit the process; reaching this means + # an install path leaked into the background task. Contain it — a + # propagated SystemExit would tear down the user's session (the + # exact mid-session kill this path is designed to prevent). + logger.error("Background update task attempted to exit the process; suppressed.") + return None except Exception: # Boundary-only recovery: update failure must not abort the shell, # and run_update_job has already persisted status/log details. @@ -2263,20 +2291,29 @@ def _schedule_startup_update_task(self) -> None: - env kill-switch set → nothing (cache filters already suppress the notice, matching today's hard-disable behavior). - - enabled → silent background install. - - config-disabled OR source checkout → refresh the persistent notice - only (`_auto_update`); self-suppresses for source checkouts because - `pending_update_notice()` returns None in that path. - - non-PythinkerSoul → same notice-refresh path (no runtime config to + - `off` (or source checkout) → nothing. + - `notify` → refresh the persistent notice only (`_auto_update`). + - `download` / `apply_on_exit` → background download-and-stage + (`_silent_auto_update`); never installs mid-session. + - non-PythinkerSoul → the notice-refresh path (no runtime config to consult), matching the prior unconditional `_auto_update` behavior. """ + from pythinker_code.config import AutoUpdateMode + if get_env_bool("PYTHINKER_CLI_NO_AUTO_UPDATE"): logger.info("Auto-update disabled by PYTHINKER_CLI_NO_AUTO_UPDATE environment variable") return - if isinstance(self.soul, PythinkerSoul) and auto_update_enabled(self.soul.runtime.config): - self._start_background_task(self._silent_auto_update()) - else: + if not isinstance(self.soul, PythinkerSoul): self._start_background_task(self._auto_update()) + return + mode = resolve_auto_update_mode(self.soul.runtime.config) + if mode is AutoUpdateMode.OFF: + logger.info("Startup update task disabled by auto_update policy 'off'") + return + if mode is AutoUpdateMode.NOTIFY: + self._start_background_task(self._auto_update()) + return + self._start_background_task(self._silent_auto_update()) def _start_background_task(self, coro: Coroutine[Any, Any, Any]) -> asyncio.Task[Any]: task = asyncio.create_task(coro) @@ -2289,9 +2326,10 @@ def _cleanup(t: asyncio.Task[Any]) -> None: except asyncio.CancelledError: pass except SystemExit: - # The silent updater's Windows native/pip path raises SystemExit - # so the installer can replace the binary; don't crash the shell. - logger.info("Background task requested process exit (update installer launched).") + # Defense in depth: no background task is allowed to request + # process exit (updates stage for restart instead). If one + # slips through, contain it here rather than killing the shell. + logger.error("Background task raised SystemExit; suppressed to keep the session.") except Exception: logger.exception("Background task failed:") diff --git a/src/pythinker_code/ui/shell/selectors/settings.py b/src/pythinker_code/ui/shell/selectors/settings.py index 69ea8691..82d21e44 100644 --- a/src/pythinker_code/ui/shell/selectors/settings.py +++ b/src/pythinker_code/ui/shell/selectors/settings.py @@ -9,7 +9,7 @@ from typing import Any, cast -from pythinker_code.config import Config +from pythinker_code.config import AutoUpdateMode, Config from pythinker_code.llm import derive_model_capabilities from pythinker_code.thinking import ( EXTENDED_THINKING_LEVELS, @@ -23,6 +23,7 @@ ) _BOOL_VALUES = ("true", "false") +_AUTO_UPDATE_VALUES = tuple(mode.value for mode in AutoUpdateMode) _NONE_MODEL_VALUE = "(none)" @@ -167,16 +168,16 @@ def _build_settings_config(config: Config) -> SettingsListConfig: id="auto_update", label="Auto-update", description=( - "Silently install new releases in the background at startup " - "(applied on next restart)." + "Startup update policy: off, notify, download (stage in background, " + "apply on restart), or apply_on_exit." if _auto_update_override is None else f"Auto-update is {_auto_update_override}; that override outranks this setting." ), # Show the *effective* state, and make the row read-only when an # override (env kill-switch / source checkout) forces it off, so the # panel never offers a no-op toggle. - current_value=(_bool(config.auto_update) if _auto_update_override is None else "false"), - values=_BOOL_VALUES if _auto_update_override is None else None, + current_value=(config.auto_update.value if _auto_update_override is None else "off"), + values=_AUTO_UPDATE_VALUES if _auto_update_override is None else None, ), SettingItem( id="merge_all_available_skills", @@ -364,9 +365,9 @@ def mark(setting_id: str) -> None: case "auto_update": # Only reached for the live (non-override) row; a read-only row # never submits a change. - new = value == "true" - if config.auto_update != new: - config.auto_update = new + new_mode = AutoUpdateMode(value) + if config.auto_update != new_mode: + config.auto_update = new_mode mark(setting_id) case "merge_all_available_skills": new = value == "true" diff --git a/src/pythinker_code/ui/shell/slash.py b/src/pythinker_code/ui/shell/slash.py index 97f6c948..1556660b 100644 --- a/src/pythinker_code/ui/shell/slash.py +++ b/src/pythinker_code/ui/shell/slash.py @@ -11,7 +11,7 @@ from pythinker_code.auth.platforms import get_platform_name_for_provider, refresh_managed_models from pythinker_code.cli import Reload, SwitchToDashboard, SwitchToWeb -from pythinker_code.config import StatusLineConfig, load_config, save_config +from pythinker_code.config import AutoUpdateMode, StatusLineConfig, load_config, save_config from pythinker_code.exception import ConfigError from pythinker_code.session import Session from pythinker_code.soul.pythinkersoul import PythinkerSoul @@ -2224,7 +2224,7 @@ async def show_memory(app: Shell, args: str): @registry.command(name="update", aliases=["upgrade"]) async def update_command(app: Shell, args: str): """Check for updates, or `auto [on|off]` to toggle silent startup auto-updates.""" - from pythinker_code.ui.shell.update import UpdateResult, run_update_prompt + from pythinker_code.ui.shell.update import UpdateIntent, UpdateResult, run_update_prompt from pythinker_code.ui.shell.update_orchestrator import run_update_job parts = args.strip().split() @@ -2243,13 +2243,14 @@ async def update_command(app: Shell, args: str): if action != "check": return - async def _runner(*, print_output: bool, check_only: bool) -> UpdateResult: - return await run_update_job( - print_output=print_output, check_only=check_only, source="slash" - ) + async def _runner(*, print_output: bool, intent: UpdateIntent) -> UpdateResult: + return await run_update_job(print_output=print_output, intent=intent, source="slash") result = await run_update_prompt(update_runner=_runner) if result is UpdateResult.UPDATED: + # A stage created via /update must honor the apply_on_exit policy just + # like the silent startup stage does. + app._maybe_arm_windows_apply_on_exit() # pyright: ignore[reportPrivateUsage] console.print("Updated — restart Pythinker to use the new version.") @@ -2263,12 +2264,12 @@ async def _prompt_update_action(app: Shell) -> str | None: """ from prompt_toolkit.shortcuts.choice_input import ChoiceInput - from pythinker_code.update_policy import auto_update_enabled + from pythinker_code.update_policy import resolve_auto_update_mode auto_label = "Auto-update on startup" if isinstance(app.soul, PythinkerSoul): - state = "on" if auto_update_enabled(app.soul.runtime.config) else "off" - auto_label = f"{auto_label}: {state}" + mode = resolve_auto_update_mode(app.soul.runtime.config) + auto_label = f"{auto_label}: {mode.value}" try: selection = await ChoiceInput( @@ -2286,15 +2287,20 @@ async def _prompt_update_action(app: Shell) -> str | None: async def _auto_update_toggle(app: Shell, args: list[str]) -> None: - """Show or set the silent startup auto-update preference. - - `/update auto on|off` sets it directly; `/update auto` with no value opens an - interactive On/Off picker (or, when an external override has made the setting - read-only, reports the effective state instead of popping a no-op picker). + """Show or set the startup auto-update policy. + + `/update auto ` sets it directly (modes: off, notify, download, + apply_on_exit). Legacy `on` maps to download; `off` selects the fully-off + policy — use `notify` for the old "no install, keep notices" behavior. + `/update auto` + with no value opens an interactive picker (or, when an external override + has made the setting read-only, reports the effective state instead of + popping a no-op picker). """ + from pythinker_code.config import coerce_auto_update_mode from pythinker_code.telemetry import track from pythinker_code.ui.theme import get_tui_tokens as _get_tok - from pythinker_code.update_policy import auto_update_enabled, auto_update_override_reason + from pythinker_code.update_policy import auto_update_override_reason, resolve_auto_update_mode _t = _get_tok() soul = ensure_pythinker_soul(app) @@ -2308,28 +2314,44 @@ def _print_override() -> None: console.print(f"[{_t.muted}]Note: {override}; this overrides the setting.[/]") if args: - value = args[0].lower() - if len(args) > 1 or value not in {"on", "off"}: - console.print(f"[{_t.warning}]Usage: /update auto [on|off][/]") + raw = args[0].lower() + usage = f"[{_t.warning}]Usage: /update auto [on|off|notify|download|apply_on_exit][/]" + if len(args) > 1: + console.print(usage) return - enabled = value == "on" + if raw == "on": + # `on` keeps its historical meaning: background auto-update, which + # now downloads and stages instead of installing mid-session. + selected_mode = AutoUpdateMode.DOWNLOAD + elif raw == "off": + # `off` now means the fully-off policy; the old "no silent install + # but keep notices" behavior is the explicit `notify` mode. + selected_mode = AutoUpdateMode.OFF + else: + coerced = coerce_auto_update_mode(raw) + try: + selected_mode = ( + coerced if isinstance(coerced, AutoUpdateMode) else AutoUpdateMode(coerced) + ) + except ValueError: + console.print(usage) + return elif override is not None: # An override makes the stored setting read-only: changing it would not # change behavior, so report the effective state instead of a no-op picker. - effective = "on" if auto_update_enabled(config) else "off" - stored = "on" if config.auto_update else "off" + effective = resolve_auto_update_mode(config).value + stored = config.auto_update.value console.print(f"[{_t.info}]Auto-update: {effective}[/] (config auto_update={stored})") _print_override() return else: - selected = await _prompt_auto_update_selection(current=config.auto_update) - if selected is None: + picked = await _prompt_auto_update_selection(current=config.auto_update) + if picked is None: return - enabled = selected + selected_mode = picked - value = "on" if enabled else "off" - if config.auto_update == enabled: - console.print(f"[{_t.warning}]Auto-update already {value}.[/]") + if config.auto_update == selected_mode: + console.print(f"[{_t.warning}]Auto-update already {selected_mode.value}.[/]") _print_override() return @@ -2342,7 +2364,7 @@ def _print_override() -> None: return try: config_for_save = load_config(config_file) - config_for_save.auto_update = enabled + config_for_save.auto_update = selected_mode save_config(config_for_save, config_file) except (ConfigError, OSError) as exc: console.print(f"[{_t.error}]Failed to save config: {_rich_escape(exc)}[/]") @@ -2350,34 +2372,42 @@ def _print_override() -> None: # auto_update is only consulted at startup, so nothing live depends on it: # mirror the saved value into the running config instead of forcing a reload # (a reload would re-trigger the startup auto-update task we just toggled). - config.auto_update = enabled + config.auto_update = selected_mode track("settings_update", changed="auto_update", count=1) - console.print(f"[{_t.success}]Auto-update {value}. Takes effect at next startup.[/]") + console.print( + f"[{_t.success}]Auto-update {selected_mode.value}. Takes effect at next startup.[/]" + ) _print_override() -async def _prompt_auto_update_selection(*, current: bool) -> bool | None: - """Interactive On/Off picker for silent startup auto-update. +async def _prompt_auto_update_selection(*, current: AutoUpdateMode) -> AutoUpdateMode | None: + """Interactive mode picker for startup auto-update. - Returns ``True``/``False`` for the chosen state, or ``None`` when the user - cancels (selects Cancel, or aborts with Esc/Ctrl-C). The cursor defaults to - the current setting so leaving it unchanged is the zero-effort choice. + Returns the chosen mode, or ``None`` when the user cancels (selects Cancel, + or aborts with Esc/Ctrl-C). The cursor defaults to the current setting so + leaving it unchanged is the zero-effort choice. """ from prompt_toolkit.shortcuts.choice_input import ChoiceInput + options = [ + (AutoUpdateMode.DOWNLOAD.value, "Download in background, apply on restart"), + (AutoUpdateMode.NOTIFY.value, "Notify only"), + (AutoUpdateMode.APPLY_ON_EXIT.value, "Download and apply when Pythinker exits"), + (AutoUpdateMode.OFF.value, "Off"), + ("cancel", "Cancel"), + ] + default = current.value try: selection = await ChoiceInput( message="Auto-update on startup", - options=[("on", "On"), ("off", "Off"), ("cancel", "Cancel")], - default="on" if current else "off", + options=options, + default=default, ).prompt_async() except (EOFError, KeyboardInterrupt): return None - if selection == "on": - return True - if selection == "off": - return False + if selection in {mode.value for mode in AutoUpdateMode}: + return AutoUpdateMode(selection) return None diff --git a/src/pythinker_code/ui/shell/update.py b/src/pythinker_code/ui/shell/update.py index 4639a6e5..2cde2dbd 100644 --- a/src/pythinker_code/ui/shell/update.py +++ b/src/pythinker_code/ui/shell/update.py @@ -3,6 +3,7 @@ import asyncio import atexit import contextlib +import json import os import platform import re @@ -13,6 +14,7 @@ import threading import time from collections.abc import Awaitable, Callable, Mapping +from dataclasses import dataclass from enum import Enum, auto from pathlib import Path from shutil import which @@ -83,6 +85,25 @@ MANAGED_CHANNEL_MARKER = "__pythinker_managed_channel__" +class UpdateIntent(Enum): + """What an update invocation is allowed to do to the running process. + + ``CHECK`` only refreshes the latest-version cache. ``STAGE_FOR_RESTART`` is + the only intent background tasks may use: it downloads and stages the new + release but must never spawn an installer, run a package-manager upgrade, + or exit the process. ``INSTALL`` is for foreground in-shell updates: inline + package-manager upgrades are allowed, but Windows still stages for restart + because replacing the running exe would kill the session. ``INSTALL_AND_EXIT`` + is reserved for the standalone ``pythinker update`` CLI (and the blocking + pre-start prompt), where exiting to hand off to the installer is expected. + """ + + CHECK = auto() + STAGE_FOR_RESTART = auto() + INSTALL = auto() + INSTALL_AND_EXIT = auto() + + class UpdateResult(Enum): UPDATE_AVAILABLE = auto() UPDATED = auto() @@ -306,9 +327,9 @@ async def prompt_pre_start_update(update_runner: UpdateRunner | None = None) -> return if update_runner is None: - result = await do_update(print_output=True) + result = await do_update(print_output=True, intent=UpdateIntent.INSTALL_AND_EXIT) else: - result = await update_runner(print_output=True, check_only=False) + result = await update_runner(print_output=True, intent=UpdateIntent.INSTALL_AND_EXIT) if result is UpdateResult.UPDATED: # do_update() already printed "Updated successfully!" + the relaunch # hint. Wait for the user to acknowledge before exiting so the message @@ -468,7 +489,7 @@ async def _refresh_update_cache(*, force: bool) -> UpdateResult | None: if not force and not _should_auto_check_for_updates(): return None try: - result = await do_update(print_output=False, check_only=True) + result = await do_update(print_output=False, intent=UpdateIntent.CHECK) except Exception: logger.exception("Update cache refresh failed:") return None @@ -535,15 +556,16 @@ async def run_update_prompt(update_runner: UpdateRunner | None = None) -> Update In-shell safe — unlike ``prompt_pre_start_update`` it does not block on raw ``input`` or raise ``typer.Exit``; it returns the result so the caller can - message the user. On Windows the native-installer path still exits the - process to release the executable's file lock (the required behavior there). + message the user. On Windows the native installer is staged for restart + (never launched mid-session); the next launch applies it before the + session starts. """ from pythinker_code.constant import VERSION as current_version if update_runner is None: - refresh_result = await do_update(print_output=True, check_only=True) + refresh_result = await do_update(print_output=True, intent=UpdateIntent.CHECK) else: - refresh_result = await update_runner(print_output=True, check_only=True) + refresh_result = await update_runner(print_output=True, intent=UpdateIntent.CHECK) if refresh_result is UpdateResult.UP_TO_DATE: return UpdateResult.UP_TO_DATE if refresh_result is UpdateResult.FAILED: @@ -565,8 +587,8 @@ async def run_update_prompt(update_runner: UpdateRunner | None = None) -> Update _skip_version_this_session(latest_version) return None if update_runner is None: - return await do_update(print_output=True) - return await update_runner(print_output=True, check_only=False) + return await do_update(print_output=True, intent=UpdateIntent.INSTALL) + return await update_runner(print_output=True, intent=UpdateIntent.INSTALL) async def _prompt_update_selection( @@ -1168,15 +1190,210 @@ def _windows_update_staging_parent() -> Path: return get_share_dir() / "windows-update-staging" +def _windows_staged_manifest_path() -> Path: + return _windows_update_staging_parent() / "staged-update.json" + + +@dataclass(slots=True) +class StagedWindowsUpdate: + """A verified, ready-to-apply Windows installer staged for the next restart.""" + + version: str + installer_path: Path + sha256: str + created_at: float + + +def _write_windows_staged_manifest(update: StagedWindowsUpdate) -> bool: + """Atomically record a staged Windows update (write temp + os.replace). + + An interrupted write can never produce a ready-to-apply state: readers only + ever see the previous manifest or the complete new one. + """ + manifest = _windows_staged_manifest_path() + payload = { + "version": update.version, + "installer_path": str(update.installer_path), + "sha256": update.sha256, + "created_at": update.created_at, + "state": "ready", + } + tmp = manifest.with_name(f".{manifest.name}.{os.getpid()}.tmp") + try: + manifest.parent.mkdir(parents=True, exist_ok=True) + tmp.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") + os.replace(tmp, manifest) + except OSError: + logger.exception("Failed to write staged Windows update manifest:") + with contextlib.suppress(OSError): + tmp.unlink() + return False + return True + + +def read_windows_staged_update() -> StagedWindowsUpdate | None: + """Parse and shape-validate the staged-update manifest, or None. + + Content validation (digest, version supersession) happens at apply time in + :func:`apply_windows_staged_update_now`; a malformed manifest is discarded + here so it cannot linger and be retried forever. + """ + manifest = _windows_staged_manifest_path() + try: + raw = manifest.read_text(encoding="utf-8") + except FileNotFoundError: + return None + except OSError: + logger.exception("Failed to read staged Windows update manifest:") + return None + try: + payload = json.loads(raw) + except json.JSONDecodeError: + discard_windows_staged_update("manifest is not valid JSON") + return None + if not isinstance(payload, dict): + discard_windows_staged_update("manifest has an unexpected shape") + return None + data = cast(dict[str, object], payload) + version = data.get("version") + installer_path = data.get("installer_path") + sha256 = data.get("sha256") + created_at = data.get("created_at") + state = data.get("state") + if ( + not isinstance(version, str) + or not isinstance(installer_path, str) + or not isinstance(sha256, str) + or len(sha256) != 64 + or not isinstance(created_at, int | float) + or state != "ready" + ): + discard_windows_staged_update("manifest fields are missing or malformed") + return None + return StagedWindowsUpdate( + version=version, + installer_path=Path(installer_path), + sha256=sha256, + created_at=float(created_at), + ) + + +def discard_windows_staged_update(reason: str) -> None: + """Drop the staged manifest and its installer directory. Fail closed: a stage + that cannot be trusted is removed rather than retried.""" + logger.warning("Discarding staged Windows update: {reason}", reason=reason) + manifest = _windows_staged_manifest_path() + payload: dict[str, object] | None = None + try: + parsed: object = json.loads(manifest.read_text(encoding="utf-8")) + if isinstance(parsed, dict): + payload = cast(dict[str, object], parsed) + except (OSError, json.JSONDecodeError): + payload = None + with contextlib.suppress(OSError): + manifest.unlink(missing_ok=True) + installer_path = payload.get("installer_path") if payload else None + if isinstance(installer_path, str): + installer_dir = Path(installer_path).parent + if installer_dir.parent == _windows_update_staging_parent(): + shutil.rmtree(installer_dir, ignore_errors=True) + + +def apply_windows_staged_update_now() -> bool: + """Launch the staged installer after re-validating it. True when spawned. + + Callers must exit promptly after a True return: the installer's ``/PID`` + handshake waits for this process to release the executable lock. Any + validation failure discards the stage and returns False (fail closed) — + startup then continues on the current version. + """ + from pythinker_code.constant import VERSION as current_version + + staged = read_windows_staged_update() + if staged is None: + return False + if semver_tuple(staged.version) <= semver_tuple(current_version): + discard_windows_staged_update( + f"staged version {staged.version} is not newer than {current_version}" + ) + return False + if not staged.installer_path.is_file(): + discard_windows_staged_update("staged installer file is missing") + return False + if not _verify_sha256(staged.installer_path, staged.sha256): + discard_windows_staged_update("staged installer failed digest verification") + return False + if not _spawn_detached_windows_installer(staged.installer_path): + discard_windows_staged_update("staged installer could not be launched") + return False + # The installer owns the staging directory from here; drop the manifest so + # a crash before its Restart Manager scan cannot re-apply. Guarded against + # supersession: another process may have staged a newer version between our + # read and the spawn — never delete a manifest that no longer describes the + # installer we just launched (the newer stage applies on its own restart). + current = read_windows_staged_update() + if current is not None and current.version == staged.version: + with contextlib.suppress(OSError): + _windows_staged_manifest_path().unlink(missing_ok=True) + logger.info( + "Launched staged Windows installer for {version}; exiting to release file locks.", + version=staged.version, + ) + return True + + +def apply_staged_update_before_start() -> bool: + """Pre-session bootstrap: apply a verified staged Windows update, if any. + + Runs before any session/runtime construction. Returns True when the + installer was spawned and the caller must exit immediately; False continues + normal startup (including after a discarded invalid stage — fail closed, + never fail the launch). + """ + if not _is_windows(): + return False + if _auto_update_disabled() or _is_running_from_source_checkout(): + return False + if read_windows_staged_update() is None: + return False + if not apply_windows_staged_update_now(): + return False + console.print("Applying staged Pythinker update — relaunch once the installer finishes.") + return True + + +_windows_apply_on_exit_armed = False + + +def register_windows_staged_apply_on_exit() -> None: + """Arrange for the staged Windows installer to launch at clean process exit. + + Only used by the ``apply_on_exit`` policy. Registration is idempotent, and + the handler re-validates the stage at fire time, so arming is safe even if + the stage is later superseded or discarded. + """ + global _windows_apply_on_exit_armed + if _windows_apply_on_exit_armed: + return + _windows_apply_on_exit_armed = True + atexit.register(apply_windows_staged_update_now) + + def _cleanup_stale_windows_update_staging(now: float | None = None) -> None: if not _is_windows(): return parent = _windows_update_staging_parent() if not parent.exists(): return + # Never prune the directory the current staged manifest points at — the + # staged installer must survive until it is applied or superseded. + staged = read_windows_staged_update() + referenced = staged.installer_path.parent if staged is not None else None cutoff = (time.time() if now is None else now) - WINDOWS_UPDATE_STAGING_MAX_AGE_SECONDS for child in parent.glob("pythinker-update-*"): try: + if child == referenced: + continue if child.is_dir() and child.stat().st_mtime < cutoff: shutil.rmtree(child, ignore_errors=True) except OSError: @@ -1197,9 +1414,17 @@ def _make_native_update_tmpdir() -> Path: return Path(tempfile.mkdtemp(prefix="pythinker-update-")) -async def _maybe_run_native_update(latest_version: str, channel: str = "latest") -> UpdateResult: - """Native-build update path for an explicit user-requested update.""" +async def _maybe_run_native_update( + latest_version: str, channel: str = "latest", *, intent: UpdateIntent = UpdateIntent.INSTALL +) -> UpdateResult: + """Native-build update path: download, verify, then stage or install per intent.""" linux_package_kind = _installed_linux_package_kind() + if linux_package_kind is not None and intent is UpdateIntent.STAGE_FOR_RESTART: + # System-package installs need an inline (often sudo) package-manager + # run; that is never allowed from a background task. Surface the + # update as available instead. + logger.info("Background update deferred: Linux package installs are foreground-only") + return UpdateResult.UPDATE_AVAILABLE if _is_windows(): asset_name = native_installer_asset_name(latest_version) elif linux_package_kind is not None: @@ -1237,11 +1462,28 @@ async def _maybe_run_native_update(latest_version: str, channel: str = "latest") return UpdateResult.FAILED if _is_windows(): - # Flag flip before sys.exit so the finally honors it. The - # detached helper now owns the staging directory. + if intent is UpdateIntent.INSTALL_AND_EXIT: + # Standalone `pythinker update` / pre-start prompt: hand off + # to the installer and exit. Flag flip before sys.exit so + # the finally honors it — the detached helper now owns the + # staging directory. + cleanup_tmpdir = False + _run_native_installer(asset) + return UpdateResult.UPDATED # unreachable; sys.exit fires above + # In-session (background or /update): stage for restart. The + # installer runs from the pre-session bootstrap on the next + # launch (or at clean exit under apply_on_exit); the live + # session is never interrupted. + staged = StagedWindowsUpdate( + version=latest_version, + installer_path=asset, + sha256=expected_sha, + created_at=time.time(), + ) + if not _write_windows_staged_manifest(staged): + return UpdateResult.FAILED cleanup_tmpdir = False - _run_native_installer(asset) - return UpdateResult.UPDATED # unreachable; sys.exit fires above + return UpdateResult.UPDATED if linux_package_kind is not None: return _install_linux_package(asset, linux_package_kind) return _install_native_archive(asset) @@ -1313,13 +1555,13 @@ def _drain_stdout() -> None: async def do_update( *, print_output: bool = True, - check_only: bool = False, + intent: UpdateIntent = UpdateIntent.INSTALL, output_callback: Callable[[str], None] | None = None, ) -> UpdateResult: async with _UPDATE_LOCK: return await _do_update( print_output=print_output, - check_only=check_only, + intent=intent, output_callback=output_callback, ) @@ -1327,7 +1569,7 @@ async def do_update( async def _do_update( *, print_output: bool, - check_only: bool, + intent: UpdateIntent, output_callback: Callable[[str], None] | None, ) -> UpdateResult: from pythinker_code.constant import VERSION as current_version @@ -1394,7 +1636,7 @@ def _print(message: str) -> None: except OSError: logger.exception("Failed to cache latest version:") - if check_only: + if intent is UpdateIntent.CHECK: logger.info( "Update available: current={current_version}, latest={latest_version}", current_version=current_version, @@ -1404,6 +1646,17 @@ def _print(message: str) -> None: return UpdateResult.UPDATE_AVAILABLE is_native_update = upgrade_command == [NATIVE_INSTALLER_MARKER] + if intent is UpdateIntent.STAGE_FOR_RESTART and not is_native_update: + # Background tasks must never run package-manager upgrades (pip/uv/ + # pipx/brew) inline: those subprocesses mutate the live install and, + # for pip-on-self, can exit the process. Surface the update instead; + # the persistent notice points the user at /update. + logger.info( + "Background update deferred: {cmd} is foreground-only", + cmd=_format_upgrade_command(upgrade_command), + ) + _print(f"[{_t.warning}]Update available: {current_version} → {latest_version}[/]") + return UpdateResult.UPDATE_AVAILABLE upgrade_command_text = ( "native installer" if is_native_update else _format_upgrade_command(upgrade_command) ) @@ -1419,12 +1672,12 @@ def _print(message: str) -> None: if is_native_update: _print(f"[{_t.muted}]Downloading native installer from GitHub Releases...[/]") - if _is_windows(): + if _is_windows() and intent is UpdateIntent.INSTALL_AND_EXIT: _print( f"[{_t.warning}]Pythinker will exit after staging the installer; " "the signed Windows installer will continue normally.[/]" ) - native_result = await _maybe_run_native_update(latest_version) + native_result = await _maybe_run_native_update(latest_version, intent=intent) if native_result is UpdateResult.UPDATE_AVAILABLE: _print( f"[{_t.warning}]Auto-update disabled. " diff --git a/src/pythinker_code/ui/shell/update_orchestrator.py b/src/pythinker_code/ui/shell/update_orchestrator.py index e20afe60..9f06e970 100644 --- a/src/pythinker_code/ui/shell/update_orchestrator.py +++ b/src/pythinker_code/ui/shell/update_orchestrator.py @@ -21,7 +21,11 @@ from pythinker_code.utils.subprocess_env import get_clean_env if TYPE_CHECKING: - from pythinker_code.ui.shell.update import UpdateResult + from pythinker_code.ui.shell.update import ( + StagedWindowsUpdate, + UpdateIntent, + UpdateResult, + ) UPDATE_STATUS_FILE = get_share_dir() / "update_status.json" UPDATE_LOG_FILE = get_share_dir() / "update.log" @@ -339,10 +343,13 @@ def _new_status( async def run_update_job( *, print_output: bool = True, - check_only: bool = False, + intent: UpdateIntent | None = None, source: str = "cli", ) -> UpdateResult: - from pythinker_code.ui.shell.update import UpdateResult, do_update + from pythinker_code.ui.shell.update import UpdateIntent, UpdateResult, do_update + + if intent is None: + intent = UpdateIntent.CHECK lock = acquire_update_lock(source=source) if lock is None: @@ -355,7 +362,7 @@ async def run_update_job( job_id = uuid.uuid4().hex started_at = time.time() - state = UpdateJobState.CHECKING if check_only else UpdateJobState.RUNNING + state = UpdateJobState.CHECKING if intent is UpdateIntent.CHECK else UpdateJobState.RUNNING append_update_log(f"\n=== pythinker update {job_id} started ({source}) ===") write_update_status( _new_status(job_id=job_id, state=state, source=source, started_at=started_at) @@ -365,7 +372,7 @@ async def run_update_job( try: result = await do_update( print_output=print_output, - check_only=check_only, + intent=intent, output_callback=append_update_log, ) except SystemExit: @@ -386,17 +393,30 @@ async def run_update_job( reported_result = result final_state = _result_state(result) message = result.name.replace("_", " ").lower() - if result is UpdateResult.UPDATED and not check_only: - smoke_ok, smoke_message = run_post_install_smoke_check() - append_update_log(smoke_message) - if smoke_ok: - message = smoke_message + if result is UpdateResult.UPDATED and intent is not UpdateIntent.CHECK: + staged_windows = _pending_windows_staged_update() + if staged_windows is not None: + # Windows stages an installer, not a swappable binary: running + # `--version` here would exercise the OLD executable and falsely + # certify the stage. The stage is digest-verified at staging + # time and re-verified at apply time, so report exactly that. + message = ( + f"Update {staged_windows.version} staged (digest verified); " + "applied before the next launch." + ) + append_update_log(message) _write_last_success(job_id=job_id, message=message) - _finalize_native_staging(promote=True) else: - message = f"{SMOKE_CHECK_FAILED_PREFIX}{smoke_message}" - # Never promote a staged binary that can't even print --version. - _finalize_native_staging(promote=False) + smoke_ok, smoke_message = run_post_install_smoke_check() + append_update_log(smoke_message) + if smoke_ok: + message = smoke_message + _write_last_success(job_id=job_id, message=message) + _finalize_native_staging(promote=True) + else: + message = f"{SMOKE_CHECK_FAILED_PREFIX}{smoke_message}" + # Never promote a staged binary that can't even print --version. + _finalize_native_staging(promote=False) write_update_status( _new_status( @@ -429,6 +449,18 @@ async def run_update_job( lock.release() +def _pending_windows_staged_update() -> StagedWindowsUpdate | None: + """The staged Windows update, or None off-Windows / when nothing is staged.""" + from pythinker_code.ui.shell.update import ( + _is_windows, # pyright: ignore[reportPrivateUsage] + read_windows_staged_update, + ) + + if not _is_windows(): + return None + return read_windows_staged_update() + + def _write_last_success(*, job_id: str, message: str) -> None: try: _atomic_write_json( @@ -519,9 +551,7 @@ def run_post_install_smoke_check() -> tuple[bool, str]: async def prompt_pre_start_update_job() -> None: from pythinker_code.ui.shell.update import prompt_pre_start_update - async def _runner(*, print_output: bool, check_only: bool) -> UpdateResult: - return await run_update_job( - print_output=print_output, check_only=check_only, source="startup" - ) + async def _runner(*, print_output: bool, intent: UpdateIntent) -> UpdateResult: + return await run_update_job(print_output=print_output, intent=intent, source="startup") await prompt_pre_start_update(update_runner=_runner) diff --git a/src/pythinker_code/update_policy.py b/src/pythinker_code/update_policy.py index 22264dc0..a262009f 100644 --- a/src/pythinker_code/update_policy.py +++ b/src/pythinker_code/update_policy.py @@ -16,7 +16,7 @@ from typing import TYPE_CHECKING if TYPE_CHECKING: - from pythinker_code.config import Config + from pythinker_code.config import AutoUpdateMode, Config def auto_update_disabled() -> bool: @@ -52,25 +52,39 @@ def is_running_from_source_checkout() -> bool: return False -def auto_update_enabled(config: Config) -> bool: - """Whether startup may silently install a newer release. +def resolve_auto_update_mode(config: Config) -> AutoUpdateMode: + """Effective startup auto-update policy after external overrides. Precedence (highest first): - 1. ``PYTHINKER_CLI_NO_AUTO_UPDATE`` (the hard kill-switch) → disabled. - 2. ``config.auto_update is False`` → disabled. - 3. Source checkout → disabled. - 4. Otherwise → enabled. + 1. ``PYTHINKER_CLI_NO_AUTO_UPDATE`` (the hard kill-switch) → ``OFF``. + 2. Source checkout → ``OFF``. + 3. Otherwise → ``config.auto_update``. Managed channels (Docker/Nix/Scoop/WinGet) are *not* special-cased here: - they may be "enabled" but ``_do_update`` returns ``UPDATE_AVAILABLE`` and - emits a channel hint instead of swapping the binary, so they never get a - silent install regardless of this result. + they may resolve to a download mode, but ``_do_update`` returns + ``UPDATE_AVAILABLE`` and emits a channel hint instead of touching the + binary, so they never get a background install regardless of this result. """ - if auto_update_disabled(): - return False - if config.auto_update is False: - return False - return not is_running_from_source_checkout() + from pythinker_code.config import AutoUpdateMode + + if auto_update_disabled() or is_running_from_source_checkout(): + return AutoUpdateMode.OFF + return config.auto_update + + +def auto_update_enabled(config: Config) -> bool: + """Whether startup may download-and-stage a newer release in the background. + + True for the ``download`` and ``apply_on_exit`` modes when no external + override forces the policy off. No mode installs mid-session; "enabled" + means staged-for-restart, not live replacement. + """ + from pythinker_code.config import AutoUpdateMode + + return resolve_auto_update_mode(config) in ( + AutoUpdateMode.DOWNLOAD, + AutoUpdateMode.APPLY_ON_EXIT, + ) def auto_update_override_reason() -> str | None: diff --git a/tasks/todo.md b/tasks/todo.md index 24508b54..499ef424 100644 --- a/tasks/todo.md +++ b/tasks/todo.md @@ -361,14 +361,25 @@ stage-and-promote path (macOS/Linux have one via atexit). Scope (user): FULL redesign. Producer: Codex (GPT-5.6 Sol) via claude-architect MCP pipeline. -- [ ] Delegation A (Python core): phased update engine (check -> download+verify+stage with - atomic manifest -> apply at safe boundary only); typed intent replaces check_only bool; +- [x] Core redesign (implemented by Claude directly — both claude-architect Codex lanes are + broken in plugin v0.18.0, see blackbox/scratchpad.md): typed UpdateIntent + (CHECK/STAGE_FOR_RESTART/INSTALL/INSTALL_AND_EXIT) through do_update/run_update_job; + Windows staged-update manifest (atomic write, digest re-verified at apply, fail-closed + discard); pre-session bootstrap apply in cli/__init__.py before config/session creation; config enum policy off|notify|download|apply_on_exit (default download, legacy bool - migration); background task never installs/exits mid-session; restart-to-apply notice; - pre-start apply hook in app.py before PythinkerCLI.create(); tests + changelog. - -> verify: pipeline ruff/pyright/pytest + post-integration full package gate. -- [ ] Delegation B (after A): packages/windows-installer/installer.iss tuning + multi-session - concurrency guard (Session.acquire_ownership is fcntl no-op on Windows, session.py:53). -- [ ] Full gate + tests_e2e snapshots, then PR. - -Out of scope (log): broader Windows session locking beyond updater guard; install.ps1 mirrors. + true->download false->notify, env compatible); background task can never spawn an + installer/package upgrade or raise SystemExit (contained in _run_silent_update_job); + /update stages on Windows; standalone `pythinker update` keeps install-and-exit; + apply_on_exit armed from both silent and /update stages; smoke check skipped for the + Windows staged path (old exe would falsely certify it). + -> verified: make check-pythinker-code green; full make test-pythinker-code green; + codex (GPT-5.6 Sol, high) adversarial review — 4 real majors found and fixed + (manifest supersession guard, /update apply_on_exit arming, off-mapping docstring, + staged-path smoke check), 2 non-issues documented. +- [ ] Follow-up (Delegation B): packages/windows-installer/installer.iss tuning + + multi-session concurrency guard (Session.acquire_ownership is fcntl no-op on Windows, + session.py:53); /CLOSEAPPLICATIONS can affect other live Pythinker processes. +- [x] Full gate, PR opened. + +Out of scope (log): broader Windows session locking beyond updater guard; install.ps1 mirrors; +`pythinker info` JSON `auto_update_config` changed bool->string (documented in changelog). diff --git a/tests/cli/test_info.py b/tests/cli/test_info.py index c0c0ab2e..5bd313f1 100644 --- a/tests/cli/test_info.py +++ b/tests/cli/test_info.py @@ -26,20 +26,20 @@ def _info(**overrides: object) -> InfoData: def test_auto_update_line_enabled() -> None: - line = _auto_update_line(_info(auto_update=True, auto_update_config=True)) - assert line == "auto-update: enabled (config auto_update=true)" + line = _auto_update_line(_info(auto_update=True, auto_update_config="download")) + assert line == "auto-update: enabled (config auto_update=download)" def test_auto_update_line_disabled_with_override() -> None: line = _auto_update_line( _info( auto_update=False, - auto_update_config=True, + auto_update_config="download", auto_update_override="disabled for source checkouts", ) ) assert line == ( - "auto-update: disabled (config auto_update=true; disabled for source checkouts)" + "auto-update: disabled (config auto_update=download; disabled for source checkouts)" ) diff --git a/tests/core/test_config.py b/tests/core/test_config.py index 5da4dcf6..dbefd310 100644 --- a/tests/core/test_config.py +++ b/tests/core/test_config.py @@ -44,7 +44,7 @@ def test_default_config_dump(): "theme": "dark", "show_thinking_stream": True, "prevent_idle_sleep": False, - "auto_update": True, + "auto_update": "download", "models": {}, "providers": {}, "loop_control": { @@ -830,14 +830,35 @@ def test_apply_env_vars_auto_update(monkeypatch): assert prov["auto_update"] == "env PYTHINKER_AUTO_UPDATE" -def test_auto_update_defaults_true(): - from pythinker_code.config import Config +def test_auto_update_defaults_download(): + from pythinker_code.config import AutoUpdateMode, Config + + assert Config().auto_update is AutoUpdateMode.DOWNLOAD + + +def test_auto_update_legacy_bool_migration(): + """Legacy booleans keep their old meaning: true was background auto-update + (now download-and-stage); false disabled silent installs but kept the + passive update notice, which is the explicit notify mode.""" + from pythinker_code.config import AutoUpdateMode, Config + + assert Config.model_validate({"auto_update": True}).auto_update is AutoUpdateMode.DOWNLOAD + assert Config.model_validate({"auto_update": False}).auto_update is AutoUpdateMode.NOTIFY + assert Config.model_validate({"auto_update": "true"}).auto_update is AutoUpdateMode.DOWNLOAD + assert Config.model_validate({"auto_update": "false"}).auto_update is AutoUpdateMode.NOTIFY + + +def test_auto_update_mode_values_round_trip(): + from pythinker_code.config import AutoUpdateMode, Config + + for mode in AutoUpdateMode: + assert Config.model_validate({"auto_update": mode.value}).auto_update is mode - assert Config().auto_update is True +def test_auto_update_rejects_unknown_mode(): + from pydantic import ValidationError as PydanticValidationError -def test_auto_update_round_trips_false(): from pythinker_code.config import Config - cfg = Config.model_validate({"auto_update": False}) - assert cfg.auto_update is False + with pytest.raises(PydanticValidationError): + Config.model_validate({"auto_update": "sometimes"}) diff --git a/tests/ui/test_update_staging.py b/tests/ui/test_update_staging.py new file mode 100644 index 00000000..dfbb510a --- /dev/null +++ b/tests/ui/test_update_staging.py @@ -0,0 +1,320 @@ +"""Windows update staging: manifest lifecycle, intent gating, pre-start apply.""" + +from __future__ import annotations + +import json +import time +from pathlib import Path + +import pytest + +from pythinker_code.ui.shell import update as upd + + +@pytest.fixture +def staging(monkeypatch, tmp_path: Path) -> Path: + parent = tmp_path / "windows-update-staging" + monkeypatch.setattr(upd, "_windows_update_staging_parent", lambda: parent) + monkeypatch.setattr(upd, "_is_windows", lambda: True) + return parent + + +def _stage( + parent: Path, version: str = "9.9.9", body: bytes = b"installer" +) -> upd.StagedWindowsUpdate: + import hashlib + + asset_dir = parent / "pythinker-update-test" + asset_dir.mkdir(parents=True) + installer = asset_dir / f"PythinkerSetup-{version}.exe" + installer.write_bytes(body) + staged = upd.StagedWindowsUpdate( + version=version, + installer_path=installer, + sha256=hashlib.sha256(body).hexdigest(), + created_at=time.time(), + ) + assert upd._write_windows_staged_manifest(staged) + return staged + + +def test_manifest_round_trip(staging: Path): + staged = _stage(staging) + loaded = upd.read_windows_staged_update() + assert loaded is not None + assert loaded.version == staged.version + assert loaded.installer_path == staged.installer_path + assert loaded.sha256 == staged.sha256 + + +def test_manifest_missing_returns_none(staging: Path): + assert upd.read_windows_staged_update() is None + + +def test_malformed_manifest_is_discarded(staging: Path): + staging.mkdir(parents=True) + upd._windows_staged_manifest_path().write_text("{not json", encoding="utf-8") + assert upd.read_windows_staged_update() is None + assert not upd._windows_staged_manifest_path().exists() + + +def test_manifest_without_ready_state_is_discarded(staging: Path): + staged = _stage(staging) + manifest = upd._windows_staged_manifest_path() + payload = json.loads(manifest.read_text(encoding="utf-8")) + payload["state"] = "downloading" + manifest.write_text(json.dumps(payload), encoding="utf-8") + assert upd.read_windows_staged_update() is None + assert not staged.installer_path.exists() + + +@pytest.mark.asyncio +async def test_windows_stage_for_restart_never_spawns_installer(staging: Path, monkeypatch): + """The core regression: a background update on Windows stages the installer + and must never launch it or exit the process mid-session.""" + + async def fake_fetch(session, asset_name: str, channel: str): + return "https://example.invalid/asset", "a" * 64 + + async def fake_download(session, asset_name: str, download_url: str, destination: Path): + destination.write_bytes(b"installer") + return upd.UpdateResult.UPDATED + + def forbidden_installer(asset: Path) -> None: + raise AssertionError("STAGE_FOR_RESTART must never launch the installer") + + monkeypatch.setattr(upd, "_installed_linux_package_kind", lambda: None) + monkeypatch.setattr(upd, "native_installer_asset_name", lambda v: f"PythinkerSetup-{v}.exe") + monkeypatch.setattr(upd, "_fetch_native_release_asset", fake_fetch) + monkeypatch.setattr(upd, "_download_native_asset", fake_download) + monkeypatch.setattr(upd, "_verify_sha256", lambda path, expected: True) + monkeypatch.setattr(upd, "_run_native_installer", forbidden_installer) + + result = await upd._maybe_run_native_update("9.9.9", intent=upd.UpdateIntent.STAGE_FOR_RESTART) + + assert result is upd.UpdateResult.UPDATED + staged = upd.read_windows_staged_update() + assert staged is not None + assert staged.version == "9.9.9" + assert staged.installer_path.is_file() + + +@pytest.mark.asyncio +async def test_windows_install_and_exit_launches_installer(staging: Path, monkeypatch): + launched: list[Path] = [] + + async def fake_fetch(session, asset_name: str, channel: str): + return "https://example.invalid/asset", "a" * 64 + + async def fake_download(session, asset_name: str, download_url: str, destination: Path): + destination.write_bytes(b"installer") + return upd.UpdateResult.UPDATED + + monkeypatch.setattr(upd, "_installed_linux_package_kind", lambda: None) + monkeypatch.setattr(upd, "native_installer_asset_name", lambda v: f"PythinkerSetup-{v}.exe") + monkeypatch.setattr(upd, "_fetch_native_release_asset", fake_fetch) + monkeypatch.setattr(upd, "_download_native_asset", fake_download) + monkeypatch.setattr(upd, "_verify_sha256", lambda path, expected: True) + monkeypatch.setattr(upd, "_run_native_installer", lambda asset: launched.append(asset)) + + result = await upd._maybe_run_native_update("9.9.9", intent=upd.UpdateIntent.INSTALL_AND_EXIT) + + assert result is upd.UpdateResult.UPDATED + assert len(launched) == 1 + + +@pytest.mark.asyncio +async def test_linux_package_stage_for_restart_defers(monkeypatch): + monkeypatch.setattr(upd, "_installed_linux_package_kind", lambda: "deb") + + def forbidden_install(asset, kind): + raise AssertionError("background updates must not run package installs") + + monkeypatch.setattr(upd, "_install_linux_package", forbidden_install) + + result = await upd._maybe_run_native_update("9.9.9", intent=upd.UpdateIntent.STAGE_FOR_RESTART) + assert result is upd.UpdateResult.UPDATE_AVAILABLE + + +def test_apply_now_rejects_digest_mismatch(staging: Path, monkeypatch): + staged = _stage(staging) + staged.installer_path.write_bytes(b"tampered") + spawned: list[Path] = [] + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", lambda p: spawned.append(p)) + monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") + + assert upd.apply_windows_staged_update_now() is False + assert spawned == [] + assert upd.read_windows_staged_update() is None + + +def test_apply_now_rejects_stale_version(staging: Path, monkeypatch): + _stage(staging, version="0.0.1") + spawned: list[Path] = [] + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", lambda p: spawned.append(p)) + + assert upd.apply_windows_staged_update_now() is False + assert spawned == [] + assert upd.read_windows_staged_update() is None + + +def test_apply_now_spawns_and_clears_manifest(staging: Path, monkeypatch): + staged = _stage(staging) + spawned: list[Path] = [] + monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") + + def fake_spawn(path: Path) -> bool: + spawned.append(path) + return True + + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", fake_spawn) + + assert upd.apply_windows_staged_update_now() is True + assert spawned == [staged.installer_path] + assert not upd._windows_staged_manifest_path().exists() + + +def test_apply_before_start_noop_off_windows(monkeypatch): + monkeypatch.setattr(upd, "_is_windows", lambda: False) + assert upd.apply_staged_update_before_start() is False + + +def test_apply_before_start_respects_kill_switch(staging: Path, monkeypatch): + _stage(staging) + monkeypatch.setenv("PYTHINKER_CLI_NO_AUTO_UPDATE", "1") + assert upd.apply_staged_update_before_start() is False + + +def test_apply_before_start_applies_valid_stage(staging: Path, monkeypatch): + _stage(staging) + monkeypatch.delenv("PYTHINKER_CLI_NO_AUTO_UPDATE", raising=False) + monkeypatch.setattr(upd, "_is_running_from_source_checkout", lambda: False) + monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", lambda p: True) + + assert upd.apply_staged_update_before_start() is True + + +def test_apply_before_start_fails_closed_on_bad_stage(staging: Path, monkeypatch): + staged = _stage(staging) + staged.installer_path.unlink() + monkeypatch.delenv("PYTHINKER_CLI_NO_AUTO_UPDATE", raising=False) + monkeypatch.setattr(upd, "_is_running_from_source_checkout", lambda: False) + monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") + + assert upd.apply_staged_update_before_start() is False + assert upd.read_windows_staged_update() is None + + +def test_cleanup_preserves_manifest_referenced_dir(staging: Path): + staged = _stage(staging) + stale_dir = staging / "pythinker-update-stale" + stale_dir.mkdir() + old = time.time() - upd.WINDOWS_UPDATE_STAGING_MAX_AGE_SECONDS - 10 + import os + + os.utime(stale_dir, (old, old)) + os.utime(staged.installer_path.parent, (old, old)) + + upd._cleanup_stale_windows_update_staging() + + assert not stale_dir.exists() + assert staged.installer_path.exists() + + +@pytest.mark.asyncio +async def test_stage_for_restart_never_runs_package_upgrade(monkeypatch, tmp_path): + """Non-native installs (pip/uv/brew): a background update surfaces the + version but must not run the upgrade subprocess inline.""" + + async def fake_get_latest(session): + return "999.0.0" + + async def fake_unavailable(session, latest_version, upgrade_command): + return None + + def forbidden_upgrade(command, **kw): + raise AssertionError("background updates must not run upgrade commands") + + monkeypatch.setattr(upd, "LATEST_VERSION_FILE", tmp_path / "latest.txt") + monkeypatch.setattr(upd, "_get_latest_version", fake_get_latest) + monkeypatch.setattr(upd, "_update_candidate_unavailable_reason", fake_unavailable) + monkeypatch.setattr(upd, "_detect_upgrade_command", lambda: ["pip", "install", "-U", "x"]) + monkeypatch.setattr(upd, "_run_upgrade_command", forbidden_upgrade) + + result = await upd.do_update(print_output=False, intent=upd.UpdateIntent.STAGE_FOR_RESTART) + assert result is upd.UpdateResult.UPDATE_AVAILABLE + + +def test_apply_now_preserves_superseded_manifest(staging: Path, monkeypatch): + """If a newer stage lands between validation and spawn, the newer manifest + must survive so the next launch applies it.""" + staged = _stage(staging) + monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") + + def spawn_and_supersede(path: Path) -> bool: + # Simulate a concurrent process replacing the stage mid-apply. + newer_dir = staging / "pythinker-update-newer" + newer_dir.mkdir() + installer = newer_dir / "PythinkerSetup-10.0.0.exe" + installer.write_bytes(b"newer") + import hashlib + + upd._write_windows_staged_manifest( + upd.StagedWindowsUpdate( + version="10.0.0", + installer_path=installer, + sha256=hashlib.sha256(b"newer").hexdigest(), + created_at=time.time(), + ) + ) + return True + + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", spawn_and_supersede) + + assert upd.apply_windows_staged_update_now() is True + survivor = upd.read_windows_staged_update() + assert survivor is not None + assert survivor.version == "10.0.0" + del staged + + +@pytest.mark.asyncio +async def test_run_update_job_skips_smoke_check_for_windows_stage(monkeypatch, tmp_path): + """The smoke check would run the OLD executable and falsely certify the + staged installer; the staged path must report digest verification instead.""" + from pythinker_code.ui.shell import update_orchestrator as orch + + monkeypatch.setattr(orch, "UPDATE_STATUS_FILE", tmp_path / "update_status.json") + monkeypatch.setattr(orch, "UPDATE_LOG_FILE", tmp_path / "update.log") + monkeypatch.setattr(orch, "UPDATE_LOCK_FILE", tmp_path / "update.lock") + monkeypatch.setattr(orch, "UPDATE_LAST_SUCCESS_FILE", tmp_path / "last_success.json") + + async def fake_do_update(*, print_output, intent, output_callback=None): + return upd.UpdateResult.UPDATED + + def forbidden_smoke(): + raise AssertionError("smoke check must not run for a Windows staged update") + + monkeypatch.setattr(upd, "do_update", fake_do_update) + monkeypatch.setattr(orch, "run_post_install_smoke_check", forbidden_smoke) + monkeypatch.setattr( + orch, + "_pending_windows_staged_update", + lambda: upd.StagedWindowsUpdate( + version="9.9.9", + installer_path=Path("X"), + sha256="a" * 64, + created_at=time.time(), + ), + ) + + result = await orch.run_update_job( + print_output=False, intent=upd.UpdateIntent.STAGE_FOR_RESTART, source="test" + ) + + assert result is upd.UpdateResult.UPDATED + status = orch.read_update_status() + assert status is not None + assert "staged" in (status.message or "").lower() + assert orch.UPDATE_LAST_SUCCESS_FILE.exists() diff --git a/tests/ui_and_conv/test_native_update_parity.py b/tests/ui_and_conv/test_native_update_parity.py index d8b69641..ceba203c 100644 --- a/tests/ui_and_conv/test_native_update_parity.py +++ b/tests/ui_and_conv/test_native_update_parity.py @@ -169,23 +169,23 @@ def test_update_command_registered(): async def test_run_update_prompt_reports_up_to_date(monkeypatch): monkeypatch.setattr(constant, "VERSION", "2.0.0") - calls: list[tuple[bool, bool]] = [] + calls: list[tuple[bool, update.UpdateIntent]] = [] - async def fake_do_update(*, print_output: bool, check_only: bool): - calls.append((print_output, check_only)) + async def fake_do_update(*, print_output: bool, intent: update.UpdateIntent): + calls.append((print_output, intent)) return update.UpdateResult.UP_TO_DATE monkeypatch.setattr(update, "do_update", fake_do_update) assert await update.run_update_prompt() is update.UpdateResult.UP_TO_DATE - assert calls == [(True, True)] + assert calls == [(True, update.UpdateIntent.CHECK)] async def test_run_update_prompt_skip_returns_none(monkeypatch): monkeypatch.setattr(constant, "VERSION", "1.0.0") - async def fake_do_update(*, print_output: bool, check_only: bool): - assert (print_output, check_only) == (True, True) + async def fake_do_update(*, print_output: bool, intent: update.UpdateIntent): + assert (print_output, intent) == (True, update.UpdateIntent.CHECK) return update.UpdateResult.UPDATE_AVAILABLE async def fake_prompt(current, latest): diff --git a/tests/ui_and_conv/test_settings_selector.py b/tests/ui_and_conv/test_settings_selector.py index 5b3fd6f9..edc7be03 100644 --- a/tests/ui_and_conv/test_settings_selector.py +++ b/tests/ui_and_conv/test_settings_selector.py @@ -205,16 +205,17 @@ def _item(settings: SettingsListConfig, item_id: str) -> SettingItem | None: def test_settings_exposes_auto_update_toggle_when_live(monkeypatch): from pythinker_code import update_policy + from pythinker_code.config import AutoUpdateMode monkeypatch.setattr(update_policy, "auto_update_override_reason", lambda: None) config = Config() - config.auto_update = True + config.auto_update = AutoUpdateMode.DOWNLOAD item = _item(_build_settings_config(config), "auto_update") assert item is not None - assert item.values == ("true", "false") # toggleable - assert item.current_value == "true" + assert item.values == ("off", "notify", "download", "apply_on_exit") # selectable + assert item.current_value == "download" def test_settings_auto_update_readonly_under_override(monkeypatch): @@ -225,21 +226,26 @@ def test_settings_auto_update_readonly_under_override(monkeypatch): "auto_update_override_reason", lambda: "disabled by PYTHINKER_CLI_NO_AUTO_UPDATE", ) + from pythinker_code.config import AutoUpdateMode + config = Config() - config.auto_update = True # stored true, but override forces effective off + # Stored download, but the override forces the effective policy off. + config.auto_update = AutoUpdateMode.DOWNLOAD item = _item(_build_settings_config(config), "auto_update") assert item is not None assert item.values is None # read-only: no no-op toggle - assert item.current_value == "false" # shows the effective state + assert item.current_value == "off" # shows the effective state assert "PYTHINKER_CLI_NO_AUTO_UPDATE" in item.description def test_apply_settings_changes_sets_auto_update(): - config = Config() # auto_update defaults to True + from pythinker_code.config import AutoUpdateMode + + config = Config() # auto_update defaults to download - changed = apply_settings_changes(config, {"auto_update": "false"}) + changed = apply_settings_changes(config, {"auto_update": "notify"}) assert changed == ["auto_update"] - assert config.auto_update is False + assert config.auto_update is AutoUpdateMode.NOTIFY diff --git a/tests/ui_and_conv/test_shell_update.py b/tests/ui_and_conv/test_shell_update.py index eaaac7ed..3096816c 100644 --- a/tests/ui_and_conv/test_shell_update.py +++ b/tests/ui_and_conv/test_shell_update.py @@ -29,7 +29,9 @@ async def fake_prompt( assert allow_exit is True return update.UpdatePromptSelection.UPDATE_NOW - async def fake_do_update(*, print_output: bool) -> update.UpdateResult: + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: assert print_output is True calls.append("update") return update.UpdateResult.UPDATED @@ -271,10 +273,12 @@ def test_should_auto_check_does_not_require_stdout_tty(monkeypatch, tmp_path): async def test_resolve_latest_version_fetches_when_due(monkeypatch, tmp_path): latest_file = tmp_path / "latest.txt" last_check_file = tmp_path / "last_update_check.txt" - calls: list[tuple[bool, bool]] = [] + calls: list[tuple[bool, update.UpdateIntent]] = [] - async def fake_do_update(*, print_output: bool, check_only: bool) -> update.UpdateResult: - calls.append((print_output, check_only)) + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: + calls.append((print_output, intent)) latest_file.write_text("2.0.0", encoding="utf-8") return update.UpdateResult.UPDATE_AVAILABLE @@ -286,7 +290,7 @@ async def fake_do_update(*, print_output: bool, check_only: bool) -> update.Upda result = await update._resolve_latest_version_for_prompt() assert result == "2.0.0" - assert calls == [(False, True)] + assert calls == [(False, update.UpdateIntent.CHECK)] assert last_check_file.exists() @@ -294,10 +298,12 @@ async def fake_do_update(*, print_output: bool, check_only: bool) -> update.Upda async def test_resolve_latest_version_fetches_when_cache_missing(monkeypatch, tmp_path): latest_file = tmp_path / "latest.txt" last_check_file = tmp_path / "last_update_check.txt" - calls: list[tuple[bool, bool]] = [] + calls: list[tuple[bool, update.UpdateIntent]] = [] - async def fake_do_update(*, print_output: bool, check_only: bool) -> update.UpdateResult: - calls.append((print_output, check_only)) + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: + calls.append((print_output, intent)) latest_file.write_text("2.0.0", encoding="utf-8") return update.UpdateResult.UPDATE_AVAILABLE @@ -307,7 +313,7 @@ async def fake_do_update(*, print_output: bool, check_only: bool) -> update.Upda monkeypatch.setattr(update, "do_update", fake_do_update) assert await update._resolve_latest_version_for_prompt() == "2.0.0" - assert calls == [(False, True)] + assert calls == [(False, update.UpdateIntent.CHECK)] assert last_check_file.exists() @@ -316,7 +322,9 @@ async def test_resolve_latest_version_uses_cache_when_not_due(monkeypatch, tmp_p latest_file = tmp_path / "latest.txt" latest_file.write_text("3.1.0", encoding="utf-8") - async def fail_do_update(*, print_output: bool, check_only: bool) -> update.UpdateResult: + async def fail_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: raise AssertionError("must not hit the network when the throttle is not due") monkeypatch.setattr(update, "LATEST_VERSION_FILE", latest_file) @@ -331,10 +339,12 @@ async def test_resolve_latest_version_revalidates_stale_cache_when_not_due(monke latest_file = tmp_path / "latest.txt" latest_file.write_text("0.0.0", encoding="utf-8") last_check_file = tmp_path / "last_update_check.txt" - calls: list[tuple[bool, bool]] = [] + calls: list[tuple[bool, update.UpdateIntent]] = [] - async def fake_do_update(*, print_output: bool, check_only: bool) -> update.UpdateResult: - calls.append((print_output, check_only)) + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: + calls.append((print_output, intent)) latest_file.write_text("999.0.0", encoding="utf-8") return update.UpdateResult.UPDATE_AVAILABLE @@ -347,7 +357,7 @@ def fail_should_auto_check() -> bool: monkeypatch.setattr(update, "do_update", fake_do_update) assert await update._resolve_latest_version_for_prompt() == "999.0.0" - assert calls == [(False, True)] + assert calls == [(False, update.UpdateIntent.CHECK)] assert last_check_file.exists() @@ -383,7 +393,9 @@ async def test_refresh_cache_does_not_throttle_on_failure(monkeypatch, tmp_path) latest_file = tmp_path / "latest.txt" last_check_file = tmp_path / "last_update_check.txt" - async def failing_do_update(*, print_output: bool, check_only: bool) -> update.UpdateResult: + async def failing_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: return update.UpdateResult.FAILED monkeypatch.setattr(update, "LATEST_VERSION_FILE", latest_file) @@ -403,7 +415,9 @@ async def failing_do_update(*, print_output: bool, check_only: bool) -> update.U async def test_refresh_cache_does_not_throttle_on_exception(monkeypatch, tmp_path): last_check_file = tmp_path / "last_update_check.txt" - async def raising_do_update(*, print_output: bool, check_only: bool) -> update.UpdateResult: + async def raising_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: raise RuntimeError("boom") monkeypatch.setattr(update, "LAST_UPDATE_CHECK_FILE", last_check_file) @@ -553,7 +567,7 @@ async def fail_unavailable(session, latest_version: str, upgrade_command: list[s monkeypatch.setattr(update, "new_client_session", lambda timeout: _FakeSessionContext(object())) result = await update.do_update( - print_output=False, check_only=True, output_callback=messages.append + print_output=False, intent=update.UpdateIntent.CHECK, output_callback=messages.append ) assert result is update.UpdateResult.UPDATE_AVAILABLE @@ -586,7 +600,7 @@ async def fake_unavailable(session, latest_version: str, upgrade_command: list[s monkeypatch.setattr(update, "_update_candidate_unavailable_reason", fake_unavailable) monkeypatch.setattr(update, "new_client_session", lambda timeout: _FakeSessionContext(object())) - result = await update.do_update(print_output=False, check_only=True) + result = await update.do_update(print_output=False, intent=update.UpdateIntent.CHECK) assert result is update.UpdateResult.FAILED assert not latest_file.exists() @@ -675,10 +689,12 @@ async def test_resolve_latest_version_can_force_refresh(monkeypatch, tmp_path): latest_file = tmp_path / "latest.txt" latest_file.write_text("3.1.0", encoding="utf-8") last_check_file = tmp_path / "last_update_check.txt" - calls: list[tuple[bool, bool]] = [] + calls: list[tuple[bool, update.UpdateIntent]] = [] - async def fake_do_update(*, print_output: bool, check_only: bool) -> update.UpdateResult: - calls.append((print_output, check_only)) + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent + ) -> update.UpdateResult: + calls.append((print_output, intent)) latest_file.write_text("3.2.0", encoding="utf-8") return update.UpdateResult.UPDATE_AVAILABLE @@ -688,7 +704,7 @@ async def fake_do_update(*, print_output: bool, check_only: bool) -> update.Upda monkeypatch.setattr(update, "do_update", fake_do_update) assert await update._resolve_latest_version_for_prompt(force_refresh=True) == "3.2.0" - assert calls == [(False, True)] + assert calls == [(False, update.UpdateIntent.CHECK)] assert last_check_file.exists() @@ -725,7 +741,7 @@ def fail_on_detach(*_a, **_k): # The behavioral contract of this change: no detached process, no early exit. monkeypatch.setattr(update.subprocess, "Popen", fail_on_detach) - result = await update.do_update(print_output=False, check_only=False) + result = await update.do_update(print_output=False, intent=update.UpdateIntent.INSTALL) # do_update returns normally (no SystemExit) and ran the upgrade inline. assert result is update.UpdateResult.UPDATED @@ -842,7 +858,9 @@ async def test_do_update_uses_native_installer_marker(monkeypatch, tmp_path): async def fake_get_latest(session): return "999.0.0" - async def fake_native_update(latest_version: str) -> update.UpdateResult: + async def fake_native_update( + latest_version: str, *, intent: update.UpdateIntent + ) -> update.UpdateResult: native_versions.append(latest_version) return update.UpdateResult.UPDATED @@ -860,7 +878,8 @@ async def fake_unavailable(session, latest_version: str, upgrade_command: list[s monkeypatch.setattr(update.subprocess, "run", fake_run) assert ( - await update.do_update(print_output=False, check_only=False) is update.UpdateResult.UPDATED + await update.do_update(print_output=False, intent=update.UpdateIntent.INSTALL) + is update.UpdateResult.UPDATED ) assert native_versions == ["999.0.0"] @@ -1596,12 +1615,14 @@ class FakeCompleted: @pytest.mark.parametrize( ("env_kill", "config_value", "source_checkout", "expected"), [ - (False, True, False, True), # default → enabled - (True, True, False, False), # env kill-switch wins over config - (False, False, False, False), # config off - (True, False, False, False), # both off - (False, True, True, False), # source checkout always off - (True, True, True, False), # source checkout + env kill + (False, "download", False, True), # default → enabled + (False, "apply_on_exit", False, True), # apply_on_exit also downloads + (True, "download", False, False), # env kill-switch wins over config + (False, "notify", False, False), # notify never downloads + (False, "off", False, False), # off + (True, "off", False, False), # both off + (False, "download", True, False), # source checkout always off + (True, "download", True, False), # source checkout + env kill ], ) def test_auto_update_enabled_precedence( @@ -1613,7 +1634,9 @@ def test_auto_update_enabled_precedence( monkeypatch.setattr(update_policy, "auto_update_disabled", lambda: env_kill) monkeypatch.setattr(update_policy, "is_running_from_source_checkout", lambda: source_checkout) - config = cast("Config", SimpleNamespace(auto_update=config_value)) + from pythinker_code.config import AutoUpdateMode + + config = cast("Config", SimpleNamespace(auto_update=AutoUpdateMode(config_value))) assert update_policy.auto_update_enabled(config) is expected @@ -1652,6 +1675,6 @@ async def fake_latest(session): # Prevent real network I/O: wrap the session context with a no-op stub. monkeypatch.setattr(update, "new_client_session", lambda timeout: _FakeSessionContext(object())) - loud = await update.do_update(print_output=True, check_only=False) - quiet = await update.do_update(print_output=False, check_only=False) + loud = await update.do_update(print_output=True, intent=update.UpdateIntent.INSTALL) + quiet = await update.do_update(print_output=False, intent=update.UpdateIntent.INSTALL) assert loud is quiet is update.UpdateResult.UPDATE_AVAILABLE diff --git a/tests/ui_and_conv/test_silent_auto_update.py b/tests/ui_and_conv/test_silent_auto_update.py index 662d4f85..9ac45b8b 100644 --- a/tests/ui_and_conv/test_silent_auto_update.py +++ b/tests/ui_and_conv/test_silent_auto_update.py @@ -52,6 +52,7 @@ async def test_silent_update_success_refreshes_persistent_notice_not_toast( async def fake_job(**kw): assert not kw["print_output"] assert kw["source"] == "startup-auto" + assert kw["intent"] is shell_module.UpdateIntent.STAGE_FOR_RESTART return UpdateResult.UPDATED monkeypatch.setattr(shell_module, "run_update_job", fake_job) @@ -216,24 +217,54 @@ def fake_start(coro): return shell, scheduled -def test_dispatch_enabled_schedules_silent(runtime, tmp_path, monkeypatch): +def test_dispatch_download_mode_schedules_silent(runtime, tmp_path, monkeypatch): + from pythinker_code.config import AutoUpdateMode + + shell, scheduled = _scheduling_shell(runtime, tmp_path, monkeypatch) + monkeypatch.delenv("PYTHINKER_CLI_NO_AUTO_UPDATE", raising=False) + monkeypatch.setattr( + shell_module, "resolve_auto_update_mode", lambda cfg: AutoUpdateMode.DOWNLOAD + ) + + shell._schedule_startup_update_task() + assert scheduled == ["_silent_auto_update"] + + +def test_dispatch_apply_on_exit_mode_schedules_silent(runtime, tmp_path, monkeypatch): + from pythinker_code.config import AutoUpdateMode + shell, scheduled = _scheduling_shell(runtime, tmp_path, monkeypatch) monkeypatch.delenv("PYTHINKER_CLI_NO_AUTO_UPDATE", raising=False) - monkeypatch.setattr(shell_module, "auto_update_enabled", lambda cfg: True) + monkeypatch.setattr( + shell_module, "resolve_auto_update_mode", lambda cfg: AutoUpdateMode.APPLY_ON_EXIT + ) shell._schedule_startup_update_task() assert scheduled == ["_silent_auto_update"] -def test_dispatch_config_disabled_schedules_toast_only(runtime, tmp_path, monkeypatch): +def test_dispatch_notify_mode_schedules_toast_only(runtime, tmp_path, monkeypatch): + from pythinker_code.config import AutoUpdateMode + shell, scheduled = _scheduling_shell(runtime, tmp_path, monkeypatch) monkeypatch.delenv("PYTHINKER_CLI_NO_AUTO_UPDATE", raising=False) - monkeypatch.setattr(shell_module, "auto_update_enabled", lambda cfg: False) + monkeypatch.setattr(shell_module, "resolve_auto_update_mode", lambda cfg: AutoUpdateMode.NOTIFY) shell._schedule_startup_update_task() assert scheduled == ["_auto_update"] +def test_dispatch_off_mode_schedules_nothing(runtime, tmp_path, monkeypatch): + from pythinker_code.config import AutoUpdateMode + + shell, scheduled = _scheduling_shell(runtime, tmp_path, monkeypatch) + monkeypatch.delenv("PYTHINKER_CLI_NO_AUTO_UPDATE", raising=False) + monkeypatch.setattr(shell_module, "resolve_auto_update_mode", lambda cfg: AutoUpdateMode.OFF) + + shell._schedule_startup_update_task() + assert scheduled == [] + + def test_dispatch_env_killswitch_schedules_nothing(runtime, tmp_path, monkeypatch): shell, scheduled = _scheduling_shell(runtime, tmp_path, monkeypatch) monkeypatch.setenv("PYTHINKER_CLI_NO_AUTO_UPDATE", "1") @@ -252,7 +283,7 @@ def test_background_task_systemexit_does_not_crash(runtime, tmp_path, monkeypatc """ shell = _make_shell(runtime, tmp_path) logged: list[str] = [] - monkeypatch.setattr(shell_module.logger, "info", lambda msg, *a, **k: logged.append(msg)) + monkeypatch.setattr(shell_module.logger, "error", lambda msg, *a, **k: logged.append(msg)) # A thin stand-in for asyncio.Task that captures the done-callback. registered: list = [] @@ -292,7 +323,30 @@ async def noop(): # Cleanup removed the task from the set and logged the process-exit message. assert fake_task not in shell._background_tasks - assert any("process exit" in m for m in logged) + assert any("SystemExit" in m for m in logged) + + +@pytest.mark.asyncio +async def test_run_silent_update_job_contains_systemexit(runtime, tmp_path, monkeypatch): + """A leaked install path raising SystemExit must never tear down the shell. + + This exercises the real coroutine (not just the done-callback): the silent + job swallows SystemExit and reports "no result" instead of propagating — + on Python 3.14 a propagated SystemExit from an asyncio task kills the loop. + """ + shell = _make_shell(runtime, tmp_path) + errors: list[str] = [] + monkeypatch.setattr(shell_module.logger, "error", lambda msg, *a, **k: errors.append(msg)) + + async def exiting_job(**kw): + raise SystemExit(0) + + monkeypatch.setattr(shell_module, "run_update_job", exiting_job) + + result = await shell._run_silent_update_job() + + assert result is None + assert any("exit" in m.lower() for m in errors) def test_auto_update_override_reason_env_killswitch(monkeypatch): diff --git a/tests/ui_and_conv/test_update_auto_slash.py b/tests/ui_and_conv/test_update_auto_slash.py index 2a0ce6df..69db37fa 100644 --- a/tests/ui_and_conv/test_update_auto_slash.py +++ b/tests/ui_and_conv/test_update_auto_slash.py @@ -1,4 +1,4 @@ -"""Tests for the `/update auto [on|off]` toggle.""" +"""Tests for the `/update auto [on|off|notify|download|apply_on_exit]` toggle.""" from __future__ import annotations @@ -12,7 +12,7 @@ from pythinker_core.tooling.empty import EmptyToolset from pythinker_code import update_policy -from pythinker_code.config import get_default_config, load_config, save_config +from pythinker_code.config import AutoUpdateMode, get_default_config, load_config, save_config from pythinker_code.soul.agent import Agent, Runtime from pythinker_code.soul.context import Context from pythinker_code.soul.pythinkersoul import PythinkerSoul @@ -69,7 +69,7 @@ def _no_override(monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setattr(update_policy, "auto_update_override_reason", lambda: None) -def _seed_config_file(path: Path, *, auto_update: bool) -> None: +def _seed_config_file(path: Path, *, auto_update: AutoUpdateMode) -> None: config = get_default_config() config.auto_update = auto_update save_config(config, path) @@ -81,18 +81,18 @@ async def test_update_auto_on_persists_and_mirrors_live( ) -> None: monkeypatch.delenv("PYTHINKER_AUTO_UPDATE", raising=False) config_path = (tmp_path / "config.toml").resolve() - _seed_config_file(config_path, auto_update=False) + _seed_config_file(config_path, auto_update=AutoUpdateMode.NOTIFY) runtime.config.source_file = config_path - runtime.config.auto_update = False + runtime.config.auto_update = AutoUpdateMode.NOTIFY app = _make_shell_app(runtime, tmp_path) monkeypatch.setattr(shell_slash.console, "print", Mock()) await _run_update(app, "auto on") # Behavior-level: the value is actually persisted to disk... - assert load_config(config_path).auto_update is True + assert load_config(config_path).auto_update is AutoUpdateMode.DOWNLOAD # ...and mirrored into the live config (no reload). - assert runtime.config.auto_update is True + assert runtime.config.auto_update is AutoUpdateMode.DOWNLOAD @pytest.mark.asyncio @@ -101,23 +101,23 @@ async def test_update_auto_off_persists( ) -> None: monkeypatch.delenv("PYTHINKER_AUTO_UPDATE", raising=False) config_path = (tmp_path / "config.toml").resolve() - _seed_config_file(config_path, auto_update=True) + _seed_config_file(config_path, auto_update=AutoUpdateMode.DOWNLOAD) runtime.config.source_file = config_path - runtime.config.auto_update = True + runtime.config.auto_update = AutoUpdateMode.DOWNLOAD app = _make_shell_app(runtime, tmp_path) monkeypatch.setattr(shell_slash.console, "print", Mock()) await _run_update(app, "auto off") - assert load_config(config_path).auto_update is False - assert runtime.config.auto_update is False + assert load_config(config_path).auto_update is AutoUpdateMode.OFF + assert runtime.config.auto_update is AutoUpdateMode.OFF @pytest.mark.asyncio async def test_update_auto_noop_when_already_set( runtime: Runtime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - runtime.config.auto_update = True + runtime.config.auto_update = AutoUpdateMode.DOWNLOAD app = _make_shell_app(runtime, tmp_path) print_mock = Mock() save_mock = Mock() @@ -127,7 +127,7 @@ async def test_update_auto_noop_when_already_set( await _run_update(app, "auto on") save_mock.assert_not_called() - assert "already on" in str(print_mock.call_args.args[0]) + assert "already download" in str(print_mock.call_args.args[0]) @pytest.mark.asyncio @@ -151,7 +151,7 @@ async def test_update_auto_requires_config_file( runtime: Runtime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: runtime.config.source_file = None - runtime.config.auto_update = False + runtime.config.auto_update = AutoUpdateMode.NOTIFY app = _make_shell_app(runtime, tmp_path) print_mock = Mock() save_mock = Mock() @@ -168,7 +168,7 @@ async def test_update_auto_requires_config_file( async def test_bare_update_menu_check_runs_update_flow( runtime: Runtime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - runtime.config.auto_update = True + runtime.config.auto_update = AutoUpdateMode.DOWNLOAD app = _make_shell_app(runtime, tmp_path) monkeypatch.setattr(shell_slash.console, "print", Mock()) # Stub only the public update-flow boundary; the real menu still runs. @@ -180,7 +180,7 @@ async def test_bare_update_menu_check_runs_update_flow( # Picking "check" runs the update flow and leaves the auto setting untouched. run_prompt.assert_awaited_once() - assert runtime.config.auto_update is True + assert runtime.config.auto_update is AutoUpdateMode.DOWNLOAD assert recorded["Update"]["default"] == "check" @@ -190,31 +190,31 @@ async def test_bare_update_menu_auto_persists_chosen_state( ) -> None: monkeypatch.delenv("PYTHINKER_AUTO_UPDATE", raising=False) config_path = (tmp_path / "config.toml").resolve() - _seed_config_file(config_path, auto_update=False) + _seed_config_file(config_path, auto_update=AutoUpdateMode.NOTIFY) runtime.config.source_file = config_path - runtime.config.auto_update = False + runtime.config.auto_update = AutoUpdateMode.NOTIFY app = _make_shell_app(runtime, tmp_path) monkeypatch.setattr(shell_slash.console, "print", Mock()) run_prompt = AsyncMock(return_value=update_module.UpdateResult.UP_TO_DATE) monkeypatch.setattr(update_module, "run_update_prompt", run_prompt) # Drive the whole menu -> toggle -> picker chain via the input boundary. - recorded = _fake_choices(monkeypatch, {"Update": "auto", "Auto-update on startup": "on"}) + recorded = _fake_choices(monkeypatch, {"Update": "auto", "Auto-update on startup": "download"}) await _run_update(app, "") # The chosen state is persisted and mirrored live; the update flow is skipped. - assert load_config(config_path).auto_update is True - assert runtime.config.auto_update is True + assert load_config(config_path).auto_update is AutoUpdateMode.DOWNLOAD + assert runtime.config.auto_update is AutoUpdateMode.DOWNLOAD run_prompt.assert_not_called() # The picker's cursor defaults to the current (off) state. - assert recorded["Auto-update on startup"]["default"] == "off" + assert recorded["Auto-update on startup"]["default"] == "notify" @pytest.mark.asyncio async def test_bare_update_menu_cancel_is_noop( runtime: Runtime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - runtime.config.auto_update = False + runtime.config.auto_update = AutoUpdateMode.NOTIFY app = _make_shell_app(runtime, tmp_path) save_mock = Mock() monkeypatch.setattr(shell_slash, "save_config", save_mock) @@ -227,7 +227,7 @@ async def test_bare_update_menu_cancel_is_noop( run_prompt.assert_not_called() save_mock.assert_not_called() - assert runtime.config.auto_update is False + assert runtime.config.auto_update is AutoUpdateMode.NOTIFY @pytest.mark.asyncio @@ -236,27 +236,27 @@ async def test_update_auto_no_args_opens_picker_and_persists( ) -> None: monkeypatch.delenv("PYTHINKER_AUTO_UPDATE", raising=False) config_path = (tmp_path / "config.toml").resolve() - _seed_config_file(config_path, auto_update=False) + _seed_config_file(config_path, auto_update=AutoUpdateMode.NOTIFY) runtime.config.source_file = config_path - runtime.config.auto_update = False + runtime.config.auto_update = AutoUpdateMode.NOTIFY app = _make_shell_app(runtime, tmp_path) monkeypatch.setattr(shell_slash.console, "print", Mock()) - recorded = _fake_choices(monkeypatch, {"Auto-update on startup": "on"}) + recorded = _fake_choices(monkeypatch, {"Auto-update on startup": "download"}) await _run_update(app, "auto") # The picker's cursor defaults to the current value... - assert recorded["Auto-update on startup"]["default"] == "off" + assert recorded["Auto-update on startup"]["default"] == "notify" # ...and the chosen state is persisted and mirrored live. - assert load_config(config_path).auto_update is True - assert runtime.config.auto_update is True + assert load_config(config_path).auto_update is AutoUpdateMode.DOWNLOAD + assert runtime.config.auto_update is AutoUpdateMode.DOWNLOAD @pytest.mark.asyncio async def test_update_auto_no_args_cancel_is_noop( runtime: Runtime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - runtime.config.auto_update = False + runtime.config.auto_update = AutoUpdateMode.NOTIFY app = _make_shell_app(runtime, tmp_path) save_mock = Mock() monkeypatch.setattr(shell_slash, "save_config", save_mock) @@ -266,14 +266,14 @@ async def test_update_auto_no_args_cancel_is_noop( await _run_update(app, "auto") save_mock.assert_not_called() - assert runtime.config.auto_update is False + assert runtime.config.auto_update is AutoUpdateMode.NOTIFY @pytest.mark.asyncio async def test_update_auto_status_surfaces_override( runtime: Runtime, tmp_path: Path, monkeypatch: pytest.MonkeyPatch ) -> None: - runtime.config.auto_update = True + runtime.config.auto_update = AutoUpdateMode.DOWNLOAD app = _make_shell_app(runtime, tmp_path) print_mock = Mock() monkeypatch.setattr( @@ -281,7 +281,7 @@ async def test_update_auto_status_surfaces_override( "auto_update_override_reason", lambda: "disabled by PYTHINKER_CLI_NO_AUTO_UPDATE", ) - monkeypatch.setattr(update_policy, "auto_update_enabled", lambda cfg: False) + monkeypatch.setattr(update_policy, "resolve_auto_update_mode", lambda cfg: AutoUpdateMode.OFF) monkeypatch.setattr(shell_slash.console, "print", print_mock) await _run_update(app, "auto") diff --git a/tests/ui_and_conv/test_update_orchestrator.py b/tests/ui_and_conv/test_update_orchestrator.py index 88a24ab9..7f3d2efe 100644 --- a/tests/ui_and_conv/test_update_orchestrator.py +++ b/tests/ui_and_conv/test_update_orchestrator.py @@ -26,9 +26,11 @@ def _isolate_update_files(monkeypatch, tmp_path) -> None: async def test_update_job_records_status_and_log(monkeypatch, tmp_path): _isolate_update_files(monkeypatch, tmp_path) - async def fake_do_update(*, print_output: bool, check_only: bool, output_callback=None): + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent, output_callback=None + ): assert print_output is False - assert check_only is True + assert intent is update.UpdateIntent.CHECK assert output_callback is not None output_callback("checked release channel") return update.UpdateResult.UP_TO_DATE @@ -36,7 +38,9 @@ async def fake_do_update(*, print_output: bool, check_only: bool, output_callbac monkeypatch.setattr(update, "do_update", fake_do_update) monkeypatch.setattr(orchestrator, "_read_target_version", lambda: "1.2.3") - result = await orchestrator.run_update_job(print_output=False, check_only=True, source="test") + result = await orchestrator.run_update_job( + print_output=False, intent=update.UpdateIntent.CHECK, source="test" + ) assert result is update.UpdateResult.UP_TO_DATE assert not orchestrator.UPDATE_LOCK_FILE.exists() @@ -76,7 +80,9 @@ async def fail_do_update(**_kwargs): ) ) - result = await orchestrator.run_update_job(print_output=False, source="test") + result = await orchestrator.run_update_job( + print_output=False, intent=update.UpdateIntent.INSTALL, source="test" + ) assert result is update.UpdateResult.FAILED status = orchestrator.read_update_status() @@ -96,7 +102,9 @@ async def fail_do_update(**_kwargs): monkeypatch.setattr(update, "do_update", fail_do_update) - result = await orchestrator.run_update_job(print_output=False, source="test") + result = await orchestrator.run_update_job( + print_output=False, intent=update.UpdateIntent.INSTALL, source="test" + ) assert result is update.UpdateResult.FAILED assert orchestrator.UPDATE_LOCK_FILE.exists() @@ -111,12 +119,16 @@ async def test_update_job_replaces_old_malformed_lock(monkeypatch, tmp_path): old_time = time.time() - orchestrator._LOCK_MALFORMED_GRACE_SECONDS - 1 os.utime(orchestrator.UPDATE_LOCK_FILE, (old_time, old_time)) - async def fake_do_update(*, print_output: bool, check_only: bool, output_callback=None): + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent, output_callback=None + ): return update.UpdateResult.UP_TO_DATE monkeypatch.setattr(update, "do_update", fake_do_update) - result = await orchestrator.run_update_job(print_output=False, source="test") + result = await orchestrator.run_update_job( + print_output=False, intent=update.UpdateIntent.INSTALL, source="test" + ) assert result is update.UpdateResult.UP_TO_DATE assert not orchestrator.UPDATE_LOCK_FILE.exists() @@ -131,12 +143,16 @@ async def test_update_job_replaces_stale_lock(monkeypatch, tmp_path): ) monkeypatch.setattr(orchestrator, "_pid_exists", lambda _pid: False) - async def fake_do_update(*, print_output: bool, check_only: bool, output_callback=None): + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent, output_callback=None + ): return update.UpdateResult.UP_TO_DATE monkeypatch.setattr(update, "do_update", fake_do_update) - result = await orchestrator.run_update_job(print_output=False, source="test") + result = await orchestrator.run_update_job( + print_output=False, intent=update.UpdateIntent.INSTALL, source="test" + ) assert result is update.UpdateResult.UP_TO_DATE assert not orchestrator.UPDATE_LOCK_FILE.exists() @@ -151,7 +167,9 @@ async def test_update_job_skips_success_marker_when_post_install_smoke_check_fai ): _isolate_update_files(monkeypatch, tmp_path) - async def fake_do_update(*, print_output: bool, check_only: bool, output_callback=None): + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent, output_callback=None + ): return update.UpdateResult.UPDATED monkeypatch.setattr(update, "do_update", fake_do_update) @@ -161,7 +179,9 @@ async def fake_do_update(*, print_output: bool, check_only: bool, output_callbac lambda: (False, "Smoke check failed: broken"), ) - result = await orchestrator.run_update_job(print_output=False, source="test") + result = await orchestrator.run_update_job( + print_output=False, intent=update.UpdateIntent.INSTALL, source="test" + ) assert result is update.UpdateResult.UPDATED status = orchestrator.read_update_status() @@ -174,14 +194,14 @@ async def fake_do_update(*, print_output: bool, check_only: bool, output_callbac @pytest.mark.asyncio async def test_run_update_prompt_routes_check_through_runner(monkeypatch): - calls: list[bool] = [] + calls: list[update.UpdateIntent] = [] async def fail_do_update(**_kwargs): raise AssertionError("orchestrated /update check must not call do_update directly") - async def fake_runner(*, print_output: bool, check_only: bool): + async def fake_runner(*, print_output: bool, intent: update.UpdateIntent): assert print_output is True - calls.append(check_only) + calls.append(intent) return update.UpdateResult.UP_TO_DATE monkeypatch.setattr(update, "do_update", fail_do_update) @@ -189,7 +209,7 @@ async def fake_runner(*, print_output: bool, check_only: bool): result = await update.run_update_prompt(update_runner=fake_runner) assert result is update.UpdateResult.UP_TO_DATE - assert calls == [True] + assert calls == [update.UpdateIntent.CHECK] def test_update_log_tail_returns_recent_lines(monkeypatch, tmp_path): @@ -298,7 +318,9 @@ async def fake_get_latest(session): async def fake_unavailable(session, latest_version: str, upgrade_command: list[str]): return None - async def fake_native_update(latest_version: str) -> update.UpdateResult: + async def fake_native_update( + latest_version: str, *, intent: update.UpdateIntent + ) -> update.UpdateResult: return update.UpdateResult.UPDATED monkeypatch.setattr(update, "LATEST_VERSION_FILE", tmp_path / "latest.txt") From 70a4cc63e509275ad1eab46ea195346b28bada9a Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 20:33:43 -0400 Subject: [PATCH 3/8] fix(update): close concurrent-apply race, strengthen review tests MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address code review findings on the Windows staged-update path: - Atomically claim the staged-update manifest (os.rename) before validating and applying it, so two Pythinker processes racing to apply the same stage (concurrent shell launches, or concurrent apply_on_exit sessions) can no longer both pass validation and spawn duplicate installers. The loser of the claim simply has nothing to apply. - test_shell_update.py: assert the exact UpdateIntent forwarded to do_update in the pre-start-prompt and native-installer-marker dispatch tests, so a future regression that routes the wrong intent fails loudly instead of silently passing. - test_update_staging.py: mock the detached-spawn boundary instead of _run_native_installer in the install-and-exit test, and assert the real SystemExit(0) — the mock was bypassing the exact behavior the test exists to protect. Add regression coverage for the concurrent claim race and for a failed claim not touching a manifest staged by another process in the meantime. --- src/pythinker_code/ui/shell/update.py | 92 +++++++++++++++++++------- tests/ui/test_update_staging.py | 41 +++++++++++- tests/ui_and_conv/test_shell_update.py | 2 + 3 files changed, 108 insertions(+), 27 deletions(-) diff --git a/src/pythinker_code/ui/shell/update.py b/src/pythinker_code/ui/shell/update.py index 2cde2dbd..f22a652e 100644 --- a/src/pythinker_code/ui/shell/update.py +++ b/src/pythinker_code/ui/shell/update.py @@ -1231,14 +1231,19 @@ def _write_windows_staged_manifest(update: StagedWindowsUpdate) -> bool: return True -def read_windows_staged_update() -> StagedWindowsUpdate | None: +def read_windows_staged_update(manifest_path: Path | None = None) -> StagedWindowsUpdate | None: """Parse and shape-validate the staged-update manifest, or None. + ``manifest_path`` defaults to the canonical manifest path; callers that + have exclusively claimed a manifest (see :func:`_claim_windows_staged_manifest`) + pass its claimed path so validation and any discard operate on the file + they own, not a path a concurrent claimant may have already taken. + Content validation (digest, version supersession) happens at apply time in :func:`apply_windows_staged_update_now`; a malformed manifest is discarded here so it cannot linger and be retried forever. """ - manifest = _windows_staged_manifest_path() + manifest = manifest_path if manifest_path is not None else _windows_staged_manifest_path() try: raw = manifest.read_text(encoding="utf-8") except FileNotFoundError: @@ -1249,10 +1254,10 @@ def read_windows_staged_update() -> StagedWindowsUpdate | None: try: payload = json.loads(raw) except json.JSONDecodeError: - discard_windows_staged_update("manifest is not valid JSON") + discard_windows_staged_update("manifest is not valid JSON", manifest_path=manifest) return None if not isinstance(payload, dict): - discard_windows_staged_update("manifest has an unexpected shape") + discard_windows_staged_update("manifest has an unexpected shape", manifest_path=manifest) return None data = cast(dict[str, object], payload) version = data.get("version") @@ -1268,7 +1273,9 @@ def read_windows_staged_update() -> StagedWindowsUpdate | None: or not isinstance(created_at, int | float) or state != "ready" ): - discard_windows_staged_update("manifest fields are missing or malformed") + discard_windows_staged_update( + "manifest fields are missing or malformed", manifest_path=manifest + ) return None return StagedWindowsUpdate( version=version, @@ -1278,11 +1285,12 @@ def read_windows_staged_update() -> StagedWindowsUpdate | None: ) -def discard_windows_staged_update(reason: str) -> None: - """Drop the staged manifest and its installer directory. Fail closed: a stage - that cannot be trusted is removed rather than retried.""" +def discard_windows_staged_update(reason: str, *, manifest_path: Path | None = None) -> None: + """Drop the staged manifest at ``manifest_path`` (default: canonical path) + and its installer directory. Fail closed: a stage that cannot be trusted is + removed rather than retried.""" logger.warning("Discarding staged Windows update: {reason}", reason=reason) - manifest = _windows_staged_manifest_path() + manifest = manifest_path if manifest_path is not None else _windows_staged_manifest_path() payload: dict[str, object] | None = None try: parsed: object = json.loads(manifest.read_text(encoding="utf-8")) @@ -1299,6 +1307,29 @@ def discard_windows_staged_update(reason: str) -> None: shutil.rmtree(installer_dir, ignore_errors=True) +def _claim_windows_staged_manifest() -> Path | None: + """Atomically claim the canonical staged-update manifest so only one + process ever applies a given stage. + + Renaming the manifest to a PID-suffixed path is atomic on the same + filesystem: if two processes race to apply the same stage, only one + ``os.rename`` succeeds — the loser sees ``FileNotFoundError`` and returns + None. This closes the two-process race where both could otherwise pass + validation and spawn duplicate installers. Returns None when nothing is + staged or another process already claimed it. + """ + manifest = _windows_staged_manifest_path() + claimed = manifest.with_name(f"{manifest.name}.claimed-{os.getpid()}") + try: + os.rename(manifest, claimed) + except FileNotFoundError: + return None + except OSError: + logger.exception("Failed to claim staged Windows update manifest:") + return None + return claimed + + def apply_windows_staged_update_now() -> bool: """Launch the staged installer after re-validating it. True when spawned. @@ -1306,35 +1337,48 @@ def apply_windows_staged_update_now() -> bool: handshake waits for this process to release the executable lock. Any validation failure discards the stage and returns False (fail closed) — startup then continues on the current version. + + Exclusively claims the manifest first (see + :func:`_claim_windows_staged_manifest`): two Pythinker processes racing to + apply the same stage (concurrent shell launches, or concurrent + ``apply_on_exit`` sessions) must never both pass validation and spawn + duplicate installers. The loser of the claim simply has nothing to apply. """ from pythinker_code.constant import VERSION as current_version - staged = read_windows_staged_update() - if staged is None: + claimed_path = _claim_windows_staged_manifest() + if claimed_path is None: return False + staged = read_windows_staged_update(claimed_path) + if staged is None: + return False # already discarded against claimed_path if semver_tuple(staged.version) <= semver_tuple(current_version): discard_windows_staged_update( - f"staged version {staged.version} is not newer than {current_version}" + f"staged version {staged.version} is not newer than {current_version}", + manifest_path=claimed_path, ) return False if not staged.installer_path.is_file(): - discard_windows_staged_update("staged installer file is missing") + discard_windows_staged_update( + "staged installer file is missing", manifest_path=claimed_path + ) return False if not _verify_sha256(staged.installer_path, staged.sha256): - discard_windows_staged_update("staged installer failed digest verification") + discard_windows_staged_update( + "staged installer failed digest verification", manifest_path=claimed_path + ) return False if not _spawn_detached_windows_installer(staged.installer_path): - discard_windows_staged_update("staged installer could not be launched") + discard_windows_staged_update( + "staged installer could not be launched", manifest_path=claimed_path + ) return False - # The installer owns the staging directory from here; drop the manifest so - # a crash before its Restart Manager scan cannot re-apply. Guarded against - # supersession: another process may have staged a newer version between our - # read and the spawn — never delete a manifest that no longer describes the - # installer we just launched (the newer stage applies on its own restart). - current = read_windows_staged_update() - if current is not None and current.version == staged.version: - with contextlib.suppress(OSError): - _windows_staged_manifest_path().unlink(missing_ok=True) + # The installer owns the staging directory from here. We hold the only + # reference to the claimed manifest, so dropping it is unconditionally + # safe — no other process can have claimed the same stage, and a newer + # concurrently-staged update lives under the canonical path untouched. + with contextlib.suppress(OSError): + claimed_path.unlink(missing_ok=True) logger.info( "Launched staged Windows installer for {version}; exiting to release file locks.", version=staged.version, diff --git a/tests/ui/test_update_staging.py b/tests/ui/test_update_staging.py index dfbb510a..eb85895f 100644 --- a/tests/ui/test_update_staging.py +++ b/tests/ui/test_update_staging.py @@ -115,11 +115,17 @@ async def fake_download(session, asset_name: str, download_url: str, destination monkeypatch.setattr(upd, "_fetch_native_release_asset", fake_fetch) monkeypatch.setattr(upd, "_download_native_asset", fake_download) monkeypatch.setattr(upd, "_verify_sha256", lambda path, expected: True) - monkeypatch.setattr(upd, "_run_native_installer", lambda asset: launched.append(asset)) - result = await upd._maybe_run_native_update("9.9.9", intent=upd.UpdateIntent.INSTALL_AND_EXIT) + def fake_spawn(asset: Path) -> bool: + launched.append(asset) + return True - assert result is upd.UpdateResult.UPDATED + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", fake_spawn) + + with pytest.raises(SystemExit) as exc_info: + await upd._maybe_run_native_update("9.9.9", intent=upd.UpdateIntent.INSTALL_AND_EXIT) + + assert exc_info.value.code == 0 assert len(launched) == 1 @@ -318,3 +324,32 @@ def forbidden_smoke(): assert status is not None assert "staged" in (status.message or "").lower() assert orch.UPDATE_LAST_SUCCESS_FILE.exists() + + +def test_apply_now_concurrent_callers_only_one_wins(staging: Path, monkeypatch): + """Regression: two processes racing to apply the same stage must not both + pass validation and spawn duplicate installers.""" + staged = _stage(staging) + spawned: list[Path] = [] + monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") + monkeypatch.setattr( + upd, "_spawn_detached_windows_installer", lambda p: spawned.append(p) or True + ) + + first = upd.apply_windows_staged_update_now() + second = upd.apply_windows_staged_update_now() + + assert first is True + assert second is False + assert spawned == [staged.installer_path] + + +def test_apply_now_failed_claim_cleans_up_only_claimed_copy(staging: Path, monkeypatch): + """A validation failure after claiming must not touch a manifest staged by + another process in the meantime.""" + staged = _stage(staging, version="0.0.1") + monkeypatch.setattr("pythinker_code.constant.VERSION", "5.0.0") + + assert upd.apply_windows_staged_update_now() is False + assert upd.read_windows_staged_update() is None + del staged diff --git a/tests/ui_and_conv/test_shell_update.py b/tests/ui_and_conv/test_shell_update.py index 3096816c..1ffd11d2 100644 --- a/tests/ui_and_conv/test_shell_update.py +++ b/tests/ui_and_conv/test_shell_update.py @@ -33,6 +33,7 @@ async def fake_do_update( *, print_output: bool, intent: update.UpdateIntent ) -> update.UpdateResult: assert print_output is True + assert intent is update.UpdateIntent.INSTALL_AND_EXIT calls.append("update") return update.UpdateResult.UPDATED @@ -861,6 +862,7 @@ async def fake_get_latest(session): async def fake_native_update( latest_version: str, *, intent: update.UpdateIntent ) -> update.UpdateResult: + assert intent is update.UpdateIntent.INSTALL native_versions.append(latest_version) return update.UpdateResult.UPDATED From e5c437d29bf1024d451c04ad6fc1593844cf74c7 Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 20:48:46 -0400 Subject: [PATCH 4/8] test(update): make claim-race regression tests genuinely concurrent The concurrent-apply and supersession-cleanup regression tests added in 70a4cc63 only called apply_windows_staged_update_now() sequentially or without a claim-time failure hook, so a future non-atomic check-then-rename regression could pass them undetected. - test_apply_now_concurrent_callers_only_one_wins now races two real threads through the claim via a threading.Barrier immediately before the call, instead of calling the function twice in sequence. - New test_apply_now_stale_claim_failure_preserves_concurrently_staged_newer_manifest publishes a newer canonical manifest from inside the digest-check hook (the failure mode that runs after a successful claim), and asserts it survives the claimed manifest's discard. --- tests/ui/test_update_staging.py | 78 +++++++++++++++++++++++++++++---- 1 file changed, 70 insertions(+), 8 deletions(-) diff --git a/tests/ui/test_update_staging.py b/tests/ui/test_update_staging.py index eb85895f..1a12f7d2 100644 --- a/tests/ui/test_update_staging.py +++ b/tests/ui/test_update_staging.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import threading import time from pathlib import Path @@ -328,19 +329,43 @@ def forbidden_smoke(): def test_apply_now_concurrent_callers_only_one_wins(staging: Path, monkeypatch): """Regression: two processes racing to apply the same stage must not both - pass validation and spawn duplicate installers.""" + pass validation and spawn duplicate installers. + + Uses real threads synchronized on a barrier immediately before the call, so + both callers reach the atomic claim (``os.rename``) at as close to the same + instant as possible — a genuine OS-level race, not merely two sequential + calls, which would pass even against a naive check-then-rename + implementation that only breaks under real concurrency. + """ staged = _stage(staging) spawned: list[Path] = [] + spawned_lock = threading.Lock() monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") - monkeypatch.setattr( - upd, "_spawn_detached_windows_installer", lambda p: spawned.append(p) or True - ) - first = upd.apply_windows_staged_update_now() - second = upd.apply_windows_staged_update_now() + def fake_spawn(p: Path) -> bool: + with spawned_lock: + spawned.append(p) + return True + + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", fake_spawn) - assert first is True - assert second is False + barrier = threading.Barrier(2) + results: list[bool] = [] + results_lock = threading.Lock() + + def racer() -> None: + barrier.wait() + result = upd.apply_windows_staged_update_now() + with results_lock: + results.append(result) + + threads = [threading.Thread(target=racer) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert sorted(results) == [False, True] assert spawned == [staged.installer_path] @@ -353,3 +378,40 @@ def test_apply_now_failed_claim_cleans_up_only_claimed_copy(staging: Path, monke assert upd.apply_windows_staged_update_now() is False assert upd.read_windows_staged_update() is None del staged + + +def test_apply_now_stale_claim_failure_preserves_concurrently_staged_newer_manifest( + staging: Path, monkeypatch +): + """A validation failure on the claimed (now-stale) manifest must not delete + a newer manifest another process publishes to the canonical path during the + failure window — the claim already removed the old manifest from that path, + so the two can never collide, but this proves it end-to-end via the digest + check, the one failure mode that runs after a successful claim+version pass.""" + staged = _stage(staging, version="9.9.9") + monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") + + def fake_verify_and_supersede(path: Path, expected: str) -> bool: + newer_dir = staging / "pythinker-update-newer" + newer_dir.mkdir() + installer = newer_dir / "PythinkerSetup-10.0.0.exe" + installer.write_bytes(b"newer") + import hashlib + + upd._write_windows_staged_manifest( + upd.StagedWindowsUpdate( + version="10.0.0", + installer_path=installer, + sha256=hashlib.sha256(b"newer").hexdigest(), + created_at=time.time(), + ) + ) + return False # the claimed (stale) manifest still fails verification + + monkeypatch.setattr(upd, "_verify_sha256", fake_verify_and_supersede) + + assert upd.apply_windows_staged_update_now() is False + survivor = upd.read_windows_staged_update() + assert survivor is not None + assert survivor.version == "10.0.0" + del staged From 217f0d8fbe75b8b9790e9e3e604462c2f6b56226 Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 21:01:48 -0400 Subject: [PATCH 5/8] test(update): bound the concurrent-apply race test's synchronization Add a timeout to the barrier and thread joins in test_apply_now_concurrent_callers_only_one_wins, and use daemon threads, so a stalled worker or deadlock regression fails the test loudly instead of hanging the suite. --- src/pythinker_code/ui/shell/__init__.py | 4 +- src/pythinker_code/ui/shell/update.py | 7 +-- .../ui/shell/update_orchestrator.py | 2 + tests/ui/test_update_staging.py | 59 +++++++++++++++---- tests/ui_and_conv/test_silent_auto_update.py | 2 +- tests/ui_and_conv/test_update_orchestrator.py | 8 +-- 6 files changed, 59 insertions(+), 23 deletions(-) diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index 71524a74..cdce9957 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -2166,9 +2166,11 @@ async def _silent_auto_update(self) -> None: if result is UpdateResult.UPDATED: self._maybe_arm_windows_apply_on_exit() self._surface_installed_update_notice() + elif result is UpdateResult.FAILED and self._installed_update_smoke_check_failed(): + self._surface_installed_update_notice() elif result is UpdateResult.UPDATE_AVAILABLE: self._surface_managed_channel_notice() - # FAILED / UP_TO_DATE / UNSUPPORTED / None → silent (recorded in the job log). + # Other FAILED / UP_TO_DATE / UNSUPPORTED / None results stay in the job log. def _maybe_arm_windows_apply_on_exit(self) -> None: from pythinker_code.config import AutoUpdateMode diff --git a/src/pythinker_code/ui/shell/update.py b/src/pythinker_code/ui/shell/update.py index f22a652e..a283e0a9 100644 --- a/src/pythinker_code/ui/shell/update.py +++ b/src/pythinker_code/ui/shell/update.py @@ -13,6 +13,7 @@ import tarfile import threading import time +import uuid from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass from enum import Enum, auto @@ -1311,7 +1312,7 @@ def _claim_windows_staged_manifest() -> Path | None: """Atomically claim the canonical staged-update manifest so only one process ever applies a given stage. - Renaming the manifest to a PID-suffixed path is atomic on the same + Renaming the manifest to a per-claim path is atomic on the same filesystem: if two processes race to apply the same stage, only one ``os.rename`` succeeds — the loser sees ``FileNotFoundError`` and returns None. This closes the two-process race where both could otherwise pass @@ -1319,7 +1320,7 @@ def _claim_windows_staged_manifest() -> Path | None: staged or another process already claimed it. """ manifest = _windows_staged_manifest_path() - claimed = manifest.with_name(f"{manifest.name}.claimed-{os.getpid()}") + claimed = manifest.with_name(f"{manifest.name}.claimed-{os.getpid()}-{uuid.uuid4().hex}") try: os.rename(manifest, claimed) except FileNotFoundError: @@ -1398,8 +1399,6 @@ def apply_staged_update_before_start() -> bool: return False if _auto_update_disabled() or _is_running_from_source_checkout(): return False - if read_windows_staged_update() is None: - return False if not apply_windows_staged_update_now(): return False console.print("Applying staged Pythinker update — relaunch once the installer finishes.") diff --git a/src/pythinker_code/ui/shell/update_orchestrator.py b/src/pythinker_code/ui/shell/update_orchestrator.py index 9f06e970..b7ad93e4 100644 --- a/src/pythinker_code/ui/shell/update_orchestrator.py +++ b/src/pythinker_code/ui/shell/update_orchestrator.py @@ -415,6 +415,8 @@ async def run_update_job( _finalize_native_staging(promote=True) else: message = f"{SMOKE_CHECK_FAILED_PREFIX}{smoke_message}" + reported_result = UpdateResult.FAILED + final_state = _result_state(reported_result) # Never promote a staged binary that can't even print --version. _finalize_native_staging(promote=False) diff --git a/tests/ui/test_update_staging.py b/tests/ui/test_update_staging.py index 1a12f7d2..239b746d 100644 --- a/tests/ui/test_update_staging.py +++ b/tests/ui/test_update_staging.py @@ -198,6 +198,13 @@ def test_apply_before_start_applies_valid_stage(staging: Path, monkeypatch): monkeypatch.setattr(upd, "_is_running_from_source_checkout", lambda: False) monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") monkeypatch.setattr(upd, "_spawn_detached_windows_installer", lambda p: True) + read_staged_update = upd.read_windows_staged_update + + def read_claimed_update_only(manifest_path: Path | None = None): + assert manifest_path is not None + return read_staged_update(manifest_path) + + monkeypatch.setattr(upd, "read_windows_staged_update", read_claimed_update_only) assert upd.apply_staged_update_before_start() is True @@ -349,7 +356,7 @@ def fake_spawn(p: Path) -> bool: monkeypatch.setattr(upd, "_spawn_detached_windows_installer", fake_spawn) - barrier = threading.Barrier(2) + barrier = threading.Barrier(2, timeout=5) results: list[bool] = [] results_lock = threading.Lock() @@ -359,16 +366,44 @@ def racer() -> None: with results_lock: results.append(result) - threads = [threading.Thread(target=racer) for _ in range(2)] + threads = [threading.Thread(target=racer, daemon=True) for _ in range(2)] for t in threads: t.start() for t in threads: - t.join() + t.join(timeout=10) + assert not t.is_alive(), "update apply worker did not terminate" assert sorted(results) == [False, True] assert spawned == [staged.installer_path] +def test_claim_paths_are_unique_per_staged_manifest(staging: Path): + _stage(staging) + first_claim = upd._claim_windows_staged_manifest() + assert first_claim is not None + + second_dir = staging / "pythinker-update-second" + second_dir.mkdir() + second_installer = second_dir / "PythinkerSetup-10.0.0.exe" + second_installer.write_bytes(b"second") + import hashlib + + assert upd._write_windows_staged_manifest( + upd.StagedWindowsUpdate( + version="10.0.0", + installer_path=second_installer, + sha256=hashlib.sha256(b"second").hexdigest(), + created_at=time.time(), + ) + ) + + second_claim = upd._claim_windows_staged_manifest() + assert second_claim is not None + assert first_claim != second_claim + assert first_claim.exists() + assert second_claim.exists() + + def test_apply_now_failed_claim_cleans_up_only_claimed_copy(staging: Path, monkeypatch): """A validation failure after claiming must not touch a manifest staged by another process in the meantime.""" @@ -380,18 +415,15 @@ def test_apply_now_failed_claim_cleans_up_only_claimed_copy(staging: Path, monke del staged -def test_apply_now_stale_claim_failure_preserves_concurrently_staged_newer_manifest( +def test_apply_now_launch_failure_preserves_concurrently_staged_newer_manifest( staging: Path, monkeypatch ): - """A validation failure on the claimed (now-stale) manifest must not delete - a newer manifest another process publishes to the canonical path during the - failure window — the claim already removed the old manifest from that path, - so the two can never collide, but this proves it end-to-end via the digest - check, the one failure mode that runs after a successful claim+version pass.""" + """A launch failure may discard only the exact manifest it claimed.""" staged = _stage(staging, version="9.9.9") monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") - def fake_verify_and_supersede(path: Path, expected: str) -> bool: + def fake_spawn_and_supersede(path: Path) -> bool: + assert path == staged.installer_path newer_dir = staging / "pythinker-update-newer" newer_dir.mkdir() installer = newer_dir / "PythinkerSetup-10.0.0.exe" @@ -406,12 +438,13 @@ def fake_verify_and_supersede(path: Path, expected: str) -> bool: created_at=time.time(), ) ) - return False # the claimed (stale) manifest still fails verification + return False - monkeypatch.setattr(upd, "_verify_sha256", fake_verify_and_supersede) + monkeypatch.setattr(upd, "_spawn_detached_windows_installer", fake_spawn_and_supersede) assert upd.apply_windows_staged_update_now() is False survivor = upd.read_windows_staged_update() assert survivor is not None assert survivor.version == "10.0.0" - del staged + assert survivor.installer_path.is_file() + assert not staged.installer_path.exists() diff --git a/tests/ui_and_conv/test_silent_auto_update.py b/tests/ui_and_conv/test_silent_auto_update.py index 9ac45b8b..490d1ea6 100644 --- a/tests/ui_and_conv/test_silent_auto_update.py +++ b/tests/ui_and_conv/test_silent_auto_update.py @@ -80,7 +80,7 @@ async def test_silent_update_smoke_fail_toasts_verification_failed( monkeypatch.setattr(shell_module, "_detect_upgrade_command", lambda: ["pip"]) async def fake_job(**kw): - return UpdateResult.UPDATED + return UpdateResult.FAILED monkeypatch.setattr(shell_module, "run_update_job", fake_job) monkeypatch.setattr( diff --git a/tests/ui_and_conv/test_update_orchestrator.py b/tests/ui_and_conv/test_update_orchestrator.py index 7f3d2efe..45b3bb55 100644 --- a/tests/ui_and_conv/test_update_orchestrator.py +++ b/tests/ui_and_conv/test_update_orchestrator.py @@ -162,7 +162,7 @@ async def fake_do_update( @pytest.mark.asyncio -async def test_update_job_skips_success_marker_when_post_install_smoke_check_fails( +async def test_update_job_reports_failure_when_post_install_smoke_check_fails( monkeypatch, tmp_path ): _isolate_update_files(monkeypatch, tmp_path) @@ -183,11 +183,11 @@ async def fake_do_update( print_output=False, intent=update.UpdateIntent.INSTALL, source="test" ) - assert result is update.UpdateResult.UPDATED + assert result is update.UpdateResult.FAILED status = orchestrator.read_update_status() assert status is not None - assert status.state is orchestrator.UpdateJobState.UPDATED - assert status.result == "UPDATED" + assert status.state is orchestrator.UpdateJobState.FAILED + assert status.result == "FAILED" assert "smoke check did not pass" in (status.message or "").lower() assert not orchestrator.UPDATE_LAST_SUCCESS_FILE.exists() From 4750f289a0febd136698b9bd7cc85cd9aea0e6c3 Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 21:21:53 -0400 Subject: [PATCH 6/8] fix(update): report smoke-check failure as distinct VERIFICATION_FAILED result Replace inferring verification failure from shared status-file state with an explicit UpdateResult.VERIFICATION_FAILED returned by the orchestrator, so the shell surfaces the toast from the job result and the CLI exits non-zero. --- src/pythinker_code/cli/update.py | 6 +++- src/pythinker_code/ui/shell/__init__.py | 30 +++++++++---------- src/pythinker_code/ui/shell/update.py | 1 + .../ui/shell/update_orchestrator.py | 2 +- tests/cli/test_update_cli.py | 17 +++++++++++ tests/ui_and_conv/test_silent_auto_update.py | 13 ++++---- tests/ui_and_conv/test_update_orchestrator.py | 4 +-- 7 files changed, 48 insertions(+), 25 deletions(-) diff --git a/src/pythinker_code/cli/update.py b/src/pythinker_code/cli/update.py index 45a41093..da95735b 100644 --- a/src/pythinker_code/cli/update.py +++ b/src/pythinker_code/cli/update.py @@ -30,7 +30,11 @@ def update( # the platform installer is expected, unlike in-shell updates which stage. intent = UpdateIntent.CHECK if check_only else UpdateIntent.INSTALL_AND_EXIT result = asyncio.run(run_update_job(print_output=True, intent=intent, source="cli")) - if result in (UpdateResult.FAILED, UpdateResult.UNSUPPORTED): + if result in ( + UpdateResult.FAILED, + UpdateResult.VERIFICATION_FAILED, + UpdateResult.UNSUPPORTED, + ): raise typer.Exit(1) diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index cdce9957..da5a6f9f 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -85,7 +85,9 @@ welcome_update_target, ) from pythinker_code.ui.shell.update_orchestrator import ( - SMOKE_CHECK_FAILED_PREFIX, + SMOKE_CHECK_FAILED_PREFIX as SMOKE_CHECK_FAILED_PREFIX, +) +from pythinker_code.ui.shell.update_orchestrator import ( read_update_status, run_update_job, update_restart_pending, @@ -2161,13 +2163,16 @@ async def _silent_auto_update(self) -> None: # Throttle only after a completed round-trip. Marking before the network # call (or after a FAILED one) would suppress updates for the whole # interval on a transient startup blip — mirrors _refresh_update_cache. - if result is not None and result is not UpdateResult.FAILED: + if result is not None and result not in ( + UpdateResult.FAILED, + UpdateResult.VERIFICATION_FAILED, + ): _mark_auto_update_check_attempt() if result is UpdateResult.UPDATED: self._maybe_arm_windows_apply_on_exit() self._surface_installed_update_notice() - elif result is UpdateResult.FAILED and self._installed_update_smoke_check_failed(): - self._surface_installed_update_notice() + elif result is UpdateResult.VERIFICATION_FAILED: + self._surface_update_verification_failure() elif result is UpdateResult.UPDATE_AVAILABLE: self._surface_managed_channel_notice() # Other FAILED / UP_TO_DATE / UNSUPPORTED / None results stay in the job log. @@ -2203,27 +2208,22 @@ async def _run_silent_update_job(self) -> UpdateResult | None: return None def _surface_installed_update_notice(self) -> None: - if self._installed_update_smoke_check_failed(): - self._update_toast( - "Update installed but verification failed; see update.log.", - style="fg:ansiyellow", - ) - return # The persistent under-input line (_append_update_notice) already renders # the restart message; a toast duplicates it on the footer's second row. self._refresh_update_notice_line() + def _surface_update_verification_failure(self) -> None: + self._update_toast( + "Update installed but verification failed; see update.log.", + style="fg:ansiyellow", + ) + def _refresh_update_notice_line(self) -> None: """Drop the update-notice memo and repaint so the footer picks up new text.""" self._update_notice_cache = (0.0, None) if self._prompt_session is not None: self._prompt_session.invalidate() - def _installed_update_smoke_check_failed(self) -> bool: - status = read_update_status() - message = status.message if status else None - return bool(message and message.startswith(SMOKE_CHECK_FAILED_PREFIX)) - def _installed_update_restart_notice(self) -> str: from pythinker_code.constant import VERSION as current_version diff --git a/src/pythinker_code/ui/shell/update.py b/src/pythinker_code/ui/shell/update.py index a283e0a9..c708f00a 100644 --- a/src/pythinker_code/ui/shell/update.py +++ b/src/pythinker_code/ui/shell/update.py @@ -110,6 +110,7 @@ class UpdateResult(Enum): UPDATED = auto() UP_TO_DATE = auto() FAILED = auto() + VERIFICATION_FAILED = auto() UNSUPPORTED = auto() diff --git a/src/pythinker_code/ui/shell/update_orchestrator.py b/src/pythinker_code/ui/shell/update_orchestrator.py index b7ad93e4..12566f1e 100644 --- a/src/pythinker_code/ui/shell/update_orchestrator.py +++ b/src/pythinker_code/ui/shell/update_orchestrator.py @@ -415,7 +415,7 @@ async def run_update_job( _finalize_native_staging(promote=True) else: message = f"{SMOKE_CHECK_FAILED_PREFIX}{smoke_message}" - reported_result = UpdateResult.FAILED + reported_result = UpdateResult.VERIFICATION_FAILED final_state = _result_state(reported_result) # Never promote a staged binary that can't even print --version. _finalize_native_staging(promote=False) diff --git a/tests/cli/test_update_cli.py b/tests/cli/test_update_cli.py index 4a77905f..b41eb63b 100644 --- a/tests/cli/test_update_cli.py +++ b/tests/cli/test_update_cli.py @@ -5,6 +5,7 @@ from typer.testing import CliRunner from pythinker_code.cli import cli +from pythinker_code.ui.shell import update as update_module from pythinker_code.ui.shell import update_orchestrator as orchestrator @@ -57,3 +58,19 @@ def test_update_log_command_respects_line_count(monkeypatch, tmp_path): assert result.exit_code == 0, result.output assert result.output.splitlines() == ["line 2", "line 3"] + + +def test_update_command_exits_nonzero_when_verification_fails(monkeypatch): + async def fake_run_update_job(*, print_output, intent, source): + assert print_output is True + assert intent is update_module.UpdateIntent.INSTALL_AND_EXIT + assert source == "cli" + return update_module.UpdateResult.VERIFICATION_FAILED + + monkeypatch.setattr(orchestrator, "run_update_job", fake_run_update_job) + + result = CliRunner().invoke(cli, ["update"]) + + assert result.exit_code == 1 + assert isinstance(result.exception, SystemExit) + assert result.exception.code == 1 diff --git a/tests/ui_and_conv/test_silent_auto_update.py b/tests/ui_and_conv/test_silent_auto_update.py index 490d1ea6..abe21389 100644 --- a/tests/ui_and_conv/test_silent_auto_update.py +++ b/tests/ui_and_conv/test_silent_auto_update.py @@ -56,11 +56,11 @@ async def fake_job(**kw): return UpdateResult.UPDATED monkeypatch.setattr(shell_module, "run_update_job", fake_job) - monkeypatch.setattr( - shell_module, - "read_update_status", - lambda: SimpleNamespace(message="updated", target_version="0.43.0"), - ) + + def forbidden_status_read(): + raise AssertionError("the current update result must not be inferred from shared status") + + monkeypatch.setattr(shell_module, "read_update_status", forbidden_status_read) await shell._silent_auto_update() @@ -80,7 +80,7 @@ async def test_silent_update_smoke_fail_toasts_verification_failed( monkeypatch.setattr(shell_module, "_detect_upgrade_command", lambda: ["pip"]) async def fake_job(**kw): - return UpdateResult.FAILED + return UpdateResult.VERIFICATION_FAILED monkeypatch.setattr(shell_module, "run_update_job", fake_job) monkeypatch.setattr( @@ -119,6 +119,7 @@ async def fake_job(**kw): ("result", "expected_marks"), [ (UpdateResult.FAILED, 0), + (UpdateResult.VERIFICATION_FAILED, 0), (UpdateResult.UP_TO_DATE, 1), (UpdateResult.UPDATED, 1), ], diff --git a/tests/ui_and_conv/test_update_orchestrator.py b/tests/ui_and_conv/test_update_orchestrator.py index 45b3bb55..1cdab372 100644 --- a/tests/ui_and_conv/test_update_orchestrator.py +++ b/tests/ui_and_conv/test_update_orchestrator.py @@ -183,11 +183,11 @@ async def fake_do_update( print_output=False, intent=update.UpdateIntent.INSTALL, source="test" ) - assert result is update.UpdateResult.FAILED + assert result is update.UpdateResult.VERIFICATION_FAILED status = orchestrator.read_update_status() assert status is not None assert status.state is orchestrator.UpdateJobState.FAILED - assert status.result == "FAILED" + assert status.result == "VERIFICATION_FAILED" assert "smoke check did not pass" in (status.message or "").lower() assert not orchestrator.UPDATE_LAST_SUCCESS_FILE.exists() From 52fe8ed3e7bd75c340c7d2aa57b2ae9fe08348c4 Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 22:07:54 -0400 Subject: [PATCH 7/8] fix(update): verify upgraded binary in smoke check, harden restart notice The post-install smoke check ran sys.executable for non-native installs; on Homebrew that exercises the still-running old keg, so it certified the pre-upgrade version right after an upgrade. Target the brew opt-linked launcher instead and assert the reported version matches the update target, reporting VERIFICATION_FAILED on mismatch. Derive the restart-to-apply notice from the recorded update-job status instead of the dismissal-filtered update cache, so dismissing a version's install prompt no longer hides the restart notice once that version is installed; ignore stale UPDATED statuses for versions not newer than the running one. --- CHANGELOG.md | 8 ++ src/pythinker_code/ui/shell/__init__.py | 23 +++-- .../ui/shell/update_orchestrator.py | 44 ++++++++- tests/ui_and_conv/test_silent_auto_update.py | 23 +++++ tests/ui_and_conv/test_update_orchestrator.py | 90 ++++++++++++++++++- 5 files changed, 178 insertions(+), 10 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1eb8ed98..b64dbe35 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,14 @@ GitHub Releases page; `0.8.0` is the new starting line. ## Unreleased +- **Post-update smoke check now verifies the upgraded binary and version.** On + Homebrew installs the smoke check exercised the still-running old keg via + `sys.executable`, so it could report "passed" with the pre-upgrade version; + it now targets the brew `opt`-linked launcher and fails (as + `VERIFICATION_FAILED`) when the reported version does not match the update + target. The persistent "restart to apply" notice is also derived from the + recorded update status alone, so dismissing a version's install prompt no + longer hides the restart notice after that version is installed. - **Updates never interrupt a running session.** On Windows, the background auto-updater previously launched the installer mid-session, force-closing the active Pythinker session. Updates are now downloaded and staged with a verified diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index da5a6f9f..ff514d82 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -82,6 +82,7 @@ read_windows_staged_update, refresh_update_cache_if_due, register_windows_staged_apply_on_exit, + semver_tuple, welcome_update_target, ) from pythinker_code.ui.shell.update_orchestrator import ( @@ -2273,16 +2274,26 @@ def _update_notice_text(self) -> str | None: return text def _compute_update_notice(self) -> str | None: - target = welcome_update_target() - if not target: - return None + from pythinker_code.constant import VERSION as current_version + # A release already installed this session needs a restart, not /update — - # surface that here instead of telling the user to re-run an update that - # has already landed. + # surface that instead of telling the user to re-run an update that has + # already landed. Checked against the recorded job status alone, NOT the + # dismissal-filtered update-available cache: dismissing a version's + # install prompt must not also hide the restart notice once that version + # is actually installed. status = read_update_status() - if update_restart_pending(status, target): + installed_target = status.target_version if status is not None else None + if ( + installed_target + and semver_tuple(installed_target) > semver_tuple(current_version) + and update_restart_pending(status, installed_target) + ): text = self._installed_update_restart_notice() else: + target = welcome_update_target() + if not target: + return None text = f"↑ Update available — v{target} · /update" if ascii_glyphs_enabled(): text = text.translate(_WELCOME_ASCII_FALLBACKS) diff --git a/src/pythinker_code/ui/shell/update_orchestrator.py b/src/pythinker_code/ui/shell/update_orchestrator.py index 12566f1e..a1c963f2 100644 --- a/src/pythinker_code/ui/shell/update_orchestrator.py +++ b/src/pythinker_code/ui/shell/update_orchestrator.py @@ -3,6 +3,7 @@ import contextlib import json import os +import re import subprocess import sys import time @@ -407,7 +408,9 @@ async def run_update_job( append_update_log(message) _write_last_success(job_id=job_id, message=message) else: - smoke_ok, smoke_message = run_post_install_smoke_check() + smoke_ok, smoke_message = run_post_install_smoke_check( + target_version=_read_target_version() + ) append_update_log(smoke_message) if smoke_ok: message = smoke_message @@ -487,9 +490,30 @@ def _smoke_check_command() -> list[str]: staged = staged_native_path() exe = str(staged) if staged.is_file() else sys.executable return [exe, "--version"] + brew_exe = _homebrew_linked_executable() + if brew_exe is not None: + return [str(brew_exe), "--version"] return [sys.executable, "-P", "-m", "pythinker_code", "--version"] +def _homebrew_linked_executable() -> Path | None: + """The stable brew-linked launcher for a Homebrew install, or None otherwise. + + `brew upgrade` installs the new keg side-by-side and repoints + ``/opt/pythinker-code``; the running interpreter still lives in the + OLD keg, so smoke-checking ``sys.executable`` would certify the pre-upgrade + install (and pass with the old version). Returns the opt-linked launcher + path even if it does not exist — a missing launcher after an upgrade is a + smoke-check failure, not a reason to fall back to the old binary. + """ + exe = sys.executable.replace("\\", "/") + marker = "/cellar/pythinker-code/" + idx = exe.lower().find(marker) + if idx < 0: + return None + return Path(exe[:idx]) / "opt" / "pythinker-code" / "bin" / "pythinker" + + def _finalize_native_staging(*, promote: bool) -> None: """After the smoke check, either arm the staged native binary for exit-time promotion (it ran) or discard it (it failed). No-op for non-native installs and @@ -524,7 +548,7 @@ def _smoke_check_env() -> dict[str, str]: return env -def run_post_install_smoke_check() -> tuple[bool, str]: +def run_post_install_smoke_check(target_version: str | None = None) -> tuple[bool, str]: command = _smoke_check_command() try: result = subprocess.run( @@ -547,7 +571,21 @@ def run_post_install_smoke_check() -> tuple[bool, str]: return False, f"Smoke check failed: {detail}" if not output or not any(ch.isdigit() for ch in output): return False, "Smoke check did not report a version." - return True, f"Smoke check passed: {output.splitlines()[0]}" + first_line = output.splitlines()[0] + if target_version is not None: + from pythinker_code.ui.shell.update import semver_tuple + + reported = re.search(r"\d+\.\d+\.\d+", first_line) + if reported is None: + return False, f"Smoke check did not report a parseable version: {first_line}" + if semver_tuple(reported.group(0)) != semver_tuple(target_version): + # The upgraded binary must identify as the target release; matching + # the OLD version means the check exercised the pre-upgrade install + # (or the upgrade silently no-oped). + return False, ( + f"Smoke check reported {reported.group(0)}, expected {target_version}: {first_line}" + ) + return True, f"Smoke check passed: {first_line}" async def prompt_pre_start_update_job() -> None: diff --git a/tests/ui_and_conv/test_silent_auto_update.py b/tests/ui_and_conv/test_silent_auto_update.py index abe21389..bc767763 100644 --- a/tests/ui_and_conv/test_silent_auto_update.py +++ b/tests/ui_and_conv/test_silent_auto_update.py @@ -428,6 +428,29 @@ def test_update_notice_previous_process_success_falls_back(runtime, tmp_path, mo def test_update_notice_none_when_up_to_date(runtime, tmp_path, monkeypatch): shell = _make_shell(runtime, tmp_path) monkeypatch.setattr(shell_module, "welcome_update_target", lambda: None) + monkeypatch.setattr(shell_module, "read_update_status", lambda: None) + assert shell._compute_update_notice() is None + + +def test_update_notice_restart_survives_dismissed_version(runtime, tmp_path, monkeypatch): + """Regression: the restart notice must come from the recorded job status, not + the dismissal-filtered cache. Dismissing a version's install prompt (which + nulls welcome_update_target) must not hide 'restart to apply' after that + version has actually been installed this session.""" + shell = _make_shell(runtime, tmp_path) + monkeypatch.setattr(shell_module, "welcome_update_target", lambda: None) + monkeypatch.setattr(shell_module, "ascii_glyphs_enabled", lambda: False) + monkeypatch.setattr(shell_module, "read_update_status", lambda: _updated_status("9.9.9")) + text = shell._compute_update_notice() + assert text is not None and "Restart" in text and "9.9.9" in text + + +def test_update_notice_ignores_stale_status_for_older_version(runtime, tmp_path, monkeypatch): + # A leftover UPDATED status for a version we are already running (or older) + # must not claim a restart is pending. + shell = _make_shell(runtime, tmp_path) + monkeypatch.setattr(shell_module, "welcome_update_target", lambda: None) + monkeypatch.setattr(shell_module, "read_update_status", lambda: _updated_status("0.0.1")) assert shell._compute_update_notice() is None diff --git a/tests/ui_and_conv/test_update_orchestrator.py b/tests/ui_and_conv/test_update_orchestrator.py index 1cdab372..c578fbe8 100644 --- a/tests/ui_and_conv/test_update_orchestrator.py +++ b/tests/ui_and_conv/test_update_orchestrator.py @@ -176,7 +176,7 @@ async def fake_do_update( monkeypatch.setattr( orchestrator, "run_post_install_smoke_check", - lambda: (False, "Smoke check failed: broken"), + lambda **_kw: (False, "Smoke check failed: broken"), ) result = await orchestrator.run_update_job( @@ -368,3 +368,91 @@ def fake_run(command, **kwargs): assert ok is False assert "broken" in message + + +def test_brew_smoke_check_targets_opt_linked_launcher(monkeypatch): + # Regression: a brew upgrade installs the new keg side-by-side while the + # running interpreter stays in the OLD keg. Smoke-checking sys.executable + # certified the pre-upgrade install ("Smoke check passed: ... 0.58.0" right + # after installing 0.59.0). The check must target the brew opt link, which + # the upgrade repoints at the new keg. + monkeypatch.setattr(orchestrator, "is_native_build", lambda: False) + monkeypatch.setattr( + orchestrator.sys, + "executable", + "/opt/homebrew/Cellar/pythinker-code/0.58.0/libexec/bin/python", + ) + + assert orchestrator._smoke_check_command() == [ + "/opt/homebrew/opt/pythinker-code/bin/pythinker", + "--version", + ] + + +def test_brew_launcher_detection_ignores_non_brew_installs(monkeypatch): + monkeypatch.setattr(orchestrator.sys, "executable", "/tmp/venv/bin/python") + assert orchestrator._homebrew_linked_executable() is None + + +def _fake_version_run(monkeypatch, output: str) -> None: + monkeypatch.setattr(orchestrator, "_smoke_check_command", lambda: ["pythinker", "--version"]) + monkeypatch.setattr( + orchestrator.subprocess, + "run", + lambda command, **kwargs: SimpleNamespace(returncode=0, stdout=output, stderr=""), + ) + + +def test_smoke_check_fails_when_reported_version_is_not_target(monkeypatch): + _fake_version_run(monkeypatch, "pythinker, version 0.58.0\n") + + ok, message = orchestrator.run_post_install_smoke_check(target_version="0.59.0") + + assert ok is False + assert "0.58.0" in message and "0.59.0" in message + + +def test_smoke_check_passes_when_reported_version_matches_target(monkeypatch): + _fake_version_run(monkeypatch, "pythinker, version 0.59.0\n") + + ok, message = orchestrator.run_post_install_smoke_check(target_version="0.59.0") + + assert ok is True + assert message.startswith("Smoke check passed:") + + +def test_smoke_check_without_target_keeps_lenient_version_probe(monkeypatch): + # No recorded target (e.g. missing latest-version cache) keeps the original + # "any version string" behavior rather than failing every update. + _fake_version_run(monkeypatch, "pythinker, version 1.2.3\n") + + ok, _message = orchestrator.run_post_install_smoke_check() + + assert ok is True + + +@pytest.mark.asyncio +async def test_update_job_passes_target_version_to_smoke_check(monkeypatch, tmp_path): + _isolate_update_files(monkeypatch, tmp_path) + monkeypatch.setattr(orchestrator, "_read_target_version", lambda: "9.9.9") + + async def fake_do_update( + *, print_output: bool, intent: update.UpdateIntent, output_callback=None + ): + return update.UpdateResult.UPDATED + + monkeypatch.setattr(update, "do_update", fake_do_update) + seen: list[str | None] = [] + + def fake_smoke(target_version=None): + seen.append(target_version) + return True, "Smoke check passed: pythinker, version 9.9.9" + + monkeypatch.setattr(orchestrator, "run_post_install_smoke_check", fake_smoke) + + result = await orchestrator.run_update_job( + print_output=False, intent=update.UpdateIntent.INSTALL, source="test" + ) + + assert result is update.UpdateResult.UPDATED + assert seen == ["9.9.9"] From 1268ef6fc6eea9bad54b0f7b0bbf04d5bcafc7c5 Mon Sep 17 00:00:00 2001 From: elkaix Date: Fri, 17 Jul 2026 22:34:29 -0400 Subject: [PATCH 8/8] test(update): assert observable outcomes instead of internal call shapes Drop the private reader-signature wrapper in the startup staged-apply test (claim behavior is covered by the dedicated concurrency tests) and replace status-read mocks in the silent auto-update tests with conflicting or absent status records, so the returned update result is the test oracle. --- tests/ui/test_update_staging.py | 7 ------ tests/ui_and_conv/test_silent_auto_update.py | 26 +++++++++++--------- 2 files changed, 14 insertions(+), 19 deletions(-) diff --git a/tests/ui/test_update_staging.py b/tests/ui/test_update_staging.py index 239b746d..00534636 100644 --- a/tests/ui/test_update_staging.py +++ b/tests/ui/test_update_staging.py @@ -198,13 +198,6 @@ def test_apply_before_start_applies_valid_stage(staging: Path, monkeypatch): monkeypatch.setattr(upd, "_is_running_from_source_checkout", lambda: False) monkeypatch.setattr("pythinker_code.constant.VERSION", "0.1.0") monkeypatch.setattr(upd, "_spawn_detached_windows_installer", lambda p: True) - read_staged_update = upd.read_windows_staged_update - - def read_claimed_update_only(manifest_path: Path | None = None): - assert manifest_path is not None - return read_staged_update(manifest_path) - - monkeypatch.setattr(upd, "read_windows_staged_update", read_claimed_update_only) assert upd.apply_staged_update_before_start() is True diff --git a/tests/ui_and_conv/test_silent_auto_update.py b/tests/ui_and_conv/test_silent_auto_update.py index bc767763..3a4a4609 100644 --- a/tests/ui_and_conv/test_silent_auto_update.py +++ b/tests/ui_and_conv/test_silent_auto_update.py @@ -57,10 +57,17 @@ async def fake_job(**kw): monkeypatch.setattr(shell_module, "run_update_job", fake_job) - def forbidden_status_read(): - raise AssertionError("the current update result must not be inferred from shared status") - - monkeypatch.setattr(shell_module, "read_update_status", forbidden_status_read) + # Conflicting shared status: if the outcome were inferred from the status + # file instead of the job's returned result, this smoke-failed record would + # flip the flow to the verification-failure toast. + monkeypatch.setattr( + shell_module, + "read_update_status", + lambda: SimpleNamespace( + message=shell_module.SMOKE_CHECK_FAILED_PREFIX + "boom", + target_version="0.43.0", + ), + ) await shell._silent_auto_update() @@ -83,14 +90,9 @@ async def fake_job(**kw): return UpdateResult.VERIFICATION_FAILED monkeypatch.setattr(shell_module, "run_update_job", fake_job) - monkeypatch.setattr( - shell_module, - "read_update_status", - lambda: SimpleNamespace( - message="Updated, but smoke check did not pass: boom", - target_version="0.43.0", - ), - ) + # Contradicting shared status (no record at all): the toast must be driven + # by the job's VERIFICATION_FAILED result, not inferred from the status file. + monkeypatch.setattr(shell_module, "read_update_status", lambda: None) await shell._silent_auto_update()