fix: let the reload policy decide whether subscribe refreshes the cache

README.md

@@ -78,7 +78,7 @@ PromptClient(host=None, token=None, tls=False, ca_cert=None,
 client.get(uri, ref="")          # fetch a prompt (ref = branch or commit hash to pin a version)
 client.list(prefix="")           # browse a repo (URI prefix) like a folder
 client.diff(uri, new_template)   # semantic diff: stored version vs. your draft
-client.subscribe(uri, on_change) # be notified the moment the prompt changes
+client.subscribe(uri, on_change, reload=None)  # be notified when the prompt changes
 client.close()
 ```
 
@@ -131,7 +131,11 @@ def on_change(version, classification):
     else:
         alert_a_human(version)      # structural, or unclassified — hold it
 
-client.subscribe("priompt://acme/support/agent", on_change)   # needs the [nats] extra
+# reload= is the same policy for the client cache: an accepted change is fetched
+# into the cache before on_change runs; a held one is not, so get() keeps
+# returning the old version until cache_ttl expires. Default: hold everything.
+client.subscribe("priompt://acme/support/agent", on_change,
+                 reload=AUTO_RELOAD)                          # needs the [nats] extra
 ```
 
 **The verdict measures spread, not risk.** A one-line edit turning *"offer a

priompt_sdk/client.py

@@ -136,14 +136,21 @@ def diff(self, uri, new_template):
             metadata=self._md,
         )
 
-    def subscribe(self, uri, on_change):
+    def subscribe(self, uri, on_change, reload=None):
         """Register as a subscriber: call on_change(version_hash, classification)
         whenever the prompt at `uri` is republished (push). `classification` is
         the semantic diff verdict (structural | localized tweak | minor edit |
         new | ""), so an agent can auto-reload a tweak but hold a structural
         change. TTL polling via cache_ttl is the pull side. Returns the daemon
         thread running the NATS subscription.
 
+        `reload` is the reload policy for the client cache: a set of verdicts,
+        e.g. {"minor edit", "localized tweak", "new"}, or a function
+        (version_hash, classification) -> bool. A change it accepts is fetched
+        into the cache before on_change runs, so get() returns it at once. Any
+        other change is held: get() keeps returning the cached version until
+        cache_ttl expires. The default (None) holds every change.
+
         Requires `pip install nats-py` and nats_url set on the client. Raises
         ConnectionError if the broker can't be reached or refuses the
         connection, e.g. a wrong NATS token ('Authorization Violation').
@@ -180,13 +187,22 @@ async def main():
                     return
 
                 async def handler(msg):
-                    self._cache.pop(uri, None)  # drop stale cache
-                    self.get(uri)  # refetch now so the next get() is warm
                     try:
                         ev = json.loads(msg.data.decode())
-                        on_change(ev.get("version", ""), ev.get("classification", ""))
+                        version = ev.get("version", "")
+                        classification = ev.get("classification", "")
                     except (ValueError, AttributeError):
-                        on_change(msg.data.decode(), "")  # pre-0.7 bare-hash body
+                        version, classification = msg.data.decode(), ""  # pre-0.7 bare-hash body
+                    if callable(reload):
+                        accept = reload(version, classification)
+                    else:
+                        accept = classification in (reload or ())
+                    # Only an accepted change refreshes the cache; a held one
+                    # keeps serving the cached version until cache_ttl expires.
+                    if accept and self._cache_ttl:
+                        self._cache.pop(uri, None)  # drop stale cache
+                        self.get(uri)  # refetch now so the next get() is warm
+                    on_change(version, classification)
 
                 await nc.subscribe(subject, cb=handler)
                 ready.set()

tests/test_client.py

@@ -154,6 +154,92 @@ def test_subscribe_requires_nats_url():
         c.subscribe("priompt://x", lambda *_: None)
 
 
+# --- subscribe: the reload policy controls the cache ----------------------
+
+@pytest.fixture
+def fake_nats(monkeypatch):
+    """Swap in a fake `nats` module that captures the subscription callback, so
+    a test can deliver an event without a broker. deliver(body) runs the
+    handler to completion on the calling thread."""
+    import asyncio
+    import json
+    import sys
+    import types
+
+    subscribed = threading.Event()
+    state = {}
+
+    class FakeNC:
+        async def subscribe(self, subject, cb):
+            state["cb"] = cb
+            subscribed.set()
+
+    async def connect(url, **kw):
+        return FakeNC()
+
+    monkeypatch.setitem(sys.modules, "nats", types.SimpleNamespace(connect=connect))
+
+    def deliver(version, classification):
+        assert subscribed.wait(5), "subscribe() never registered a handler"
+        body = json.dumps({"version": version, "classification": classification})
+        asyncio.run(state["cb"](types.SimpleNamespace(data=body.encode())))
+
+    return deliver
+
+
+def make_subscriber(cache_ttl=30):
+    c = PromptClient("localhost:1", cache_ttl=cache_ttl, nats_url="nats://fake:4222")
+    c._stub = FakeStub()
+    return c
+
+
+def test_held_change_keeps_cached_version(clock, fake_nats):
+    """#3: a change the policy holds must not refresh the cache, or get()
+    serves the held version anyway."""
+    c = make_subscriber()
+    before = c.get("priompt://x")
+    seen = []
+    c.subscribe("priompt://x", lambda v, cls: seen.append((v, cls)),
+                reload={"minor edit", "localized tweak", "new"})
+    fake_nats("v2", "structural")
+    assert seen == [("v2", "structural")]
+    assert c.get("priompt://x") == before
+    assert c._stub.get_calls == 1
+
+
+def test_default_policy_holds_every_change(clock, fake_nats):
+    c = make_subscriber()
+    before = c.get("priompt://x")
+    c.subscribe("priompt://x", lambda *_: None)
+    fake_nats("v2", "minor edit")
+    assert c.get("priompt://x") == before
+
+
+def test_accepted_change_refreshes_cache_before_on_change(clock, fake_nats):
+    """A change the policy accepts is in the cache by the time on_change runs,
+    so get() inside the callback already returns it."""
+    c = make_subscriber()
+    before = c.get("priompt://x")
+    inside = []
+    c.subscribe("priompt://x", lambda v, cls: inside.append(c.get("priompt://x")),
+                reload={"localized tweak"})
+    fake_nats("v2", "localized tweak")
+    assert inside and inside[0] != before
+    assert c.get("priompt://x") == inside[0]
+    assert c._stub.get_calls == 2
+
+
+def test_reload_policy_can_be_a_function(clock, fake_nats):
+    c = make_subscriber()
+    before = c.get("priompt://x")
+    c.subscribe("priompt://x", lambda *_: None,
+                reload=lambda version, cls: cls != "structural")
+    fake_nats("v2", "structural")
+    assert c.get("priompt://x") == before
+    fake_nats("v3", "")
+    assert c.get("priompt://x") != before
+
+
 # --- stress: lock-free cache under concurrent readers ---------------------
 
 def test_concurrent_gets_do_not_crash_or_corrupt(clock):