Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 3 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -84,11 +84,13 @@ Keeps the model resident so each subsequent call is instant:

```bash
stackvox serve # foreground; run with `nohup stackvox serve &` to background
stackvox status # is the daemon up?
stackvox status # is the daemon up? also shows version + any pending PyPI update
stackvox say "Hello" # send text to the daemon (fails if not running)
stackvox stop # graceful shutdown
```

stackvox checks PyPI for newer versions but only at two moments — when you run `stackvox status` and at daemon startup. The script-heavy paths (`say`, `speak`, `stackvox-say`, hooks, CI) never make a network call. To see notices on every invocation set `STACKVOX_UPDATE_NOTICE=1`. To disable the check entirely set `STACKVOX_NO_UPDATE_CHECK=1`. The check is auto-skipped when common CI env vars (`CI`, `GITHUB_ACTIONS`, etc.) are set so build logs stay clean.

## `stackvox-say` (bash helper, ~13ms)

When you want minimum latency from shell scripts (hooks, CI steps, etc.), skip the Python client and use the bash helper — it talks directly to the daemon's unix socket via `nc`:
Expand Down
35 changes: 31 additions & 4 deletions stackvox/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,13 @@

import argparse
import logging
import os
import sys
from pathlib import Path

import soundfile as sf

from stackvox import config, daemon
from stackvox import config, daemon, updates
from stackvox.engine import Stackvox


Expand Down Expand Up @@ -226,9 +227,18 @@ def _cmd_stop(_: argparse.Namespace) -> int:
def _cmd_status(_: argparse.Namespace) -> int:
if daemon.is_running():
print(f"running (pid {daemon.PID_PATH.read_text().strip()}) on {daemon.SOCKET_PATH}")
return 0
print("stopped")
return 1
rc = 0
else:
print("stopped")
rc = 1
# Status is the canonical "is everything OK?" query — surface update info
# here regardless of running/stopped. Synchronous fetch with 2s timeout.
info = updates.check_for_update()
if info is not None:
print(f"version: {info.current} ({updates.format_notice(info)})")
else:
print(f"version: {updates._current_version()}")
return rc


def _cmd_voices(args: argparse.Namespace) -> int:
Expand Down Expand Up @@ -289,6 +299,21 @@ def _cmd_install_helper(args: argparse.Namespace) -> int:
return 0


def _maybe_print_update_notice() -> None:
"""Opt-in stderr notice for the script-friendly default-off case.

Off by default (most stackvox invocations are non-interactive — hooks,
CI, scripts — and pollution there is worse than the missed notice).
Set `STACKVOX_UPDATE_NOTICE=1` to turn it on. Reads cache only; never
fetches from PyPI on this path.
"""
if not os.environ.get("STACKVOX_UPDATE_NOTICE"):
return
info = updates.cached_update()
if info is not None:
print(f"[stackvox] {updates.format_notice(info)}", file=sys.stderr)


def main() -> int:
_configure_logging()
argv = sys.argv[1:]
Expand All @@ -299,6 +324,8 @@ def main() -> int:
elif not argv and not sys.stdin.isatty():
argv = ["speak"]

_maybe_print_update_notice()

parser = _build_parser(config.load_defaults())
args = parser.parse_args(argv)

Expand Down
18 changes: 18 additions & 0 deletions stackvox/daemon.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@

import sounddevice as sd

from stackvox import updates
from stackvox.engine import DEFAULT_LANG, DEFAULT_SPEED, DEFAULT_VOICE, Stackvox
from stackvox.paths import pid_path, socket_path

Expand Down Expand Up @@ -260,6 +261,22 @@ def is_running() -> bool:
return _pid_alive(pid)


def _check_for_update_async() -> None:
"""Spawn a background thread that checks PyPI for an update and logs it.

Runs at daemon startup, off the critical path. The user is at a terminal
when they `stackvox serve`, so the daemon's stderr is visible — that's
the highest-leverage moment to surface "you should upgrade".
"""

def _worker() -> None:
info = updates.check_for_update()
if info is not None:
logger.info(updates.format_notice(info))

threading.Thread(target=_worker, daemon=True, name="update-check").start()


