diff --git a/README.md b/README.md index 1cff1ea..b37fce4 100644 --- a/README.md +++ b/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 diff --git a/priompt_sdk/client.py b/priompt_sdk/client.py index b649cac..e568f2f 100644 --- a/priompt_sdk/client.py +++ b/priompt_sdk/client.py @@ -136,7 +136,7 @@ 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 | @@ -144,6 +144,13 @@ def subscribe(self, uri, on_change): 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() diff --git a/tests/test_client.py b/tests/test_client.py index f5e5eac..90f856a 100644 --- a/tests/test_client.py +++ b/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):