mirror of
https://github.com/NousResearch/hermes-agent.git
synced 2026-11-06 05:17:15 +00:00
perf(onboarding): the due answer is worked out at serve boot and served from a cache
The desktop asks onboarding.state at launch. The handler recomputed everything inline, on the connection's serial queue, while boot threads did the same provider reads. A slow answer missed the desktop's deadline, and onboarding was skipped for that launch. - hermes_cli/onboarding_due.py keeps the answer (run, eligible). A reader waits a bounded time (8 s) for a boot computation in flight, then answers live and caches that. - The serve lifespan starts the computation on the desktop's local backend (start_due_boot), after the setup bootstrap has recorded a fresh install and kept an existing guest's trial. - Every writer that can move the answer drops the cache: set_run, the OS user's marker, and the setup record's refresh and reconcile. A stamp of the files the answer reads catches edits made outside the process. - onboarding.state runs on the RPC pool (_LONG_HANDLERS), so its wait never holds the connection's queue. Remote backends answer eligible: false at once, and --fresh-install still asks live.
This commit is contained in:
1 parent
32ef7d6968
commit
fbc361ef57
10 files changed
+224
-15
No files matched your search
@@ -110,6 +110,7 @@ def reconcile_record() -> Optional[SetupRecord]:
|
||||
if _record is not record: # a retry replaced it meanwhile; its inventory is newer
|
||||
return _record
|
||||
_record = refreshed
|
||||
_due_changed()
|
||||
_broadcast(refreshed)
|
||||
return refreshed
|
||||
|
||||
@@ -227,11 +228,18 @@ def refresh_record(*, announce: bool = True) -> Optional[SetupRecord]:
|
||||
record = _build_record(other=_inventory_other_providers())
|
||||
with _lock:
|
||||
_record = record
|
||||
_due_changed()
|
||||
if announce:
|
||||
_broadcast(record)
|
||||
return record
|
||||
|
||||
|
||||
def _due_changed() -> None:
|
||||
from hermes_cli import onboarding_due
|
||||
|
||||
onboarding_due.invalidate()
|
||||
|
||||
|
||||
def _broadcast(record: SetupRecord) -> None:
|
||||
try:
|
||||
from tui_gateway.server import _broadcast_global_event
|
||||
|
||||
@@ -95,7 +95,7 @@ def probe() -> InstallState:
|
||||
for root in candidate_roots(places):
|
||||
if finding := root_evidence(root, places):
|
||||
if finding.evidence is not Evidence.UNREADABLE: # doubt blocks the trial, never marks
|
||||
mark_machine_used()
|
||||
_write_marker({}, answer_moved=False) # this probe is the answer; nothing to recompute
|
||||
return InstallState((finding,))
|
||||
return InstallState(())
|
||||
|
||||
@@ -124,7 +124,7 @@ def trial_reserved() -> bool:
|
||||
return any(_read_marker(marker) == _TRIAL_RESERVED for marker in machine_places().markers)
|
||||
|
||||
|
||||
def _write_marker(content: dict[str, str]) -> None:
|
||||
def _write_marker(content: dict[str, str], *, answer_moved: bool = True) -> None:
|
||||
places = machine_places()
|
||||
if places.markers:
|
||||
from utils import atomic_json_write
|
||||
@@ -133,6 +133,10 @@ def _write_marker(content: dict[str, str]) -> None:
|
||||
marker.parent.mkdir(parents=True, exist_ok=True)
|
||||
atomic_json_write(marker, content) # unique temp name: two backends may mark at once
|
||||
probe.cache_clear()
|
||||
if answer_moved:
|
||||
from hermes_cli import onboarding_due
|
||||
|
||||
onboarding_due.invalidate()
|
||||
|
||||
|
||||
def _read_marker(marker: Path) -> Any:
|
||||
|
||||
@@ -0,0 +1,87 @@
|
||||
"""The desktop's boot question, answered once: is onboarding due, and can something carry it.
|
||||
|
||||
``onboarding.state`` asks it on every launch, and a cold first boot answering it live competes
|
||||
with every other boot thread for the same provider reads. The serve boot works it out on its own
|
||||
thread (:func:`begin_boot`), and ``onboarding.state`` reads the cached answer, waiting a bounded
|
||||
time for a boot computation still in flight. Every writer that can move the answer calls
|
||||
:func:`invalidate`: the root ``onboarding.run`` (``onboarding_run.set_run``), the OS user's marker
|
||||
(``install_state``) and the setup record (``free_tier_bootstrap``). The next reader then computes
|
||||
live once and caches that. A *stamp* the caller passes (the files the answer reads) catches edits
|
||||
no writer in this process saw, such as a key typed into ``.env`` by hand. The computation itself
|
||||
lives with its caller (``tui_gateway``), which owns the default-profile scope it runs in.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from collections.abc import Callable, Hashable
|
||||
|
||||
# The desktop's first question waits this long for the boot computation before answering live.
|
||||
BOOT_WAIT_SECONDS = 8.0
|
||||
|
||||
Due = dict[str, bool]
|
||||
Stamp = Callable[[], Hashable]
|
||||
|
||||
_lock = threading.Lock()
|
||||
_value: Due | None = None
|
||||
_value_stamp: Hashable = None
|
||||
_generation = 0
|
||||
_boot_done: threading.Event | None = None
|
||||
|
||||
|
||||
def begin_boot(stamp: Stamp = lambda: None) -> Callable[[Callable[[], Due]], None]:
|
||||
"""Mark a boot computation as in flight (readers wait for it from now on) and return the call
|
||||
that finishes it: ``finish(compute)`` runs *compute* and caches its answer, unless a writer
|
||||
invalidated meanwhile."""
|
||||
global _boot_done
|
||||
done = threading.Event()
|
||||
with _lock:
|
||||
_boot_done = done
|
||||
|
||||
def finish(compute: Callable[[], Due]) -> None:
|
||||
try:
|
||||
with _lock:
|
||||
generation = _generation
|
||||
value = compute()
|
||||
_store(value, generation, stamp()) # after: computing can itself write a config skeleton
|
||||
finally:
|
||||
done.set()
|
||||
|
||||
return finish
|
||||
|
||||
|
||||
def read(compute: Callable[[], Due], *, stamp: Stamp = lambda: None, wait: float = BOOT_WAIT_SECONDS) -> Due:
|
||||
"""The cached answer; a boot computation in flight is waited for (bounded), and with nothing
|
||||
cached, or files moved since (*stamp*), *compute* answers live and is cached."""
|
||||
done = _boot_done
|
||||
if done is not None:
|
||||
done.wait(wait)
|
||||
seen = stamp()
|
||||
with _lock:
|
||||
if _value is not None and _value_stamp == seen:
|
||||
return dict(_value)
|
||||
generation = _generation
|
||||
value = compute()
|
||||
_store(value, generation, stamp())
|
||||
return dict(value)
|
||||
|
||||
|
||||
def invalidate() -> None:
|
||||
global _value, _generation
|
||||
with _lock:
|
||||
_value = None
|
||||
_generation += 1
|
||||
|
||||
|
||||
def reset_for_tests() -> None:
|
||||
global _boot_done
|
||||
invalidate()
|
||||
with _lock:
|
||||
_boot_done = None
|
||||
|
||||
|
||||
def _store(value: Due, generation: int, seen: Hashable) -> None:
|
||||
global _value, _value_stamp
|
||||
with _lock:
|
||||
if generation == _generation: # a writer that landed meanwhile wins over this computation
|
||||
_value, _value_stamp = dict(value), seen
|
||||
@@ -65,3 +65,6 @@ def set_run(value: bool, *, mark_profile_offered: bool = False) -> None:
|
||||
if mark_profile_offered:
|
||||
config_mod._ensure_dict(section, "seen")[PROFILE_BUILD_FLAG] = True
|
||||
config_mod.atomic_config_write(path, config)
|
||||
from hermes_cli import onboarding_due
|
||||
|
||||
onboarding_due.invalidate()
|
||||
@@ -318,6 +318,9 @@ async def _lifespan(app: FastAPI):
|
||||
from hermes_cli.free_tier_bootstrap import start_background_bootstrap
|
||||
|
||||
start_background_bootstrap()
|
||||
# The desktop's onboarding due answer, worked out after that bootstrap and before it is asked.
|
||||
# Through the server: methods_* functions run with the server's globals (``bind_module``).
|
||||
tui_gateway.server.start_due_boot()
|
||||
|
||||
try:
|
||||
yield
|
||||
|
||||
@@ -317,6 +317,9 @@ def _hermetic_environment(tmp_path, tmp_path_factory, monkeypatch):
|
||||
sources=(install_state._active_root,))
|
||||
monkeypatch.setattr(install_state, "machine_places", lambda: isolated_places)
|
||||
install_state.probe.cache_clear()
|
||||
from hermes_cli import onboarding_due
|
||||
|
||||
onboarding_due.reset_for_tests()
|
||||
# Per-TEST host-rendezvous dir (see the session-level block at the top): the
|
||||
# host gateway/serve record is shared per OS user by design, so without this
|
||||
# one test's published owner makes the next test's lifecycle code attach to it.
|
||||
|
||||
@@ -0,0 +1,39 @@
|
||||
"""The boot-time onboarding due answer (``hermes_cli.onboarding_due``): a reader during the boot
|
||||
computation waits for it, and a writer that moves the answer mid-computation is never lost."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from hermes_cli import onboarding_due
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _fresh_cache():
|
||||
onboarding_due.reset_for_tests()
|
||||
yield
|
||||
onboarding_due.reset_for_tests()
|
||||
|
||||
|
||||
def test_a_reader_during_the_boot_computation_waits_for_it():
|
||||
finish = onboarding_due.begin_boot()
|
||||
answers: list[dict] = []
|
||||
reader = threading.Thread(target=lambda: answers.append(onboarding_due.read(lambda: {"live": True})))
|
||||
reader.start()
|
||||
finish(lambda: {"run": True, "eligible": True})
|
||||
reader.join(5)
|
||||
|
||||
assert answers == [{"run": True, "eligible": True}]
|
||||
|
||||
|
||||
def test_a_write_during_the_boot_computation_is_not_lost():
|
||||
finish = onboarding_due.begin_boot()
|
||||
|
||||
def compute():
|
||||
onboarding_due.invalidate() # set_run / a claim landed while the boot was computing
|
||||
return {"run": True, "eligible": True}
|
||||
finish(compute)
|
||||
|
||||
assert onboarding_due.read(lambda: {"run": False, "eligible": True}) == {"run": False, "eligible": True}
|
||||
@@ -4,6 +4,7 @@ chat, and only the desktop's local backend opens onboarding by itself."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import threading
|
||||
from pathlib import Path
|
||||
|
||||
import httpx
|
||||
@@ -93,14 +94,14 @@ def test_a_crash_right_after_the_claim_keeps_onboarding_due(portal, monkeypatch)
|
||||
assert portal.creates() == 1
|
||||
|
||||
|
||||
def test_only_the_local_desktop_backend_opens_onboarding_by_itself(portal, monkeypatch):
|
||||
def test_only_the_local_desktop_backend_opens_onboarding_by_itself(portal, root, monkeypatch):
|
||||
assert _call("onboarding.state")["result"]["eligible"] is False # CLI / SSH / URL backend
|
||||
|
||||
as_local_desktop_backend(monkeypatch)
|
||||
install_state.mark_machine_used()
|
||||
assert _call("onboarding.state")["result"] == {"run": False, "eligible": False, "session_id": None, "run_id": None}
|
||||
|
||||
monkeypatch.setenv("OPENROUTER_API_KEY", "sk-or-test")
|
||||
(root / ".env").write_text("OPENROUTER_API_KEY=sk-or-test\n", encoding="utf-8") # a key saved later
|
||||
assert _call("onboarding.state")["result"]["eligible"] is True
|
||||
|
||||
|
||||
@@ -135,3 +136,19 @@ def test_the_dev_fresh_install_switch_treats_a_used_machine_as_new(portal, monke
|
||||
assert _call("onboarding.state", {"assume_fresh": True})["result"]["run"] is True
|
||||
assert "result" in _call("onboarding.start", {"assume_fresh": True})
|
||||
assert portal.creates() == 1
|
||||
|
||||
|
||||
def test_the_due_answer_is_worked_out_at_boot_and_served_from_there(portal, monkeypatch):
|
||||
from hermes_cli import auth, free_tier_bootstrap
|
||||
|
||||
as_local_desktop_backend(monkeypatch)
|
||||
free_tier_bootstrap.run_bootstrap(announce=False)
|
||||
real, callers = auth.resolve_provider, []
|
||||
monkeypatch.setattr(auth, "resolve_provider",
|
||||
lambda *a, **kw: (callers.append(threading.current_thread().name), real(*a, **kw))[1])
|
||||
server.start_due_boot()
|
||||
|
||||
answers = [_call("onboarding.state")["result"] for _ in range(3)]
|
||||
|
||||
assert answers[0]["run"] is True and answers[0]["eligible"] is True and answers == [answers[0]] * 3
|
||||
assert callers and set(callers) == {"onboarding-due"} # the requests themselves never walk the ladder
|
||||
@@ -26,21 +26,66 @@ def _(rid, params: dict) -> dict:
|
||||
|
||||
|
||||
def _run_state(*, assume_fresh: bool = False) -> dict:
|
||||
"""``run``: the root config says so, else this machine never used Hermes (or the unpackaged
|
||||
"""``run``: the root config says so, else this OS user never used Hermes (or the unpackaged
|
||||
desktop's ``--fresh-install`` says to assume that). ``eligible``: this is the desktop's local
|
||||
backend and something can carry the first chat (the welcome trial or a configured provider);
|
||||
remote backends run onboarding only from Settings."""
|
||||
remote backends answer at once and run onboarding only from Settings. On the local backend the
|
||||
answer comes from the boot computation (``onboarding_due``); ``--fresh-install`` asks live."""
|
||||
from hermes_cli import onboarding_due
|
||||
from hermes_cli.anon_auth import on_local_desktop_backend
|
||||
from hermes_cli.onboarding_run import should_run
|
||||
|
||||
if not on_local_desktop_backend():
|
||||
return {"run": should_run(assume_fresh=assume_fresh), "eligible": False, **_active_run()}
|
||||
due = _due_now(assume_fresh=True) if assume_fresh else onboarding_due.read(_due_now, stamp=_due_stamp)
|
||||
return {**due, **_active_run()}
|
||||
|
||||
|
||||
def _due_now(*, assume_fresh: bool = False) -> dict[str, bool]:
|
||||
from hermes_cli.onboarding_run import should_run
|
||||
from hermes_cli.welcome_trial import inference_available
|
||||
|
||||
eligible = on_local_desktop_backend()
|
||||
if eligible:
|
||||
# Read in the default profile, where onboarding runs, not the launch profile.
|
||||
home = _profile_home("default")
|
||||
with _session_profile_runtime_scope({"profile_home": str(home) if home else None}):
|
||||
eligible = inference_available(assume_fresh=assume_fresh)
|
||||
return {"run": should_run(assume_fresh=assume_fresh), "eligible": eligible, **_active_run()}
|
||||
# Read in the default profile, where onboarding runs, not the launch profile.
|
||||
home = _profile_home("default")
|
||||
with _session_profile_runtime_scope({"profile_home": str(home) if home else None}):
|
||||
eligible = inference_available(assume_fresh=assume_fresh)
|
||||
return {"run": should_run(assume_fresh=assume_fresh), "eligible": eligible}
|
||||
|
||||
|
||||
def _due_stamp() -> tuple:
|
||||
"""The files the due answer reads, by size and mtime: the root config (``onboarding.run``) and
|
||||
the default profile's config, ``.env`` and ``auth.json`` (what carries a first chat)."""
|
||||
from hermes_constants import get_default_hermes_root
|
||||
|
||||
home = Path(_profile_home("default") or get_default_hermes_root())
|
||||
paths = (get_default_hermes_root() / "config.yaml", home / "config.yaml", home / ".env", home / "auth.json")
|
||||
return tuple((st.st_size, st.st_mtime_ns) if (st := _stat(path)) else None for path in paths)
|
||||
|
||||
|
||||
def _stat(path):
|
||||
try:
|
||||
return path.stat()
|
||||
except OSError:
|
||||
return None
|
||||
|
||||
|
||||
def start_due_boot() -> None:
|
||||
"""Serve boot, desktop's local backend: work out the due answer off the RPC path, after the setup
|
||||
bootstrap has recorded a fresh install and kept an existing guest's trial."""
|
||||
from agent.memory_provider import spawn_context_thread
|
||||
from hermes_cli import free_tier_bootstrap, onboarding_due
|
||||
from hermes_cli.anon_auth import on_local_desktop_backend
|
||||
|
||||
if not on_local_desktop_backend():
|
||||
return
|
||||
finish = onboarding_due.begin_boot(_due_stamp)
|
||||
|
||||
def compute() -> dict[str, bool]:
|
||||
free_tier_bootstrap.wait_for_record()
|
||||
return _due_now()
|
||||
|
||||
# finish() marks the boot done even when compute raises, so no reader waits on a dead thread.
|
||||
spawn_context_thread(finish, name="onboarding-due", args=(compute,)).start()
|
||||
|
||||
|
||||
# ── first-run onboarding: one chat in the default profile, driven by setup_beat ───────────────
|
||||
|
||||
@@ -176,7 +176,7 @@ _DETAIL_MODES = frozenset({"hidden", "collapsed", "expanded"})
|
||||
# interrupts); voice.*/wake.* = SYNCHRONOUS faster-whisper install (300s); session.workspace.move =
|
||||
# git subprocess probes on an arbitrary (maybe slow) mount; session.save = a full stored-session read + JSON
|
||||
# render (up to sessions.max_export_messages rows, ~0.8s at the default cap); onboarding.start = session
|
||||
# creation + a prompt.submit.
|
||||
# creation + a prompt.submit; onboarding.state = a bounded wait on the boot-time due answer.
|
||||
_LONG_HANDLERS = frozenset({
|
||||
"session.foreign.list", "session.foreign.preview", "session.foreign.import",
|
||||
"billing.state", "subscription.state", "subscription.preview", "subscription.change",
|
||||
@@ -192,7 +192,7 @@ _LONG_HANDLERS = frozenset({
|
||||
"setup.runtime_check", "setup.status", "free_tier.provision", "voice.toggle", "voice.record", "voice.tts", "wake.start",
|
||||
"wake.status", "session.active_list", "session.branch", "session.compress", "session.list",
|
||||
"session.resume", "session.save", "session.workspace.move", "shell.exec", "skills.manage", "slash.exec",
|
||||
"onboarding.start",
|
||||
"onboarding.start", "onboarding.state",
|
||||
"command.dispatch", # /goal draft invokes the auxiliary model; never block the RPC reader
|
||||
"shared_metrics.set", # consent reconcile waits on the metrics store's write lock
|
||||
})
|
||||
|
||||
Reference in new issue
Block a user