def serve(voice: str = DEFAULT_VOICE, speed: float = DEFAULT_SPEED, lang: str = DEFAULT_LANG) -> None:
if is_running():
raise RuntimeError(f"daemon already running (pid {PID_PATH.read_text().strip()})")
Expand All @@ -273,6 +290,7 @@ def serve(voice: str = DEFAULT_VOICE, speed: float = DEFAULT_SPEED, lang: str =
server.state = state # type: ignore[attr-defined]

PID_PATH.write_text(str(os.getpid()))
_check_for_update_async()

def handle_signal(signum: int, frame: object) -> None:
state.shutdown()
Expand Down
184 changes: 184 additions & 0 deletions stackvox/updates.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,184 @@
"""Best-effort PyPI update check.

Surfaces "a newer stackvox is on PyPI" without polluting the script paths
that make up most of stackvox's invocations. The actual fetch is opt-in
(only `stackvox status` and the daemon's startup trigger it); everywhere
else reads from a 24h cache.

Cache schema at `~/.cache/stackvox/update-check.json`::

{"checked_at": "2026-04-30T13:42:00+00:00", "latest": "0.4.0"}

Disable entirely with `STACKVOX_NO_UPDATE_CHECK=1`. Auto-skipped when any
common CI env var is set so build logs stay clean.
"""

from __future__ import annotations

import json
import logging
import os
import urllib.error
import urllib.request
from dataclasses import dataclass
from datetime import datetime, timedelta, timezone
from importlib.metadata import PackageNotFoundError
from importlib.metadata import version as _pkg_version
from pathlib import Path

from stackvox.paths import cache_dir

logger = logging.getLogger(__name__)


def _current_version() -> str:
"""Read our own installed version. Late-bound so importing this module
early in the package init chain doesn't trip a circular import."""
try:
return _pkg_version("stackvox")
except PackageNotFoundError:
return "0.0.0+unknown"


PYPI_JSON_URL = "https://pypi.org/pypi/stackvox/json"
CACHE_TTL = timedelta(hours=24)
FETCH_TIMEOUT_SECONDS = 2.0

# Env vars set in common CI environments. Presence of any disables the check.
_CI_ENV_VARS = ("CI", "GITHUB_ACTIONS", "BUILDKITE", "CIRCLECI", "GITLAB_CI", "TRAVIS")


@dataclass(frozen=True)
class UpdateInfo:
current: str
latest: str

@property
def is_outdated(self) -> bool:
return _is_newer(self.latest, self.current)


def cache_path() -> Path:
return cache_dir() / "update-check.json"


def is_disabled() -> bool:
"""Whether the update check should be skipped at all."""
if os.environ.get("STACKVOX_NO_UPDATE_CHECK"):
return True
return any(os.environ.get(v) for v in _CI_ENV_VARS)


def _is_newer(latest: str, current: str) -> bool:
"""Compare two PEP-440-ish dotted version strings.

Handles the X.Y.Z and X.Y.Z+suffix forms stackvox actually publishes.
Non-integer segments (e.g. an `rc1` tail) sort below numeric ones so
`0.4.0` > `0.4.0rc1`, but two distinct pre-release labels compare
equal — fine for stackvox since we don't ship alphas/betas.
"""

def _key(v: str) -> tuple[int, ...]:
head = v.split("+", 1)[0] # drop +local
parts: list[int] = []
for piece in head.split("."):
try:
parts.append(int(piece))
except ValueError:
parts.append(-1)
return tuple(parts)

return _key(latest) > _key(current)


def fetch_latest_version(timeout: float = FETCH_TIMEOUT_SECONDS) -> str | None:
"""Hit PyPI's JSON API for the latest stackvox version.

Returns None on any network or parse error — this is best-effort.
"""
if is_disabled():
return None
try:
ua = f"stackvox/{_current_version()} update-check"
req = urllib.request.Request(PYPI_JSON_URL, headers={"User-Agent": ua})
with urllib.request.urlopen(req, timeout=timeout) as resp: # noqa: S310 - constant URL
payload = json.loads(resp.read().decode("utf-8"))
version = payload.get("info", {}).get("version")
if isinstance(version, str):
return version
return None
except (urllib.error.URLError, TimeoutError, OSError, ValueError) as exc:
logger.debug("update check failed: %s", exc)
return None


def write_cache(latest: str, *, now: datetime | None = None) -> None:
"""Persist the most recent successful check."""
path = cache_path()
path.parent.mkdir(parents=True, exist_ok=True)
payload = {
"checked_at": (now or datetime.now(timezone.utc)).isoformat(),
"latest": latest,
}
path.write_text(json.dumps(payload), encoding="utf-8")


def read_cache() -> tuple[datetime, str] | None:
"""Return (timestamp, latest) from the cache file, or None if absent/broken."""
path = cache_path()
if not path.is_file():
return None
try:
payload = json.loads(path.read_text(encoding="utf-8"))
when = datetime.fromisoformat(payload["checked_at"])
latest = payload["latest"]
if not isinstance(latest, str):
return None
return when, latest
except (OSError, ValueError, KeyError) as exc:
logger.debug("ignoring malformed update-check cache: %s", exc)
return None


def cached_update(*, now: datetime | None = None) -> UpdateInfo | None:
"""Read the cache and return UpdateInfo for an unmet upgrade, else None."""
if is_disabled():
return None
entry = read_cache()
if entry is None:
return None
_checked_at, latest = entry
info = UpdateInfo(current=_current_version(), latest=latest)
return info if info.is_outdated else None


def check_for_update(*, now: datetime | None = None) -> UpdateInfo | None:
"""Fetch from PyPI if the cache is stale, then return any pending update.

Synchronous; safe to call from the foreground only when ~2s of network
latency is acceptable (e.g. `stackvox status`). For the daemon startup
path, call this from a background thread.
"""
if is_disabled():
return None
now = now or datetime.now(timezone.utc)
entry = read_cache()
if entry is None or (now - entry[0]) > CACHE_TTL:
latest = fetch_latest_version()
if latest is not None:
write_cache(latest, now=now)
else:
# On fetch failure, fall back to whatever's already cached (which
# may be None, in which case we just give up silently).
if entry is None:
return None
latest = entry[1]
else:
latest = entry[1]
info = UpdateInfo(current=_current_version(), latest=latest)
return info if info.is_outdated else None


def format_notice(info: UpdateInfo) -> str:
"""Single-line user-facing message describing the available upgrade."""
return f"update available: {info.current} → {info.latest} (run `pipx upgrade stackvox`)"
63 changes: 63 additions & 0 deletions tests/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,17 @@ def _stdin_is_a_tty(mocker):
mocker.patch.object(cli.sys.stdin, "isatty", return_value=True)


@pytest.fixture(autouse=True)
def _no_network_update_check(mocker):
"""Block the PyPI update check from making real HTTP requests in tests.

Tests that care about update-notice behaviour opt in by patching
`cli.updates.cached_update` / `cli.updates.check_for_update` themselves.
"""
mocker.patch.object(cli.updates, "check_for_update", return_value=None)
mocker.patch.object(cli.updates, "cached_update", return_value=None)


def test_bare_text_routes_to_speak(mocker):
speak = mocker.patch.object(cli, "_cmd_speak", return_value=0)
mocker.patch.object(cli.sys, "argv", ["stackvox", "hello world"])
Expand Down Expand Up @@ -223,6 +234,58 @@ def test_stopped_returns_one(self, mocker, capsys):
assert rc == 1
assert "stopped" in capsys.readouterr().out

def test_status_prints_update_notice_when_outdated(self, mocker, capsys):
mocker.patch.object(cli.daemon, "is_running", return_value=False)
mocker.patch.object(
cli.updates,
"check_for_update",
return_value=cli.updates.UpdateInfo(current="0.3.1", latest="0.4.0"),
)
cli._cmd_status(_ns())
out = capsys.readouterr().out
assert "0.3.1" in out
assert "0.4.0" in out
assert "pipx upgrade" in out

def test_status_prints_plain_version_when_up_to_date(self, mocker, capsys):
mocker.patch.object(cli.daemon, "is_running", return_value=False)
mocker.patch.object(cli.updates, "check_for_update", return_value=None)
mocker.patch.object(cli.updates, "_current_version", return_value="0.4.0")
cli._cmd_status(_ns())
out = capsys.readouterr().out
assert "version: 0.4.0" in out
assert "pipx upgrade" not in out


class TestUpdateNotice:
def test_silent_by_default(self, mocker, monkeypatch, capsys):
monkeypatch.delenv("STACKVOX_UPDATE_NOTICE", raising=False)
mocker.patch.object(
cli.updates,
"cached_update",
return_value=cli.updates.UpdateInfo(current="0.3.1", latest="0.4.0"),
)
cli._maybe_print_update_notice()
assert capsys.readouterr().err == ""

def test_prints_to_stderr_when_opted_in(self, mocker, monkeypatch, capsys):
monkeypatch.setenv("STACKVOX_UPDATE_NOTICE", "1")
mocker.patch.object(
cli.updates,
"cached_update",
return_value=cli.updates.UpdateInfo(current="0.3.1", latest="0.4.0"),
)
cli._maybe_print_update_notice()
err = capsys.readouterr().err
assert "0.3.1" in err
assert "0.4.0" in err

def test_silent_when_no_update_even_with_opt_in(self, mocker, monkeypatch, capsys):
monkeypatch.setenv("STACKVOX_UPDATE_NOTICE", "1")
mocker.patch.object(cli.updates, "cached_update", return_value=None)
cli._maybe_print_update_notice()
assert capsys.readouterr().err == ""


class TestCmdVoices:
def test_prints_one_voice_per_line(self, fake_stackvox, capsys):
Expand Down
Loading