feat(startup-ux): warming note on priming + recall fallbacks

The fallback already worked during warmup — silently. Now every consumer
phrases it honestly via the new _daemon.status() ('ready'/'warming'/'down',
one socket probe + the lifecycle flag + spawn-stamp breadcrumbs):

- Hook priming appends "*daemon still warming — lexical-fallback results;
  ranked recall returns within a minute or two*" to the bullets.
- MCP ultan_recall prefixes the same note (a warming "no match" may just
  mean the ranked index isn't serving yet).
- A stale state flag with a dead pid never claims warming.

Tests: 5 status-classification cases (incl. a real AF_UNIX listener for
'ready') + note-present/note-absent on the hook path. 30 root tests green
in both the full dev venv and the simulated thin CI env.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: nickroci

tests/test_warming_status.py

@@ -0,0 +1,101 @@
+"""During daemon warmup the fallback must keep working WITH a note — the
+agent should know priming/recall results are provisional, and `_daemon.status`
+is the single source of that truth for every consumer."""
+
+import json
+import os
+import socket
+from pathlib import Path
+
+import pytest
+
+from ultan import _daemon, _hooks
+
+
+@pytest.fixture()
+def home(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> Path:
+    h = tmp_path / "agent-mem"
+    h.mkdir()
+    monkeypatch.setenv("AGENT_MEM_HOME", str(h))
+    return h
+
+
+# ── _daemon.status() ──────────────────────────────────────────────────
+
+
+def test_status_down_with_no_signals(home: Path) -> None:
+    assert _daemon.status() == "down"
+
+
+def test_status_warming_from_state_flag_with_live_pid(home: Path) -> None:
+    (home / "daemon.state").write_text(
+        json.dumps({"phase": "warming", "pid": os.getpid(), "since": "x"}),
+        encoding="utf-8",
+    )
+    assert _daemon.status() == "warming"
+
+
+def test_status_ignores_stale_flag_with_dead_pid(home: Path) -> None:
+    (home / "daemon.state").write_text(
+        json.dumps({"phase": "warming", "pid": 99999999, "since": "x"}),
+        encoding="utf-8",
+    )
+    assert _daemon.status() == "down"
+
+
+def test_status_warming_from_fresh_spawn_stamp(home: Path) -> None:
+    (home / ".daemon-spawn-attempt").write_text("x", encoding="utf-8")
+    assert _daemon.status() == "warming"
+
+
+def test_status_ready_when_socket_answers(monkeypatch: pytest.MonkeyPatch) -> None:
+    # AF_UNIX paths cap at ~104 bytes on macOS; pytest tmp dirs are too deep,
+    # so bind in a short mkdtemp under /tmp instead.
+    import shutil
+    import tempfile
+
+    short_home = Path(tempfile.mkdtemp(prefix="ultan-t-", dir="/tmp"))
+    monkeypatch.setenv("AGENT_MEM_HOME", str(short_home))
+    srv = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+    try:
+        srv.bind(str(short_home / "priming.sock"))
+        srv.listen(1)
+        assert _daemon.status() == "ready"
+    finally:
+        srv.close()
+        shutil.rmtree(short_home, ignore_errors=True)
+
+
+# ── hook priming carries the warming note ─────────────────────────────
+
+
+def _dispatch_priming(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    status: str,
+) -> str:
+    monkeypatch.setattr(_hooks._daemon, "ensure_running", lambda: True)
+    monkeypatch.setattr(_hooks._daemon, "status", lambda: status)
+    monkeypatch.setattr(_hooks._priming, "get_priming", lambda *a, **k: "## bullets\n")
+    rc = _hooks.dispatch("user-prompt-submit", {"session_id": "s", "prompt": "q"})
+    assert rc == 0
+    out = capsys.readouterr().out
+    data = json.loads(out)
+    context: str = data["hookSpecificOutput"]["additionalContext"]
+    return context
+
+
+def test_priming_notes_lexical_fallback_while_warming(
+    home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    context = _dispatch_priming(monkeypatch, capsys, "warming")
+    assert "## bullets" in context  # the fallback still delivers
+    assert "warming up" in context  # ...but says so
+
+
+def test_priming_carries_no_note_when_ready(
+    home: Path, monkeypatch: pytest.MonkeyPatch, capsys: pytest.CaptureFixture[str]
+) -> None:
+    context = _dispatch_priming(monkeypatch, capsys, "ready")
+    assert "## bullets" in context
+    assert "warming" not in context

ultan/_daemon.py

@@ -16,12 +16,14 @@
 from __future__ import annotations
 
 import fcntl
+import json
 import os
 import socket
 import subprocess
 import sys
 import time
 from pathlib import Path
+from typing import Any, cast
 
 # The daemon console script ships in the same venv as this `ultan` install, so
 # it lives next to the running interpreter (sys.executable).
@@ -57,6 +59,44 @@ def _socket_answering() -> bool:
     return True
 
 
+def _pid_alive(pid: object) -> bool:
+    if not isinstance(pid, int) or pid <= 0:
+        return False
+    try:
+        os.kill(pid, 0)
+    except ProcessLookupError:
+        return False
+    except PermissionError:
+        return True  # exists, owned by someone else
+    except OSError:
+        return False
+    return True
+
+
+def status() -> str:
+    """Coarse daemon state for consumers that must phrase fallbacks honestly:
+    ``"ready"`` (socket answering), ``"warming"`` (daemon alive or freshly
+    spawned but not serving yet), or ``"down"``. One socket probe plus cheap
+    file stats — fine on the hook hot path."""
+    if _socket_answering():
+        return "ready"
+    home = _home()
+    # The daemon's own lifecycle flag (written between pid-acquire and exit).
+    try:
+        raw: Any = json.loads((home / "daemon.state").read_text(encoding="utf-8"))
+        if isinstance(raw, dict) and _pid_alive(cast("dict[str, Any]", raw).get("pid")):
+            return "warming"
+    except (OSError, ValueError):
+        pass
+    # A spawn was attempted moments ago (pre-flag window, or older daemon).
+    try:
+        if (time.time() - (home / ".daemon-spawn-attempt").stat().st_mtime) < 300.0:
+            return "warming"
+    except OSError:
+        pass
+    return "down"
+
+
 def run_foreground(extra_args: list[str]) -> int:
     """`ultan daemon` — run the installed agent-mem-daemon in the foreground."""
     daemon = _daemon_bin()

ultan/_hooks.py

@@ -73,6 +73,15 @@ def _user_prompt_submit(payload: dict[str, Any]) -> int:
         prompt if isinstance(prompt, str) else "",
         session_id=session_id if isinstance(session_id, str) else None,
     )
+    # Fallback honesty: while the daemon warms (first start loads models for
+    # minutes), priming comes from the crude lexical scan. Say so, so the
+    # agent treats the bullets as provisional rather than the library's best.
+    if md and _daemon.status() == "warming":
+        md += (
+            "\n*Ultan's daemon is still warming up — the bullets above are "
+            "lexical-fallback results; full ranked recall returns within a "
+            "minute or two.*\n"
+        )
     _emit_additional_context("UserPromptSubmit", md)
     return 0
 

ultan/_mcp.py

@@ -25,7 +25,7 @@ def build_server() -> "FastMCP":
     daemon spawn, no I/O — so it's unit-testable."""
     from mcp.server.fastmcp import FastMCP
 
-    from . import _priming
+    from . import _daemon, _priming
 
     server = FastMCP("ultan")
 
@@ -34,7 +34,15 @@ def ultan_recall(query: str) -> str:  # pyright: ignore[reportUnusedFunction]  #
         """Recall relevant lessons, preferences, and conventions from the
         user's Ultan memory library for the given query. Returns markdown
         wikilink bullets (open them with the Read tool) or a no-match note."""
-        return _priming.get_priming(query, k=5) or "(no relevant Ultan memory for this query)"
+        result = _priming.get_priming(query, k=5) or "(no relevant Ultan memory for this query)"
+        # Fallback honesty: during warmup results come from the lexical scan,
+        # and a "no match" may just mean the ranked index isn't serving yet.
+        if _daemon.status() == "warming":
+            return (
+                "*(Ultan daemon is warming up — lexical-fallback results; "
+                "retry in a minute or two for full ranked recall.)*\n\n" + result
+            )
+        return result
 
     return server