fix(achievements): apply the sticky-unlock floor to in-flight snapshots too

_compute_from_scan(is_partial=True) evaluated against an empty unlock
set, and _run_background_scan publishes those partials to
_SNAPSHOT_CACHE every progress_every sessions. So for the duration of
every background rescan an already-earned badge rendered as locked and
unlocked_count collapsed then climbed back — the exact 39<->40 flicker
from #112273 that the finished-scan floor alone did not remove.

Read state.json for partials as well (the floor applies); partials
still never record new unlocks or save_state, since an unlock time from
half a scan could be invalidated by a later session.

Also:
- tests/plugins fake SessionDB accepts include_compacted, which the
  scanner now passes (this was the red 'Python tests' check).
- engine tests patch get_hermes_home so _data_file's legacy migration
  cannot copy the developer's real state.json into the temp dir.

Part of #112273
This commit is contained in:
teknium1
2026-09-16 12:31:12 -07:00
committed by Teknium
parent ce71a85a9a
commit 44657ff628
3 changed files with 12 additions and 6 deletions

View File

@@ -711,9 +711,11 @@ def _compute_from_scan(scan: Dict[str, Any], *, is_partial: bool = False) -> Dic
"""Evaluate every achievement definition against a scan result. Used by ``compute_all``
for finished scans AND by the background progress callback for in-flight snapshots;
``is_partial=True`` skips persisting ``state.json`` unlocks — an "unlock time" from
half a scan could be invalidated by a later session."""
half a scan could be invalidated by a later session. Persisted unlocks are still read
for partials: the background scan publishes them to the cache, so an earned badge would
otherwise render as locked for the whole rescan (#112273)."""
aggregate = scan.get("aggregate", {})
state = load_state() if not is_partial else {"unlocks": {}}
state = load_state()
unlocks = state.setdefault("unlocks", {})
now = int(time.time())
evaluated = []

View File

@@ -67,15 +67,19 @@ class AchievementEngineTests(unittest.TestCase):
"threshold_metric": "total_terminal_calls",
"tiers": [{"name": "Copper", "threshold": 40}],
}
with TemporaryDirectory() as data_dir, patch.object(plugin_api, "ACHIEVEMENTS", [definition]), patch.object(plugin_api, "_data_dir", return_value=Path(data_dir)):
with TemporaryDirectory() as data_dir, patch.object(plugin_api, "ACHIEVEMENTS", [definition]), patch.object(plugin_api, "_data_dir", return_value=Path(data_dir)), patch.object(plugin_api, "get_hermes_home", return_value=Path(data_dir)):
unlocked = plugin_api._compute_from_scan({"aggregate": {"total_terminal_calls": 40}, "sessions": []})
rescanned = plugin_api._compute_from_scan({"aggregate": {"total_terminal_calls": 39}, "sessions": []})
partial = plugin_api._compute_from_scan({"aggregate": {"total_terminal_calls": 39}, "sessions": []}, is_partial=True)
persisted = plugin_api.load_state()["unlocks"]
self.assertTrue(unlocked["achievements"][0]["unlocked"])
self.assertTrue(rescanned["achievements"][0]["unlocked"])
self.assertEqual(rescanned["achievements"][0]["state"], "unlocked")
self.assertFalse(partial["achievements"][0]["unlocked"])
# In-flight snapshots are published to the cache during rescans: the floor applies there too.
self.assertTrue(partial["achievements"][0]["unlocked"])
self.assertEqual(partial["unlocked_count"], 1)
self.assertEqual(list(persisted), ["durable_unlock"])
def test_secret_achievement_stays_hidden_without_progress(self):
definition = {
@@ -197,7 +201,7 @@ class CompactionScanTests(unittest.TestCase):
import hermes_state
from hermes_state import SessionDB
with TemporaryDirectory() as tmp, patch.object(plugin_api, "_data_dir", return_value=Path(tmp) / "data"):
with TemporaryDirectory() as tmp, patch.object(plugin_api, "_data_dir", return_value=Path(tmp) / "data"), patch.object(plugin_api, "get_hermes_home", return_value=Path(tmp)):
db = SessionDB(Path(tmp) / "state.db")
try:
db.create_session("s1", "cli", model="m")

View File

@@ -100,7 +100,7 @@ class _FakeSessionDB:
for i in range(effective)
]
def get_messages(self, session_id: str) -> List[Dict[str, Any]]:
def get_messages(self, session_id: str, include_compacted: bool = False) -> List[Dict[str, Any]]:
self.messages_calls += 1
return [
{"role": "user", "content": f"ask {session_id}"},