diff --git a/Dockerfile b/Dockerfile index 6f0ba7e2..cc6d187e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -30,6 +30,10 @@ ENV UV_COMPILE_BYTECODE=1 UV_LINK_MODE=copy PYTHONUNBUFFERED=1 WORKDIR /app # Copy project metadata + every app package (flat top-level layout). +# +# This list is explicit, so a NEW APP MUST BE ADDED HERE as well as to +# INSTALLED_APPS and the wheel `packages` list — three places, none of which +# fail at test time. tests/test_app_package_names.py asserts all three agree. COPY pyproject.toml uv.lock manage.py ./ COPY config/ ./config/ COPY jawafdehi_shared/ ./jawafdehi_shared/ @@ -42,6 +46,7 @@ COPY review/ ./review/ COPY case_proposals/ ./case_proposals/ COPY newsletter/ ./newsletter/ COPY jobs/ ./jobs/ +COPY case_events/ ./case_events/ COPY llm/ ./llm/ COPY search/ ./search/ COPY discovery/ ./discovery/ diff --git a/case_events/__init__.py b/case_events/__init__.py new file mode 100644 index 00000000..e921bee2 --- /dev/null +++ b/case_events/__init__.py @@ -0,0 +1,17 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""The case-enrichment event bus (NATS + JetStream). + +Producers publish observed facts to ``jaw.signal.>``; consumers turn them into +*proposals*; a caseworker approves. Automation never writes a case directly, and +the bus is transport — **not** a system of record. Everything durable lives in +the case record and the proposal store, which is what makes a single-replica +pilot broker an acceptable trade. + +The one invariant worth stating up front: **publishing is best-effort and must +never fail a write.** A broker outage degrades enrichment; it does not degrade +the archive. See :mod:`case_events.bus`. + +Nothing here is imported at Django startup, and with ``NATS_URL`` unset every +publish is a logged no-op — so the monolith runs unchanged with no broker at +all, and dev/CI need none. +""" diff --git a/case_events/apps.py b/case_events/apps.py new file mode 100644 index 00000000..a13aaa9f --- /dev/null +++ b/case_events/apps.py @@ -0,0 +1,14 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +from django.apps import AppConfig + + +class EventsConfig(AppConfig): + """The case-enrichment event bus. + + Registered as an app for management-command discovery (the consumer runner + lands here). It has no models and therefore no migrations, and it opens no + connection at startup — the bus connects lazily on first publish. + """ + + default_auto_field = "django.db.models.BigAutoField" + name = "case_events" diff --git a/case_events/bus.py b/case_events/bus.py new file mode 100644 index 00000000..6d719b7b --- /dev/null +++ b/case_events/bus.py @@ -0,0 +1,240 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Publishing to the bus from synchronous Django code. + +``nats-py`` is asyncio-only; Django's request path is not. The bridge is one +background thread per process running an event loop, holding one long-lived +connection, started lazily on first publish. Publishes are handed to that loop +with ``run_coroutine_threadsafe``. + +Three rules this module exists to enforce: + +**Publishing is best-effort and never raises.** Every entry point swallows +its exceptions and logs. A broker outage must degrade enrichment, never a case +write — the bus is transport, the case record is the truth. If you find yourself +wanting to propagate an error from here, the thing you actually want is a job. + +**One connection per process, not one per publish.** Connecting per call would +turn a burst of approvals into a burst of TCP handshakes, and this cluster has +already had an incident where a per-operation connection pattern exhausted a +server's connection cap. The loop thread and its connection are reused. + +**A dead broker must not slow the request path.** The first publish after a +connection failure does not retry immediately; it fails fast for +:data:`CONNECT_RETRY_SECONDS` before trying again. Publishes are fire-and-forget +by default, so nothing in a web request waits on the network. + +With ``settings.NATS_URL`` unset every publish is a logged no-op and no thread +is ever started, which is what lets this ship before the broker exists. +""" + +from __future__ import annotations + +import asyncio +import json +import os +import threading +import time +from typing import Any + +import structlog +from django.conf import settings + +logger = structlog.get_logger(__name__) + +#: After a failed connect, fail fast for this long before trying again. Without +#: it, every request on a broker outage pays a full connect timeout. +CONNECT_RETRY_SECONDS = 30 + +#: Ceiling on how long the lazy startup may block the calling thread. Deliberately +#: short: this runs inside a web request the first time. +STARTUP_TIMEOUT_SECONDS = 5 + +#: Ceiling for a publish when the caller opts into waiting (``wait=True``). +PUBLISH_TIMEOUT_SECONDS = 5 + + +def enabled() -> bool: + """True when a broker is configured. Everything here no-ops when False.""" + return bool(getattr(settings, "NATS_URL", "")) + + +class _Bus: + """Owns the loop thread and the connection. One instance per process.""" + + def __init__(self): + self._lock = threading.Lock() + self._loop: asyncio.AbstractEventLoop | None = None + self._thread: threading.Thread | None = None + self._nc = None + self._js = None + self._pid: int | None = None + self._last_failure_at: float = 0.0 + + # ── lifecycle ──────────────────────────────────────────────────────────── + + def _reset_locked(self): + self._loop = None + self._thread = None + self._nc = None + self._js = None + self._pid = None + + def _ensure_started(self) -> bool: + """Start the loop thread and connect if needed. False if unavailable.""" + with self._lock: + # A forked child inherits this object's state but NOT the parent's + # threads, so the loop it points at is gone. Under gunicorn the fork + # happens before any request, so in practice we start fresh in the + # child — but only because we check. + if self._pid is not None and self._pid != os.getpid(): + logger.info("case_events.bus_reset_after_fork", inherited_pid=self._pid) + self._reset_locked() + + if self._js is not None: + return True + + if time.monotonic() - self._last_failure_at < CONNECT_RETRY_SECONDS: + return False + + try: + self._start_locked() + return True + except Exception as exc: # noqa: BLE001 - never propagate to a write path + self._last_failure_at = time.monotonic() + self._reset_locked() + logger.warning("case_events.connect_failed", error=str(exc)) + return False + + def _start_locked(self): + loop = asyncio.new_event_loop() + thread = threading.Thread( + target=loop.run_forever, name="events-bus", daemon=True + ) + thread.start() + + future = asyncio.run_coroutine_threadsafe(self._connect(), loop) + try: + self._nc, self._js = future.result(timeout=STARTUP_TIMEOUT_SECONDS) + except Exception: + loop.call_soon_threadsafe(loop.stop) + raise + + self._loop = loop + self._thread = thread + self._pid = os.getpid() + logger.info("case_events.connected", url=_redact(settings.NATS_URL)) + + async def _connect(self): + import nats + + from case_events.streams import ensure_streams + + nc = await nats.connect( + settings.NATS_URL, + name="jawafdehi-platform", + connect_timeout=STARTUP_TIMEOUT_SECONDS, + # Reconnect forever rather than giving up: this process outlives any + # broker restart, and a permanently-detached publisher that still + # looks healthy is worse than one that keeps trying. + max_reconnect_attempts=-1, + ) + js = nc.jetstream() + await ensure_streams(js) + return nc, js + + def close(self): + """Drain and disconnect. For tests and orderly shutdown.""" + with self._lock: + loop, nc = self._loop, self._nc + self._reset_locked() + if loop is None: + return + try: + if nc is not None: + asyncio.run_coroutine_threadsafe(nc.drain(), loop).result(timeout=5) + except Exception as exc: # noqa: BLE001 - shutdown is best-effort too + logger.warning("case_events.close_failed", error=str(exc)) + finally: + loop.call_soon_threadsafe(loop.stop) + + # ── publishing ─────────────────────────────────────────────────────────── + + def publish(self, subject: str, envelope: dict[str, Any], wait: bool = False) -> bool: + """Publish one envelope. Returns True if it was handed to the loop. + + A True return means accepted for delivery, not delivered — with + ``wait=False`` the JetStream ack arrives after this returns, and a + failure then surfaces in the logs via the done-callback. + """ + if not self._ensure_started(): + return False + + body = json.dumps(envelope, ensure_ascii=False, default=str).encode("utf-8") + # Nats-Msg-Id is what makes JetStream collapse a duplicate publish inside + # its dedup window — the same idempotency spine as the proposal's + # dedup_key. Omitted rather than sent empty when a producer has no key. + headers = {} + if envelope.get("dedup_key"): + headers["Nats-Msg-Id"] = envelope["dedup_key"] + + try: + future = asyncio.run_coroutine_threadsafe( + self._js.publish(subject, body, headers=headers or None), self._loop + ) + except Exception as exc: # noqa: BLE001 - loop may have died under us + logger.warning("case_events.publish_failed", subject=subject, error=str(exc)) + return False + + if wait: + try: + future.result(timeout=PUBLISH_TIMEOUT_SECONDS) + except Exception as exc: # noqa: BLE001 + logger.warning("case_events.publish_failed", subject=subject, error=str(exc)) + return False + else: + future.add_done_callback(lambda f: _log_result(f, subject)) + return True + + +def _log_result(future, subject: str): + """Surface a fire-and-forget failure. Nothing else observes these.""" + try: + future.result() + except Exception as exc: # noqa: BLE001 + logger.warning("case_events.publish_failed", subject=subject, error=str(exc)) + + +def _redact(url: str) -> str: + """Strip credentials from a nats:// URL before logging it.""" + if not url or "@" not in url: + return url + scheme, _, rest = url.partition("://") + return f"{scheme}://***@{rest.rpartition('@')[2]}" if scheme else url + + +_bus = _Bus() + + +def publish(subject: str, envelope: dict[str, Any], wait: bool = False) -> bool: + """Best-effort publish. Never raises; returns False when nothing was sent. + + Args: + subject: See :mod:`case_events.subjects`. + envelope: Built by :func:`case_events.envelope.build_envelope`. + wait: Block for the JetStream ack (up to + :data:`PUBLISH_TIMEOUT_SECONDS`). Leave False in a request path; + useful in management commands and tests where you want the result. + """ + if not enabled(): + logger.debug("case_events.publish_skipped", subject=subject, reason="NATS_URL unset") + return False + try: + return _bus.publish(subject, envelope, wait=wait) + except Exception as exc: # noqa: BLE001 - the whole point of this module + logger.warning("case_events.publish_failed", subject=subject, error=str(exc)) + return False + + +def close(): + """Drain and disconnect the process-wide connection.""" + _bus.close() diff --git a/case_events/envelope.py b/case_events/envelope.py new file mode 100644 index 00000000..bbe5bb58 --- /dev/null +++ b/case_events/envelope.py @@ -0,0 +1,93 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""The message envelope every event on the bus carries. + +One shape for signals and case events alike, so a consumer can read provenance, +identity and timing without knowing the subject. The type-specific part is +confined to ``payload``. + +Two fields carry more weight than they look: + +``subject_refs`` are stable ``@id`` IRIs (case, court-case, NES entity, +material) and are **the join key** between a message and our records. They must +be built with :mod:`jawafdehi_shared.entities.ids`, never formatted by hand: +``build_courtcase_iri`` lowercases both segments, so a hand-rolled +``.../courtcase/special/082-CR-0154`` matches nothing. + +``dedup_key`` is sent as the ``Nats-Msg-Id`` header, which is what makes +JetStream drop a duplicate publish inside its dedup window. It is the same +idempotency spine ``CaseUpdateProposal.dedup_key`` uses, and producers must +construct it deterministically from the fact — not from a timestamp or a random +id, or it defeats itself. +""" + +from __future__ import annotations + +from datetime import datetime, timezone +from typing import Any + + +def utcnow() -> datetime: + """Timezone-aware UTC now. Seam for tests to freeze.""" + return datetime.now(timezone.utc) + + +def _iso(value: datetime) -> str: + """RFC 3339 / ISO 8601 in UTC, always with a ``Z`` suffix. + + ``datetime.isoformat()`` renders UTC as ``+00:00``; normalising to ``Z`` + keeps the wire format stable for non-Python consumers. + """ + if value.tzinfo is None: + value = value.replace(tzinfo=timezone.utc) + return value.astimezone(timezone.utc).isoformat().replace("+00:00", "Z") + + +def build_envelope( + *, + subject: str, + payload: dict[str, Any], + producer: str, + subject_refs: list[str] | None = None, + dedup_key: str = "", + source: str = "", + raw_ref: str = "", + occurred_at: datetime | None = None, +) -> dict[str, Any]: + """Assemble one bus message. + + Args: + subject: The subject it will be published on (see :mod:`case_events.subjects`). + Carried in the body as well as the NATS subject so a message stays + self-describing once it has been archived, DLQ'd, or re-published + under a different subject. + payload: Type-specific body. Must be JSON-serialisable. + producer: What emitted this — ``"platform"`` for the monolith, + ``"consumer:"`` for a consumer, ``"producer:"`` for a + scraper. + subject_refs: Stable ``@id`` IRIs this message is about. + dedup_key: Deterministic idempotency key; becomes ``Nats-Msg-Id``. + source: Where the underlying fact came from (URL, ``@id``, or a + well-known token like ``"caseworker"``). + raw_ref: Pointer to the raw artefact behind the fact, when there is one. + occurred_at: When the fact happened. Defaults to now, but should be + passed whenever the real time is known — for a scraped hearing that + is the docket date, not the moment we noticed it. + + Returns: + A JSON-serialisable dict. + """ + now = utcnow() + return { + "subject": subject, + "producer": producer, + "subject_refs": list(subject_refs or []), + "dedup_key": dedup_key, + "source": source, + "raw_ref": raw_ref, + # occurred_at is when the FACT happened; published_at is when we emitted + # it. They differ by however long the producer took to notice, which is + # the number you need when auditing lag. + "occurred_at": _iso(occurred_at or now), + "published_at": _iso(now), + "payload": payload, + } diff --git a/case_events/streams.py b/case_events/streams.py new file mode 100644 index 00000000..5ad9b61a --- /dev/null +++ b/case_events/streams.py @@ -0,0 +1,101 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""JetStream stream topology, asserted from code rather than declared in YAML. + +NATS has no CRD and no declarative stream config in the server file, so the +topology has to be created by *something*. The alternative — a one-shot ``Job`` +running ``nats stream add`` — re-runs awkwardly against existing streams and +drifts silently once someone edits one by hand. Asserting it from application +startup keeps the definition next to the code that depends on it, re-applies it +on every deploy, and means a fresh or local environment needs no bootstrap step. + +``add_stream`` is upsert-like: creating a stream that already exists with the +same config is a no-op, so this is safe to call on every process start. + +**Replicas are 1 for the pilot, deliberately.** With ``local-path`` storage the +pod is pinned to one node, so that node's disk *is* the bus. Going to R3 is not +a number change here — it needs three pinned nodes with three PVCs, i.e. a +re-deploy. That is tolerable only because nothing here is a system of record: +``SIGNALS`` is re-derivable from the NGM lake and the CIAA source, and committed +updates live in the case record. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +import structlog + +from case_events import subjects + +logger = structlog.get_logger(__name__) + +#: One year, in seconds — the retention window for all three streams in the +#: pilot. SIGNALS is the one that could grow enough to want trimming first. +ONE_YEAR_SECONDS = 365 * 24 * 60 * 60 + + +@dataclass(frozen=True) +class StreamSpec: + name: str + subjects: tuple[str, ...] + description: str + max_age_seconds: int = ONE_YEAR_SECONDS + replicas: int = 1 + + +STREAMS: tuple[StreamSpec, ...] = ( + StreamSpec( + name="SIGNALS", + subjects=(subjects.ALL_SIGNALS,), + description="Raw observed facts from producers. Replayable, re-derivable.", + ), + StreamSpec( + name="CASE_EVENTS", + subjects=(subjects.ALL_CASE_EVENTS,), + description="The case-domain log: matches, proposals, and decisions.", + ), + StreamSpec( + name="DLQ", + subjects=(subjects.ALL_DLQ,), + description="Poison messages past MaxDeliver, kept for human triage.", + ), +) + + +async def ensure_streams(js) -> list[str]: + """Idempotently assert every stream in :data:`STREAMS`. + + Args: + js: A JetStream context (``nats.aio.client.Client.jetstream()``). + + Returns: + The names asserted, in order. + + Raises: + Whatever the client raises. Unlike publishing, this is NOT best-effort: + a consumer that cannot see its stream has nothing to do, and should fail + loudly at startup rather than idle while looking healthy. + """ + from nats.js.api import RetentionPolicy, StorageType, StreamConfig + + asserted = [] + for spec in STREAMS: + await js.add_stream( + StreamConfig( + name=spec.name, + subjects=list(spec.subjects), + description=spec.description, + retention=RetentionPolicy.LIMITS, + storage=StorageType.FILE, + max_age=spec.max_age_seconds, + num_replicas=spec.replicas, + ) + ) + asserted.append(spec.name) + logger.info( + "case_events.stream_asserted", + stream=spec.name, + subjects=list(spec.subjects), + replicas=spec.replicas, + ) + return asserted diff --git a/case_events/subjects.py b/case_events/subjects.py new file mode 100644 index 00000000..e0b0a8c2 --- /dev/null +++ b/case_events/subjects.py @@ -0,0 +1,44 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Subject vocabulary for the bus. + +Two tiers, and the split is deliberate: ``jaw.signal.>`` is raw observed facts +straight from producers (noisy, re-derivable), while ``jaw.case.>`` is the +case-domain log — the audit trail of what the system decided. Consumers filter +by subject, so a consumer that only cares about decisions never sees the noise. + +Subjects are string constants rather than an enum because they are also matched +with wildcards (``jaw.case.update.*``) and are compared against values arriving +off the wire, where a bare string is what you actually have. +""" + +# ── jaw.signal.> — raw observed facts from producers ───────────────────────── +SIGNAL_DOCKET_HEARING_ADDED = "jaw.signal.docket.hearing.added" +SIGNAL_DOCKET_VERDICT_ENTERED = "jaw.signal.docket.verdict.entered" +SIGNAL_DOCKET_STATUS_CHANGED = "jaw.signal.docket.status.changed" +SIGNAL_COURTORDER_PUBLISHED = "jaw.signal.courtorder.published" +SIGNAL_CIAA_PRESSRELEASE = "jaw.signal.ciaa.pressrelease" +SIGNAL_NEWS_MATCHED = "jaw.signal.news.matched" +SIGNAL_MANUAL_NOTE = "jaw.signal.manual.note" + +# ── jaw.case.> — the case-domain log ───────────────────────────────────────── +CASE_MATCHED = "jaw.case.matched" +CASE_UPDATE_PROPOSED = "jaw.case.update.proposed" +CASE_UPDATE_APPROVED = "jaw.case.update.approved" +CASE_UPDATE_REJECTED = "jaw.case.update.rejected" + +# ── wildcards, for consumer filter subjects ────────────────────────────────── +ALL_SIGNALS = "jaw.signal.>" +ALL_CASE_EVENTS = "jaw.case.>" +ALL_CASE_UPDATES = "jaw.case.update.*" +ALL_DLQ = "jaw.dlq.>" + +#: Prefix for poison messages that exhausted their delivery budget. The +#: originating subject is appended, so a message that died on +#: ``jaw.case.matched`` lands on ``jaw.dlq.jaw.case.matched`` and stays +#: attributable. Nothing is dropped silently. +DLQ_PREFIX = "jaw.dlq." + + +def dlq_subject(original_subject: str) -> str: + """The DLQ subject a poison message from ``original_subject`` republishes to.""" + return f"{DLQ_PREFIX}{original_subject}" diff --git a/case_events/tests/__init__.py b/case_events/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/case_events/tests/test_bus.py b/case_events/tests/test_bus.py new file mode 100644 index 00000000..184ece18 --- /dev/null +++ b/case_events/tests/test_bus.py @@ -0,0 +1,167 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Tests for the bus: envelope shape, stream topology, and the no-op guarantee. + +No broker is involved. The `nats` client is mocked wherever it would be reached, +which is also the point of the design under test — the disabled path must not +even try. +""" + +import json +from datetime import datetime, timezone +from unittest import mock + +import pytest +from django.test import override_settings + +from case_events import bus, streams, subjects +from case_events.envelope import build_envelope + + +class TestEnvelope: + def test_carries_the_subject_in_the_body(self): + # A DLQ'd or archived message must still say what it was. + env = build_envelope(subject="jaw.case.matched", payload={}, producer="platform") + assert env["subject"] == "jaw.case.matched" + + def test_timestamps_are_z_suffixed_utc(self): + env = build_envelope(subject="s", payload={}, producer="p") + for field in ("occurred_at", "published_at"): + assert env[field].endswith("Z"), field + assert "+00:00" not in env[field] + + def test_naive_occurred_at_is_treated_as_utc(self): + env = build_envelope( + subject="s", payload={}, producer="p", + occurred_at=datetime(2026, 7, 30, 12, 0, 0), + ) + assert env["occurred_at"] == "2026-07-30T12:00:00Z" + + def test_aware_occurred_at_is_converted_to_utc(self): + tz = timezone(offset=__import__("datetime").timedelta(hours=5, minutes=45)) + env = build_envelope( + subject="s", payload={}, producer="p", + occurred_at=datetime(2026, 7, 30, 17, 45, 0, tzinfo=tz), + ) + assert env["occurred_at"] == "2026-07-30T12:00:00Z" + + def test_occurred_at_defaults_to_now_but_is_distinct_from_published_at(self): + # Both default to "now", but they are separate fields because a producer + # that knows the real fact time must be able to set one without the other. + env = build_envelope(subject="s", payload={}, producer="p") + assert "occurred_at" in env and "published_at" in env + + def test_subject_refs_default_to_empty_list_not_none(self): + assert build_envelope(subject="s", payload={}, producer="p")["subject_refs"] == [] + + def test_is_json_serialisable_with_nepali(self): + env = build_envelope( + subject="s", producer="p", + payload={"title": "अख्तियार दुरुपयोग अनुसन्धान आयोग"}, + ) + # ensure_ascii=False keeps Devanagari readable on the wire. + raw = json.dumps(env, ensure_ascii=False) + assert "अख्तियार" in raw + + +class TestStreams: + def test_three_streams_cover_the_three_subject_trees(self): + by_name = {s.name: s for s in streams.STREAMS} + assert set(by_name) == {"SIGNALS", "CASE_EVENTS", "DLQ"} + assert by_name["SIGNALS"].subjects == (subjects.ALL_SIGNALS,) + assert by_name["CASE_EVENTS"].subjects == (subjects.ALL_CASE_EVENTS,) + assert by_name["DLQ"].subjects == (subjects.ALL_DLQ,) + + def test_pilot_is_single_replica(self): + # R1 is a deliberate pilot trade (node-local disk). If this ever changes + # to 3, the manifests need three pinned nodes and three PVCs first. + assert all(s.replicas == 1 for s in streams.STREAMS) + + def test_every_case_subject_falls_under_the_case_events_stream(self): + for subject in ( + subjects.CASE_MATCHED, + subjects.CASE_UPDATE_PROPOSED, + subjects.CASE_UPDATE_APPROVED, + subjects.CASE_UPDATE_REJECTED, + ): + assert subject.startswith(subjects.ALL_CASE_EVENTS.rstrip(">")) + + def test_dlq_subject_preserves_the_original(self): + # A poison message must stay attributable to where it died. + assert subjects.dlq_subject("jaw.case.matched") == "jaw.dlq.jaw.case.matched" + + +class TestDisabledByDefault: + @override_settings(NATS_URL="") + def test_not_enabled_without_a_url(self): + assert bus.enabled() is False + + @override_settings(NATS_URL="nats://localhost:4222") + def test_enabled_with_a_url(self): + assert bus.enabled() is True + + @override_settings(NATS_URL="") + def test_publish_is_a_noop_and_never_touches_the_bus(self): + with mock.patch.object(bus._bus, "publish") as inner: + assert bus.publish("jaw.case.update.approved", {"x": 1}) is False + inner.assert_not_called() + + @override_settings(NATS_URL="nats://localhost:4222") + def test_publish_never_raises_when_the_broker_is_unreachable(self): + # The core guarantee: a broken bus degrades to False, not an exception. + with mock.patch.object(bus._bus, "_ensure_started", return_value=False): + assert bus.publish("jaw.case.update.approved", {"x": 1}) is False + + @override_settings(NATS_URL="nats://localhost:4222") + def test_publish_swallows_an_unexpected_error(self): + with mock.patch.object(bus._bus, "publish", side_effect=RuntimeError("boom")): + assert bus.publish("s", {}) is False + + +class TestPublishMechanics: + @override_settings(NATS_URL="nats://localhost:4222") + def test_dedup_key_becomes_the_nats_msg_id_header(self): + # This header is what makes JetStream collapse a duplicate publish. + captured = {} + + def fake_run(coro, loop): + coro.close() + return mock.Mock() + + with mock.patch.object(bus._bus, "_ensure_started", return_value=True), \ + mock.patch.object(bus._bus, "_js") as js, \ + mock.patch.object(bus._bus, "_loop", mock.Mock()), \ + mock.patch("case_events.bus.asyncio.run_coroutine_threadsafe", side_effect=fake_run): + js.publish.side_effect = lambda *a, **kw: captured.update(kw) or mock.Mock() + bus._bus.publish("s", {"dedup_key": "docket:x:hearing:1"}) + + assert captured["headers"] == {"Nats-Msg-Id": "docket:x:hearing:1"} + + @override_settings(NATS_URL="nats://localhost:4222") + def test_no_header_when_there_is_no_dedup_key(self): + captured = {} + + def fake_run(coro, loop): + coro.close() + return mock.Mock() + + with mock.patch.object(bus._bus, "_ensure_started", return_value=True), \ + mock.patch.object(bus._bus, "_js") as js, \ + mock.patch.object(bus._bus, "_loop", mock.Mock()), \ + mock.patch("case_events.bus.asyncio.run_coroutine_threadsafe", side_effect=fake_run): + js.publish.side_effect = lambda *a, **kw: captured.update(kw) or mock.Mock() + bus._bus.publish("s", {"dedup_key": ""}) + + assert captured["headers"] is None + + +class TestRedaction: + @pytest.mark.parametrize( + "url,expected", + [ + ("nats://user:secret@host:4222", "nats://***@host:4222"), + ("nats://host:4222", "nats://host:4222"), + ("", ""), + ], + ) + def test_credentials_never_reach_the_logs(self, url, expected): + assert bus._redact(url) == expected diff --git a/case_proposals/publish.py b/case_proposals/publish.py new file mode 100644 index 00000000..813150c9 --- /dev/null +++ b/case_proposals/publish.py @@ -0,0 +1,114 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Publishing proposal decisions to the event bus. + +This is the first real producer on ``jaw.case.>``, and it closes the loop the +design draws: a committed update is itself an event, so downstream consumers +(notify a caseworker, refresh stats, re-index) react to a decision instead of +polling for one. + +Everything here is best-effort and fires ``on_commit``, mirroring the existing +``_schedule_reindex`` / ``_schedule_material_visibility`` hooks next door. Two +reasons it must be after commit rather than inside the transaction: a subscriber +that reacts instantly would otherwise be able to read the case *before* the +write is visible, and a rolled-back transaction would have already announced a +decision that never happened. +""" + +from __future__ import annotations + +import structlog +from django.db import transaction + +from case_events import bus, subjects +from case_events.envelope import build_envelope + +logger = structlog.get_logger(__name__) + +#: What this producer calls itself in the envelope. +PRODUCER = "platform" + +_SUBJECT_BY_STATUS = { + "approved": subjects.CASE_UPDATE_APPROVED, + "rejected": subjects.CASE_UPDATE_REJECTED, +} + + +def _case_iri(slug: str) -> str: + """The case's ``@id``, or "" if it can't be built. + + Never raises: a malformed slug must not stop the event, and a missing ref is + a degraded message rather than no message. + """ + try: + from jawafdehi_shared.entities.ids import build_case_iri + + return build_case_iri(slug) + except Exception: # noqa: BLE001 - a bad ref must not cost us the event + logger.warning("case_proposal.case_iri_failed", case_slug=slug) + return "" + + +def build_decision_envelope(proposal) -> dict: + """The envelope for an approve/reject decision on ``proposal``.""" + subject = _SUBJECT_BY_STATUS[proposal.status] + case_iri = _case_iri(proposal.case_slug) + + # The case first, then whatever the producer recorded — deduplicated and + # order-preserving so the join key a consumer needs is at a stable position. + refs = [ref for ref in [case_iri, *(proposal.subject_refs or [])] if ref] + + return build_envelope( + subject=subject, + producer=PRODUCER, + subject_refs=list(dict.fromkeys(refs)), + # Keyed on the DECISION, not the fact. proposal.dedup_key identifies the + # underlying fact and is carried in the payload; this one exists so that + # re-publishing the same decision collapses in JetStream. A proposal is + # decided at most once, so pk + status is genuinely unique. + dedup_key=f"proposal:{proposal.pk}:{proposal.status}", + source=proposal.source, + occurred_at=proposal.reviewed_at, + payload={ + "proposal_id": proposal.pk, + "case_slug": proposal.case_slug, + "case_title": proposal.case_title, + "status": proposal.status, + "intent": proposal.intent, + "confidence": proposal.confidence, + "source_kind": proposal.source_kind, + "detected_by": proposal.detected_by, + "fact_dedup_key": proposal.dedup_key, + "reviewer": proposal.reviewer, + "review_notes": proposal.review_notes, + "origin_msg_id": proposal.origin_msg_id, + }, + ) + + +def schedule_decision_event(proposal) -> None: + """Publish the decision once the surrounding transaction commits. + + Best-effort in both directions: it never raises, and with ``NATS_URL`` unset + it does nothing at all. An approval must succeed whether or not a broker is + reachable — that property is the point, and is worth an explicit test. + """ + if proposal.status not in _SUBJECT_BY_STATUS: + return + + # Snapshot now, publish later: on_commit runs after the transaction, and + # reading a mutated-or-refetched instance then would risk publishing state + # that isn't what was decided. + envelope = build_decision_envelope(proposal) + subject = envelope["subject"] + + def _run(): + try: + bus.publish(subject, envelope) + except Exception: # noqa: BLE001 - the bus is never allowed to be fatal + logger.warning( + "case_proposal.publish_failed", + subject=subject, + proposal_id=envelope["payload"]["proposal_id"], + ) + + transaction.on_commit(_run) diff --git a/case_proposals/tests/test_publish.py b/case_proposals/tests/test_publish.py new file mode 100644 index 00000000..0c7ec5cb --- /dev/null +++ b/case_proposals/tests/test_publish.py @@ -0,0 +1,187 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""The decision publisher, and the guarantee that it can never break an approval. + +Uses ``django_capture_on_commit_callbacks`` so the on_commit hook actually runs +inside the test transaction — without it these would all pass vacuously, which +is the usual way an on_commit path ships broken. +""" + +from unittest import mock + +import pytest +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Group +from django.test import override_settings +from rest_framework.test import APIClient + +from case_proposals.models import CaseUpdateProposal, ProposalStatus +from case_proposals.publish import build_decision_envelope, schedule_decision_event +from cases.models import Case, CaseType + +LIST_URL = "/api/case-update-proposals/" + + +def make_caseworker(): + User = get_user_model() + user = User.objects.create_user(username="u-caseworker", password="x") + group, _ = Group.objects.get_or_create(name="Caseworker") + user.groups.add(group) + return user + + +def caseworker_client(): + client = APIClient() + client.force_authenticate(user=make_caseworker()) + return client + + +def make_case(slug="lalita-niwas-land-scam"): + return Case.objects.create( + title="Lalita Niwas land scam", case_type=CaseType.CORRUPTION, slug=slug + ) + + +def make_proposal(**over): + data = dict( + case_slug="lalita-niwas-land-scam", + case_title="Lalita Niwas land scam", + source_kind="ngm_docket", + intent={ + "type": "append_timeline_entry", + "entry": {"date": "2026-08-12", "title": "Hearing scheduled"}, + }, + confidence=0.97, + detected_by="consumer:proposal-builder", + dedup_key="docket:x:hearing:1", + ) + data.update(over) + return CaseUpdateProposal.objects.create(**data) + + +@pytest.mark.django_db +class TestEnvelope: + def test_approved_and_rejected_map_to_distinct_subjects(self): + approved = make_proposal(status=ProposalStatus.APPROVED) + rejected = make_proposal(status=ProposalStatus.REJECTED, dedup_key="d2") + assert build_decision_envelope(approved)["subject"] == "jaw.case.update.approved" + assert build_decision_envelope(rejected)["subject"] == "jaw.case.update.rejected" + + def test_case_iri_leads_the_subject_refs(self): + p = make_proposal(status=ProposalStatus.APPROVED) + refs = build_decision_envelope(p)["subject_refs"] + assert refs[0] == "https://jawafdehi.org/case/lalita-niwas-land-scam" + + def test_producer_subject_refs_are_preserved_and_deduplicated(self): + docket = "https://jawafdehi.org/courtcase/special/082-cr-0154" + case_iri = "https://jawafdehi.org/case/lalita-niwas-land-scam" + p = make_proposal( + status=ProposalStatus.APPROVED, + subject_refs=[docket, case_iri], # case_iri duplicated on purpose + ) + refs = build_decision_envelope(p)["subject_refs"] + assert refs == [case_iri, docket] + + def test_decision_dedup_key_differs_from_the_fact_dedup_key(self): + # The fact key identifies WHAT happened; the decision key identifies the + # decision, so re-publishing a decision collapses without suppressing a + # genuinely different event about the same fact. + p = make_proposal(status=ProposalStatus.APPROVED, dedup_key="docket:x:hearing:1") + env = build_decision_envelope(p) + assert env["dedup_key"] == f"proposal:{p.pk}:approved" + assert env["payload"]["fact_dedup_key"] == "docket:x:hearing:1" + + def test_payload_carries_the_intent_and_confidence(self): + p = make_proposal(status=ProposalStatus.APPROVED) + payload = build_decision_envelope(p)["payload"] + assert payload["intent"]["type"] == "append_timeline_entry" + assert payload["confidence"] == 0.97 + assert payload["case_slug"] == "lalita-niwas-land-scam" + + def test_a_broken_case_iri_degrades_rather_than_raising(self): + p = make_proposal(status=ProposalStatus.APPROVED) + with mock.patch( + "jawafdehi_shared.entities.ids.build_case_iri", side_effect=ValueError("bad") + ): + env = build_decision_envelope(p) + assert env["subject_refs"] == [] # degraded, but still a message + + def test_pending_proposals_publish_nothing(self): + p = make_proposal(status=ProposalStatus.PENDING) + with mock.patch("case_proposals.publish.transaction.on_commit") as on_commit: + schedule_decision_event(p) + on_commit.assert_not_called() + + +@pytest.mark.django_db +class TestApprovalIsIndependentOfTheBroker: + """The property worth protecting: the archive does not depend on the bus.""" + + @override_settings(NATS_URL="") + def test_approve_succeeds_with_no_broker_configured(self, django_capture_on_commit_callbacks): + make_case() + p = make_proposal() + with django_capture_on_commit_callbacks(execute=True): + r = caseworker_client().post(f"{LIST_URL}{p.id}/approve/", {}, format="json") + + assert r.status_code == 200 + p.refresh_from_db() + assert p.status == ProposalStatus.APPROVED + # And the intent really was applied. + assert len(Case.objects.get(slug="lalita-niwas-land-scam").timeline) == 1 + + @override_settings(NATS_URL="nats://unreachable:4222") + def test_approve_succeeds_when_publishing_blows_up( + self, django_capture_on_commit_callbacks + ): + make_case() + p = make_proposal() + with mock.patch("case_events.bus.publish", side_effect=RuntimeError("broker down")): + with django_capture_on_commit_callbacks(execute=True): + r = caseworker_client().post(f"{LIST_URL}{p.id}/approve/", {}, format="json") + + assert r.status_code == 200 + p.refresh_from_db() + assert p.status == ProposalStatus.APPROVED + assert len(Case.objects.get(slug="lalita-niwas-land-scam").timeline) == 1 + + @override_settings(NATS_URL="nats://localhost:4222") + def test_approve_publishes_the_approved_event(self, django_capture_on_commit_callbacks): + make_case() + p = make_proposal() + with mock.patch("case_events.bus.publish") as publish: + with django_capture_on_commit_callbacks(execute=True): + caseworker_client().post(f"{LIST_URL}{p.id}/approve/", {}, format="json") + + subject, envelope = publish.call_args.args + assert subject == "jaw.case.update.approved" + assert envelope["payload"]["proposal_id"] == p.pk + assert envelope["payload"]["status"] == "approved" + + @override_settings(NATS_URL="nats://localhost:4222") + def test_reject_publishes_the_rejected_event(self, django_capture_on_commit_callbacks): + make_case() + p = make_proposal() + with mock.patch("case_events.bus.publish") as publish: + with django_capture_on_commit_callbacks(execute=True): + caseworker_client().post( + f"{LIST_URL}{p.id}/reject/", {"notes": "wrong person"}, format="json" + ) + + subject, envelope = publish.call_args.args + assert subject == "jaw.case.update.rejected" + assert envelope["payload"]["review_notes"] == "wrong person" + + @override_settings(NATS_URL="nats://localhost:4222") + def test_nothing_is_published_when_the_decision_is_rejected_by_a_409( + self, django_capture_on_commit_callbacks + ): + # An already-decided proposal 409s without touching the case, so it must + # not announce a second decision. + make_case() + p = make_proposal(status=ProposalStatus.APPROVED) + with mock.patch("case_events.bus.publish") as publish: + with django_capture_on_commit_callbacks(execute=True): + r = caseworker_client().post(f"{LIST_URL}{p.id}/approve/", {}, format="json") + + assert r.status_code == 409 + publish.assert_not_called() diff --git a/case_proposals/views.py b/case_proposals/views.py index f1be0acc..e866cd49 100644 --- a/case_proposals/views.py +++ b/case_proposals/views.py @@ -12,6 +12,7 @@ from .apply import apply_intent, get_case_or_400 from .models import CaseUpdateProposal, ProposalStatus +from .publish import schedule_decision_event from .serializers import ( CaseUpdateProposalSerializer, ProposalDecisionSerializer, @@ -84,6 +85,12 @@ def _decide(self, request, proposal, new_status, apply_first): proposal.save( update_fields=["status", "reviewer", "reviewed_at", "review_notes", "updated_at"] ) + # Announce the decision on the bus once this commits. Registered + # INSIDE the atomic block so a rollback discards the callback with + # the transaction; it still fires only after a successful commit. + # Best-effort and a no-op without NATS_URL — an approval must never + # depend on a broker being reachable. + schedule_decision_event(proposal) # Audit the decision: who accepted/rejected (the acceptor) + the exact # proposed change. The DB LogEntry (register_audited) records the actor # on the status transition; this structured line makes the acceptor + diff --git a/config/settings.py b/config/settings.py index 9dff2d40..c784fabe 100644 --- a/config/settings.py +++ b/config/settings.py @@ -308,6 +308,11 @@ def build_media_url( "newsletter", # ── Central job queue (platform-wide; Postgres-backed, no broker) ───────── "jobs", + # ── Case-enrichment event bus (NATS/JetStream; no models, no migrations) ── + # NOT "events": that name is taken by the `Events` dist (a transitive + # dependency of opensearch-py), and a top-level collision breaks the + # installed wheel even though a source checkout shadows it fine. + "case_events", # ── Generic LLM invocation (provider registry: bedrock/proxy/CLI harnesses) ─ "llm", # ── Unified search (platform-wide; queries all three domains' indices) ──── @@ -1110,3 +1115,19 @@ def _sqlite_alias(file_name: str, test_name: str) -> dict: OPENSEARCH_URL = os.getenv("OPENSEARCH_URL", "http://localhost:9200") OPENSEARCH_USER = os.getenv("OPENSEARCH_USER", "") OPENSEARCH_PASSWORD = os.getenv("OPENSEARCH_PASSWORD", "") + +# ============================================================================ +# Case-enrichment event bus — NATS + JetStream +# ============================================================================ +# OPTIONAL, and off by default. When NATS_URL is empty every publish is a logged +# no-op and no connection is ever opened, so the platform runs unchanged with no +# broker: dev and CI need nothing, and this code ships safely before the bus is +# deployed. Setting it is also the whole rollback — no image change required. +# +# Publishing is best-effort by design: a broker outage must never fail a case +# write, because the bus is transport and the case record is the system of +# record. See events/bus.py. +# +# Credentials ride in the URL (nats://user:pass@host:4222) and are per identity, +# not shared — the monolith publishes as itself, and consumers get their own. +NATS_URL = os.getenv("NATS_URL", "") diff --git a/llm/prompt_templates/reference/content.md b/llm/prompt_templates/reference/content.md new file mode 100644 index 00000000..a9d0c780 --- /dev/null +++ b/llm/prompt_templates/reference/content.md @@ -0,0 +1,24 @@ +{# A reference content template, exercised by llm/tests/test_templating.py. + + It deliberately contains the three things that break a naively-configured + template engine, so the tests around it fail loudly if the prompt engine is + ever swapped for the HTML one: + + - HTML tags, which autoescaping would turn into <p> + - JSON with quotes and braces, which autoescaping turns into " + and str.format() chokes on entirely + - Devanagari, which must survive as-is + + Loops and conditionals are for shaping data into prose. Computation — + json.dumps, character caps, settings lookups — belongs in Python, and the + result gets passed in as context. #} +CASE: {{ case_title }} + +{% if excerpts %}SOURCE EXCERPTS: +{% for excerpt in excerpts %}- {{ excerpt }} +{% endfor %}{% else %}No source excerpts were available. +{% endif %} +Summarise the case in one sentence. Use for the accused party's name. + +Reply EXACTLY in this JSON shape: +{"summary": "", "confidence": } diff --git a/llm/prompt_templates/reference/system.md b/llm/prompt_templates/reference/system.md new file mode 100644 index 00000000..990961d7 --- /dev/null +++ b/llm/prompt_templates/reference/system.md @@ -0,0 +1,10 @@ +{# A reference system prompt. Copy this pair when adding a real one. + + Django comment tags are stripped from the output, so notes like this one + cost nothing at inference time and never reach the model. #} +You are a meticulous research assistant for Jawafdehi.org, an open civic archive +of Nepali anti-corruption cases. + +{% if language == "np" %}Reply in Nepali (नेपाली भाषामा जवाफ दिनुहोस्).{% else %}Reply in English.{% endif %} + +Reply with a single valid JSON object and nothing else. diff --git a/llm/prompts.py b/llm/prompts.py new file mode 100644 index 00000000..379df5a1 --- /dev/null +++ b/llm/prompts.py @@ -0,0 +1,205 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""A named, versioned registry of prompts backed by template files. + +Prompts in this codebase are conventionally a module-level system constant plus +a ``_build_*`` function that assembles the content block, handed to +:func:`llm.invoke.invoke_json` with a tier and a token budget (see +``review/judge.py``). That works, but it leaves the prompt anonymous — when an +LLM-produced record later turns out to be wrong, "which prompt produced this, +and has it changed since?" has no answer — and it leaves the text itself buried +in Python, where reviewing a wording change means reading a diff of an f-string. + +A :class:`PromptSpec` is a name, a version, two template files (system and +content) and the parameters to invoke them with. The text lives in +``/prompt_templates/`` and renders through :mod:`llm.templating`, which is +a dedicated non-autoescaping, strict-variable engine — see that module for why +the HTML engine would silently corrupt a prompt. + +Templates hold *text*, not computation. Anything that needs Python — a +``json.dumps``, a character cap, a settings lookup — is done by the caller and +passed in as context. ``{% for %}`` and ``{% if %}`` are available for shaping +that data into prose, and that is the intended limit. + +This is deliberately NOT a DB-backed CMS and not an abstraction over +:mod:`llm.invoke`; ``invoke_json`` remains the thing that talks to a model, and +a spec just remembers what to pass it. + +Migrating ``review/judge.py`` and the ``casework/enrich_*`` prompts onto this is +explicitly out of scope for now: they predate the registry, they work, and +several of their tests assert on the Python constants directly. +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +import structlog + +from llm.invoke import invoke_json +from llm.templating import render_prompt + +logger = structlog.get_logger(__name__) + +#: The tiers ``llm.routing`` knows about. Worth spelling out because +#: ``provider_for_tier`` resolves "premium" and treats EVERYTHING ELSE as cheap — +#: so a typo ("premuim", "strong") silently downgrades the model rather than +#: raising. Specs that require a strong model would fail quietly and +#: intermittently, which is the worst way for this to go wrong. Hence the +#: validation in :meth:`PromptSpec.__post_init__`. +TIERS = ("premium", "cheap") + + +@dataclass(frozen=True) +class PromptSpec: + """One named, versioned prompt and the parameters it is invoked with. + + Args: + name: Stable dotted identifier, e.g. ``"case_proposal.intent"``. This is + what gets logged alongside the result, so it should not change once + anything has been produced under it. + version: Bump on ANY wording change, including edits to the template + files. Recorded with every invocation so a bad output can be traced + back to the exact prompt that produced it. + system_template: Path to the system-prompt template, relative to a + ``prompt_templates/`` directory. + content_template: Path to the user-content template. + tier: ``"premium"`` or ``"cheap"``. Validated, because a wrong value + downgrades silently rather than raising. + max_tokens: Response budget handed to ``invoke_json``. + required: Context keys that must be present when rendering. Only needed + for variables used *exclusively* inside ``{% if %}`` / ``{% for %}``, + which Django resolves to falsy without flagging them as missing — + plain ``{{ var }}`` holes are caught automatically. + """ + + name: str + version: int + system_template: str + content_template: str + tier: str = "premium" + max_tokens: int = 1500 + required: tuple[str, ...] = () + + def __post_init__(self): + if not self.name: + raise ValueError("PromptSpec.name is required.") + if self.version < 1: + raise ValueError(f"{self.name}: version must be >= 1, got {self.version!r}.") + if not self.system_template.strip(): + raise ValueError(f"{self.name}: system_template is required.") + if not self.content_template.strip(): + raise ValueError(f"{self.name}: content_template is required.") + if self.tier not in TIERS: + raise ValueError( + f"{self.name}: unknown tier {self.tier!r}. Known: {list(TIERS)}. " + "An unknown tier would route to the CHEAP model without raising." + ) + if self.max_tokens < 1: + raise ValueError(f"{self.name}: max_tokens must be >= 1, got {self.max_tokens!r}.") + + def render_system(self, **context) -> str: + """Render the system prompt. + + Takes context too: a system prompt is often parameterised (the case + scraper's switches its whole output language on one flag), and forcing + that into the content block would put it further from the instruction it + modifies. + """ + return render_prompt(self.system_template, context) + + def render(self, **context) -> str: + """Render the content block without invoking a model. + + The seam tests and prompt-review tooling hang off: it makes the exact + text sent to the model assertable without spending a call. + """ + return render_prompt(self.content_template, context, required=self.required) + + def invoke(self, usage=None, **context) -> Any: + """Render both templates and invoke the model, returning parsed JSON. + + Thin by design — ``invoke_json`` already salvages dirty/truncated output, + and re-implementing any of that here would put two behaviours in the + codebase where callers expect one. + + Both templates get the same context, so a value needed by each is passed + once. + + Args: + usage: Optional UsageAccumulator, forwarded to ``invoke_json``. + **context: Template context. + + Returns: + Parsed JSON (dict or list), per ``invoke_json``. + + Raises: + llm.templating.PromptRenderError: raised BEFORE any model call if + either template is missing or a variable did not resolve, so a + broken prompt costs nothing. + """ + system = self.render_system(**context) + content = self.render(**context) + # Logged BEFORE the call as well as after, so a spec that reliably times + # out or blows its token budget is still attributable to a version. + logger.info( + "prompt.invoke", + prompt=self.name, + version=self.version, + tier=self.tier, + max_tokens=self.max_tokens, + content_chars=len(content), + ) + return invoke_json( + system, + content, + max_tokens=self.max_tokens, + tier=self.tier, + usage=usage, + ) + + +_REGISTRY: dict[str, PromptSpec] = {} + + +def register(spec: PromptSpec) -> PromptSpec: + """Register ``spec`` under its name, replacing any existing entry. + + Replacement is allowed because registration happens at import time and + modules can be re-imported (notably under the test runner), so a strict + "already registered" error would be a false alarm far more often than a real + catch. Returns the spec so it can be assigned at module level in one line. + """ + _REGISTRY[spec.name] = spec + return spec + + +def get(name: str) -> PromptSpec: + """Return the spec registered under ``name``. + + Raises: + KeyError: if nothing is registered under that name. + + Note this deliberately differs from ``jobs.registry.get``, which returns a + default spec for an unregistered kind. A job kind has a sensible default + policy; a prompt does not — there is no "default prompt", and silently + invoking the wrong text is worse than failing. So this raises. + """ + try: + return _REGISTRY[name] + except KeyError: + raise KeyError( + f"No prompt registered as {name!r}. Registered: {known()}. " + "Prompts register at import time — check the owning app is in " + "INSTALLED_APPS and its registration module is imported." + ) from None + + +def known() -> list[str]: + """Every registered prompt name, sorted.""" + return sorted(_REGISTRY) + + +def all_specs() -> list[PromptSpec]: + """Every registered spec, ordered by name.""" + return [_REGISTRY[name] for name in known()] diff --git a/llm/templating.py b/llm/templating.py new file mode 100644 index 00000000..494bb199 --- /dev/null +++ b/llm/templating.py @@ -0,0 +1,174 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Loading prompt text from template files. + +Prompt text lives in ``/prompt_templates/*.md`` rather than in Python +string constants, so it can be read, diffed and reviewed as prose instead of as +an f-string with the indentation fought into shape. + +This uses Django's template engine, but a **dedicated instance** — not the one +in ``settings.TEMPLATES`` that renders HTML. Two of its defaults are actively +wrong for prompts, and both fail silently: + +**Autoescaping must be off.** Precisely: Django escapes *interpolated values*, +not the literal text of a template — so the danger is not the prompt wording, it +is the data. And prompt context is exactly the wrong shape for it. The judge +path passes ``json.dumps(case_summary, indent=2)``, which under autoescaping +arrives at the model as a wall of ``"`` instead of JSON; source excerpts +carry quotes and ampersands; case titles carry ``&``. The result is a prompt +that still looks like a prompt, produces plausible-but-wrong output, and raises +nothing. + +(Literal prompt text is safe either way. That asymmetry is easy to test wrong — +a template whose HTML sits in the *body* passes with escaping switched on, which +is why the tests here put the hostile characters in the context values.) + +**A missing variable must not render as empty.** Django's default is to swallow +an unknown variable and emit ``""``. For a prompt that means a renamed context +key silently ships a prompt with a hole in it and you get a bad answer rather +than a crash. So the engine is configured with a sentinel +``string_if_invalid``, and :func:`render_prompt` refuses to return any string +still containing it. + +That sentinel closes the ``{{ var }}`` case but **not** the tag case: Django +never consults ``string_if_invalid`` for ``{% if missing %}`` (falsy) or +``{% for x in missing %}`` (empty). A variable used *only* inside a tag must +therefore be declared in ``required=``, which is checked before rendering. + +One consequence of the sentinel worth knowing: ``{{ x|default:"unknown" }}`` +does not rescue an *absent* ``x`` — Django substitutes the sentinel without +applying filters. ``default`` still works for a key that is present but empty, +which is the coherent reading anyway: pass the key, and let the filter handle +the empty case. +""" + +from __future__ import annotations + +import re +from pathlib import Path +from typing import Any, Iterable + +from django.apps import apps +from django.template import Context, Engine, TemplateDoesNotExist + +# Deliberately no logger here. Rendering happens on the way to a model call that +# llm.prompts already logs (name + version + size), and anything this module +# could usefully log is the prompt text itself — which carries case content and +# does not belong in the log stream. + +#: Directory name, relative to an app package, holding that app's prompt +#: templates. NOT ``prompts``: ``llm/prompts.py`` is the registry module, and a +#: module and a package of the same name cannot coexist in one package. +PROMPT_DIR_NAME = "prompt_templates" + +#: Emitted in place of an unresolvable variable. NUL-delimited so it cannot +#: collide with anything a real template or a real context value contains. The +#: ``%s`` is Django's own convention — it substitutes the variable's name, which +#: is what makes the resulting error actionable rather than just "something was +#: missing". +MISSING_SENTINEL = "\x00prompt-missing:%s\x00" + +_MISSING_RE = re.compile("\x00prompt-missing:(.*?)\x00") + +_engine: Engine | None = None + + +class PromptRenderError(Exception): + """A prompt could not be rendered into text safe to send to a model.""" + + +def prompt_template_dirs() -> list[Path]: + """Every installed app's ``prompt_templates/`` directory that exists. + + Discovered from the app registry rather than hardcoded, so an app owns its + own prompts and adding one is a directory, not a settings edit. + """ + dirs = [] + for config in apps.get_app_configs(): + candidate = Path(config.path) / PROMPT_DIR_NAME + if candidate.is_dir(): + dirs.append(candidate) + return dirs + + +def get_engine() -> Engine: + """The process-wide prompt engine, built on first use and cached. + + Built lazily because it reads the app registry, which is not populated at + import time. + """ + global _engine + if _engine is None: + _engine = Engine( + dirs=[str(d) for d in prompt_template_dirs()], + # Templates are found by explicit dirs only. app_dirs would look in + # /templates/, which is the HTML engine's territory. + app_dirs=False, + autoescape=False, + string_if_invalid=MISSING_SENTINEL, + ) + return _engine + + +def reset_engine() -> None: + """Drop the cached engine so the next call rediscovers directories.""" + global _engine + _engine = None + + +def render_prompt( + template_name: str, + context: dict[str, Any] | None = None, + *, + required: Iterable[str] = (), +) -> str: + """Render ``template_name`` to prompt text. + + Args: + template_name: Path relative to a ``prompt_templates/`` directory, e.g. + ``"case_proposal/intent.content.md"``. + context: Template context. + required: Context keys that must be present. Needed for variables used + only inside ``{% if %}`` / ``{% for %}``, where Django resolves a + missing name to falsy without consulting ``string_if_invalid``. + + Returns: + The rendered text, with trailing whitespace stripped. + + Raises: + PromptRenderError: if the template is missing, a required key is absent, + or any variable failed to resolve. Never returns a partially-filled + prompt — sending one to a model is worse than failing, because the + failure would surface later as a bad record with no obvious cause. + """ + context = dict(context or {}) + + missing_required = [key for key in required if key not in context] + if missing_required: + raise PromptRenderError( + f"{template_name}: missing required context {sorted(missing_required)}. " + f"Got {sorted(context)}." + ) + + engine = get_engine() + try: + template = engine.get_template(template_name) + except TemplateDoesNotExist as exc: + raise PromptRenderError( + f"No prompt template {template_name!r} in any of " + f"{[str(d) for d in prompt_template_dirs()]}. Prompt templates live in " + f"/{PROMPT_DIR_NAME}/ and the app must be in INSTALLED_APPS." + ) from exc + + # autoescape is a property of the Context, not only of the Engine — building + # a bare Context() here would re-enable escaping and quietly undo the point + # of this module. + rendered = template.render(Context(context, autoescape=engine.autoescape)) + + unresolved = _MISSING_RE.findall(rendered) + if unresolved: + raise PromptRenderError( + f"{template_name}: unresolved template variables {sorted(set(unresolved))}. " + f"Context had {sorted(context)}. A prompt is never rendered with holes in it." + ) + + return rendered.rstrip() diff --git a/llm/tests/__init__.py b/llm/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/llm/tests/test_prompts.py b/llm/tests/test_prompts.py new file mode 100644 index 00000000..4187b55a --- /dev/null +++ b/llm/tests/test_prompts.py @@ -0,0 +1,205 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Tests for the PromptSpec registry. + +No DB and no model calls: every test either exercises validation or asserts on +what would have been passed to ``invoke_json``, which is mocked. The registry is +module-level global state, so tests that register anything clean up after +themselves. + +Rendering itself is covered in test_templating.py; here it only matters that a +spec wires the right template to the right invoke parameters. +""" + +from unittest import mock + +import pytest + +from llm import prompts, templating +from llm.prompts import PromptSpec +from llm.templating import PromptRenderError + +REFERENCE_SYSTEM = "reference/system.md" +REFERENCE_CONTENT = "reference/content.md" + + +def _spec(**kw): + """A valid spec, overridable per-test.""" + defaults = dict( + name="test.spec", + version=1, + system_template=REFERENCE_SYSTEM, + content_template=REFERENCE_CONTENT, + ) + return PromptSpec(**{**defaults, **kw}) + + +@pytest.fixture(autouse=True) +def _clean_registry(): + """Snapshot/restore the registry so tests can't leak into each other.""" + before = dict(prompts._REGISTRY) + yield + prompts._REGISTRY.clear() + prompts._REGISTRY.update(before) + + +@pytest.fixture(autouse=True) +def _reset_engine(): + templating.reset_engine() + yield + templating.reset_engine() + + +class TestValidation: + def test_valid_spec_constructs(self): + assert _spec().tier == "premium" + + def test_unknown_tier_raises(self): + # The point of the guard: llm.routing.provider_for_tier resolves + # "premium" and treats everything else as CHEAP, so a typo would + # silently downgrade the model instead of failing. + with pytest.raises(ValueError, match="unknown tier"): + _spec(tier="premuim") + + def test_cheap_tier_is_allowed(self): + assert _spec(tier="cheap").tier == "cheap" + + @pytest.mark.parametrize( + "kwargs,match", + [ + ({"name": ""}, "name is required"), + ({"version": 0}, "version must be >= 1"), + ({"system_template": " "}, "system_template is required"), + ({"content_template": ""}, "content_template is required"), + ({"max_tokens": 0}, "max_tokens must be >= 1"), + ], + ) + def test_rejects_bad_field(self, kwargs, match): + with pytest.raises(ValueError, match=match): + _spec(**kwargs) + + def test_spec_is_frozen(self): + spec = _spec() + with pytest.raises(Exception): + spec.version = 2 + + def test_a_nonexistent_template_is_not_caught_until_render(self): + # Construction cannot check the filesystem: specs are built at import + # time, before the app registry the loader dirs come from is populated. + spec = _spec(content_template="no/such.md") + with pytest.raises(PromptRenderError, match="no/such.md"): + spec.render(case_title="X", excerpts=[]) + + +class TestRegistry: + def test_register_then_get(self): + spec = prompts.register(_spec(name="a.b")) + assert prompts.get("a.b") is spec + + def test_register_returns_the_spec(self): + # So a module can do: SPEC = register(PromptSpec(...)) in one statement. + spec = _spec(name="a.b") + assert prompts.register(spec) is spec + + def test_register_replaces(self): + prompts.register(_spec(name="a.b", version=1)) + prompts.register(_spec(name="a.b", version=2)) + assert prompts.get("a.b").version == 2 + + def test_get_unknown_raises_with_the_known_names(self): + prompts.register(_spec(name="a.b")) + with pytest.raises(KeyError) as exc: + prompts.get("nope") + # The error should be actionable, not just "KeyError: 'nope'". + assert "a.b" in str(exc.value) + + def test_known_is_sorted(self): + prompts.register(_spec(name="z.z")) + prompts.register(_spec(name="a.a")) + assert prompts.known().index("a.a") < prompts.known().index("z.z") + + def test_all_specs_follows_known_order(self): + prompts._REGISTRY.clear() + prompts.register(_spec(name="z.z")) + prompts.register(_spec(name="a.a")) + assert [s.name for s in prompts.all_specs()] == ["a.a", "z.z"] + + +class TestInvoke: + def test_render_does_not_call_a_model(self): + with mock.patch("llm.prompts.invoke_json") as invoke: + out = _spec().render(case_title="Lalita Niwas", excerpts=[]) + assert "CASE: Lalita Niwas" in out + invoke.assert_not_called() + + def test_invoke_passes_rendered_system_content_and_params(self): + spec = _spec(tier="cheap", max_tokens=321) + with mock.patch("llm.prompts.invoke_json", return_value={"ok": True}) as invoke: + out = spec.invoke(case_title="Lalita Niwas", excerpts=["e1"], language="en") + + assert out == {"ok": True} + system, content = invoke.call_args.args + assert "Reply in English." in system + assert "CASE: Lalita Niwas" in content + assert "- e1" in content + assert invoke.call_args.kwargs["tier"] == "cheap" + assert invoke.call_args.kwargs["max_tokens"] == 321 + + def test_both_templates_receive_the_same_context(self): + # One context feeds system and content, so a shared value is passed once. + with mock.patch("llm.prompts.invoke_json", return_value={}) as invoke: + _spec().invoke(case_title="X", excerpts=[], language="np") + system, _content = invoke.call_args.args + assert "नेपाली भाषामा" in system + + def test_invoke_forwards_usage(self): + sentinel = object() + with mock.patch("llm.prompts.invoke_json", return_value={}) as invoke: + _spec().invoke(usage=sentinel, case_title="X", excerpts=[]) + assert invoke.call_args.kwargs["usage"] is sentinel + + def test_usage_is_not_treated_as_template_context(self): + # `usage` is an invoke_json concern; leaking it into the context would + # put a repr of an accumulator object into the prompt. + with mock.patch("llm.prompts.invoke_json", return_value={}) as invoke: + _spec().invoke(usage=object(), case_title="X", excerpts=[]) + _system, content = invoke.call_args.args + assert "usage" not in content + + def test_a_render_failure_costs_no_model_call(self): + # The whole reason rendering is strict: fail before spending a call, + # not after producing a record from a prompt with a hole in it. + with mock.patch("llm.prompts.invoke_json") as invoke: + with pytest.raises(PromptRenderError): + _spec().invoke(excerpts=[]) + invoke.assert_not_called() + + def test_invoke_logs_name_and_version(self): + # "Which prompt version produced this?" must be answerable from logs. + spec = _spec(name="case_proposal.intent", version=7) + with mock.patch("llm.prompts.invoke_json", return_value={}): + with mock.patch.object(prompts.logger, "info") as log: + spec.invoke(case_title="X", excerpts=[]) + + kwargs = log.call_args.kwargs + assert kwargs["prompt"] == "case_proposal.intent" + assert kwargs["version"] == 7 + + +class TestRegisteredSpecsAreLoadable: + def test_every_registered_spec_has_templates_that_exist(self): + """Catches a spec pointing at a template that was renamed or never shipped. + + **Vacuous as of this commit**, and deliberately kept anyway: nothing + registers a spec until the enrichment consumers land, so ``all_specs()`` + is empty and this loop does nothing. It is a forward guard that starts + working the moment the first real spec is registered. + + What carries the weight *today* is ``TestTheReferenceTemplates``, which + renders actual files off disk through the engine — that is what would + fail if prompt templates were excluded from the wheel or the image, the + same class of omission that already shipped a missing app directory once. + """ + engine = templating.get_engine() + for spec in prompts.all_specs(): + for template in (spec.system_template, spec.content_template): + engine.get_template(template) # raises TemplateDoesNotExist diff --git a/llm/tests/test_templating.py b/llm/tests/test_templating.py new file mode 100644 index 00000000..2f88454f --- /dev/null +++ b/llm/tests/test_templating.py @@ -0,0 +1,183 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Tests for the prompt template engine. + +The load-bearing ones are the two silent-corruption guards: that autoescaping +is off, and that an unresolved variable raises instead of rendering as empty. +Both failure modes produce a prompt that still *looks* like a prompt, so +nothing downstream would notice — the model would just answer a slightly +different question and we would blame the model. +""" + +import pytest + +from llm import templating +from llm.templating import PromptRenderError, render_prompt + +REFERENCE_SYSTEM = "reference/system.md" +REFERENCE_CONTENT = "reference/content.md" + + +@pytest.fixture(autouse=True) +def _reset_engine(): + """The engine is a cached module global; don't leak one test's dirs.""" + templating.reset_engine() + yield + templating.reset_engine() + + +@pytest.fixture +def templates(tmp_path, monkeypatch): + """Render from a throwaway directory instead of a real app's.""" + + def write(name, body): + path = tmp_path / name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(body, encoding="utf-8") + + monkeypatch.setattr(templating, "prompt_template_dirs", lambda: [tmp_path]) + templating.reset_engine() + return write + + +class TestAutoescapeIsOff: + """Prompt *data* routinely contains the characters autoescaping mangles. + + Every test here puts the hostile characters in a context VALUE, never in + the template body. Django only escapes interpolated values, so a test with + HTML in the body passes with escaping switched on and proves nothing — + these all fail if the engine is ever misconfigured. + """ + + def test_json_context_survives_intact(self, templates): + # The realistic case: review/judge.py hands the model + # json.dumps(case_summary, indent=2). Escaped, that reaches the model as + # a wall of " and stops being JSON. + import json + + templates("t.md", "CASE DATA:\n{{ case_json }}") + case_json = json.dumps({"title": "Lalita Niwas", "amount": "NPR 10,00,000"}, indent=2) + assert render_prompt("t.md", {"case_json": case_json}).endswith(case_json) + + def test_quotes_in_a_value_survive(self, templates): + templates("t.md", 'Reply EXACTLY: {"note": "{{ note }}"}') + assert render_prompt("t.md", {"note": 'he said "no"'}) == ( + 'Reply EXACTLY: {"note": "he said "no""}' + ) + + def test_html_in_a_value_survives(self, templates): + templates("t.md", "EXCERPT:\n{{ excerpt }}") + excerpt = "

The accused denied it.

" + assert render_prompt("t.md", {"excerpt": excerpt}) == f"EXCERPT:\n{excerpt}" + + def test_ampersands_and_angle_brackets_in_context_survive(self, templates): + # The value, not just the template, must come through unescaped. + templates("t.md", "{{ v }}") + assert render_prompt("t.md", {"v": "R&D 'quoted'"}) == "R&D 'quoted'" + + def test_devanagari_survives(self, templates): + templates("t.md", "{{ v }}") + agency = "अख्तियार दुरुपयोग अनुसन्धान आयोग" + assert render_prompt("t.md", {"v": agency}) == agency + + +class TestMissingVariablesRaise: + def test_unresolved_variable_raises_and_names_it(self, templates): + templates("t.md", "Case: {{ case_title }}") + with pytest.raises(PromptRenderError) as exc: + render_prompt("t.md", {}) + assert "case_title" in str(exc.value) + + def test_the_sentinel_never_leaks_into_output(self, templates): + templates("t.md", "Case: {{ nope }}") + with pytest.raises(PromptRenderError): + render_prompt("t.md", {}) + + def test_unresolved_dotted_path_raises(self, templates): + templates("t.md", "{{ case.title }}") + with pytest.raises(PromptRenderError, match="case.title"): + render_prompt("t.md", {"case": {}}) + + def test_present_but_empty_is_not_missing(self, templates): + # An empty string is a legitimate value; only absence is an error. + templates("t.md", "[{{ v }}]") + assert render_prompt("t.md", {"v": ""}) == "[]" + + def test_required_key_absent_raises_before_rendering(self, templates): + templates("t.md", "{% if flag %}on{% else %}off{% endif %}") + with pytest.raises(PromptRenderError, match="flag"): + render_prompt("t.md", {}, required=["flag"]) + + def test_required_closes_the_tag_shaped_hole(self, templates): + # Django resolves a missing name inside {% if %} / {% for %} to falsy + # WITHOUT consulting string_if_invalid, so the sentinel cannot see it. + # This test documents that gap and proves `required` is what covers it. + templates("t.md", "{% if flag %}on{% else %}off{% endif %}") + assert render_prompt("t.md", {}) == "off" # silently wrong, and allowed + with pytest.raises(PromptRenderError): + render_prompt("t.md", {}, required=["flag"]) + + def test_missing_template_names_the_search_path(self, templates): + templates("other.md", "x") + with pytest.raises(PromptRenderError) as exc: + render_prompt("absent.md", {}) + assert "absent.md" in str(exc.value) + assert "prompt_templates" in str(exc.value) + + +class TestRendering: + def test_loops_and_conditionals_shape_data_into_prose(self, templates): + templates( + "t.md", + "{% for r in rules %}- {{ r.title }}\n{% empty %}none\n{% endfor %}", + ) + out = render_prompt("t.md", {"rules": [{"title": "A"}, {"title": "B"}]}) + assert out == "- A\n- B" + + def test_comments_are_stripped(self, templates): + templates("t.md", "{# a note for humans #}text") + assert render_prompt("t.md", {}) == "text" + + def test_trailing_whitespace_is_stripped(self, templates): + templates("t.md", "text\n\n\n") + assert render_prompt("t.md", {}) == "text" + + +class TestDiscovery: + def test_finds_the_llm_apps_prompt_templates_directory(self): + # No monkeypatching here: this asserts the real on-disk convention works. + dirs = [str(d) for d in templating.prompt_template_dirs()] + assert any(d.endswith("llm/prompt_templates") for d in dirs), dirs + + def test_the_engine_is_not_the_html_engine(self): + from django.template.loader import engines + + assert templating.get_engine().autoescape is False + # The configured HTML engine still escapes; we did not change it. + assert engines["django"].engine.autoescape is True + + +class TestTheReferenceTemplates: + """The shipped reference pair renders, and proves the guards end to end.""" + + def test_content_renders_with_excerpts(self): + out = render_prompt( + REFERENCE_CONTENT, + {"case_title": "Lalita Niwas", "excerpts": ["first", "second"]}, + ) + assert "CASE: Lalita Niwas" in out + assert "- first" in out and "- second" in out + # The JSON shape and the HTML tag both survive unescaped. + assert '{"summary": "", "confidence": }' in out + assert "" in out + + def test_content_renders_without_excerpts(self): + out = render_prompt(REFERENCE_CONTENT, {"case_title": "X", "excerpts": []}) + assert "No source excerpts were available." in out + + def test_system_switches_language_and_keeps_devanagari(self): + assert "नेपाली भाषामा" in render_prompt(REFERENCE_SYSTEM, {"language": "np"}) + assert "Reply in English." in render_prompt(REFERENCE_SYSTEM, {"language": "en"}) + + def test_content_still_fails_loudly_without_its_variable(self): + with pytest.raises(PromptRenderError, match="case_title"): + render_prompt(REFERENCE_CONTENT, {"excerpts": []}) diff --git a/pyproject.toml b/pyproject.toml index fd2f27ee..450e8bf6 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -45,6 +45,11 @@ dependencies = [ # HTTP client for outbound REST calls (newsletter→SendPulse ESP; also used by # the cases enrichment management commands). "requests>=2.32", + # Client for the OPTIONAL case-enrichment event bus. Pure Python and small. + # Imported lazily inside case_events/bus.py, never at startup, and unused entirely + # when NATS_URL is unset — but it ships in the base image because the + # consumer Deployment runs off that image. + "nats-py>=2.9", # --- entities/materials (NES + NGM) --- "jsonpatch>=1.33", # RFC-6902 PATCH /api/entities/{ref} # --- ngm lakehouse / object storage --- @@ -115,6 +120,7 @@ packages = [ "review", "newsletter", "jobs", + "case_events", "llm", "search", "discovery", diff --git a/tests/test_app_package_names.py b/tests/test_app_package_names.py new file mode 100644 index 00000000..3a495cda --- /dev/null +++ b/tests/test_app_package_names.py @@ -0,0 +1,115 @@ +# SPDX-License-Identifier: Hippocratic-3.0 +"""Adding a Django app means editing three lists. These tests keep them in sync. + +An app has to be named in ``INSTALLED_APPS``, in the wheel's ``packages`` list, +and in the ``Dockerfile``'s explicit ``COPY`` block. **None of those fail at test +time** — the unit suite runs from a source checkout where every directory is +importable regardless — so getting one wrong produces a green build and an image +that dies on startup. + +Both checks here come from one incident. A new app was added to +``INSTALLED_APPS`` and to the wheel packages, but not to the ``Dockerfile``, so +it was simply absent from the image. It was also named ``events``, which is a +top-level name already shipped by the ``Events`` distribution that +``opensearch-py`` pulls in transitively — so instead of a clean +``ModuleNotFoundError``, ``import events`` silently resolved to the *dependency*, +and the failure surfaced as a baffling ImportError deep in an unrelated module. + +Two distinct problems, and both are worth blocking: + +- the missing ``COPY`` was the trigger, and :func:`test_every_first_party_app_is_copied_into_the_image` catches it; +- the name collision was what made it hard to read, and it stays latent even + when everything resolves correctly today, because a transitive dependency can + claim a bare noun in any future lockfile bump. +""" + +import re +from importlib.metadata import packages_distributions + +from django.conf import settings + +#: The distribution this project itself installs as. Packages it owns are not +#: collisions with themselves. +OWN_DISTRIBUTION = "jawafdehi" + + +def _first_party_apps(): + """Apps whose package directory lives in this repo. + + Defined by what is on disk rather than by a name prefix: a third-party app + that happens to share a name with its own distribution (``auditlog`` ships + from ``django-auditlog``, ``corsheaders`` from ``django-cors-headers``) is + not a collision — it is the same package, correctly resolved. Only a + directory we ship can shadow something. + """ + return [ + app + for app in settings.INSTALLED_APPS + if "." not in app and (settings.BASE_DIR / app / "__init__.py").exists() + ] + + +def test_no_app_shadows_an_installed_distribution(): + owners = packages_distributions() + collisions = {} + for app in _first_party_apps(): + dists = [d for d in owners.get(app, []) if d.lower() != OWN_DISTRIBUTION] + if dists: + collisions[app] = dists + + assert not collisions, ( + f"These INSTALLED_APPS share a top-level import name with an installed " + f"dependency: {collisions}. That works from a source checkout (the repo " + f"root shadows site-packages) but breaks the installed wheel, where both " + f"land in site-packages. Rename the app." + ) + + +def test_every_first_party_app_is_copied_into_the_image(): + """The Dockerfile's COPY list must name every first-party app. + + The list is explicit rather than a single ``COPY . .`` (that would drag the + venv, .git and test fixtures into the image), which means it silently drifts. + An app missing here is absent from the image entirely — the build only fails + later, at ``collectstatic`` or ``migrate``, with an error that points at the + importer rather than at the omission. + """ + dockerfile = (settings.BASE_DIR / "Dockerfile").read_text() + copied = set(re.findall(r"^COPY\s+([A-Za-z_][A-Za-z0-9_]*)/\s", dockerfile, re.MULTILINE)) + + missing = [app for app in _first_party_apps() if app not in copied] + assert not missing, ( + f"These apps are in INSTALLED_APPS but are never COPYed into the image: " + f"{missing}. Add `COPY {missing[0]}/ ./{missing[0]}/` to the Dockerfile. " + f"Check the wheel `packages` list in pyproject.toml too." + ) + + +def test_every_first_party_app_ships_in_the_wheel(): + """The same app list, third copy: hatchling's explicit ``packages``.""" + pyproject = (settings.BASE_DIR / "pyproject.toml").read_text() + block = pyproject.partition("[tool.hatch.build.targets.wheel]")[2] + packaged = set(re.findall(r'"([A-Za-z_][A-Za-z0-9_]*)"', block.partition("]")[0])) + + missing = [app for app in _first_party_apps() if app not in packaged] + assert not missing, ( + f"These apps are in INSTALLED_APPS but not in the wheel `packages` list: " + f"{missing}. They would be missing from the installed package." + ) + + +def test_the_check_can_actually_detect_a_collision(): + """Guard the guard. + + ``packages_distributions()`` returning an empty-ish map (wrong interpreter, + no dist-info, a future stdlib change) would make the test above pass + vacuously forever. ``events`` is a known-colliding name from a real transitive + dependency, so if the mechanism works at all it must see this one. + """ + owners = packages_distributions() + assert owners.get("events"), ( + "Expected the `Events` distribution (transitive via opensearch-py) to " + "claim the top-level name `events`. If this dependency is genuinely gone, " + "swap in another known top-level name rather than deleting the check — " + "otherwise the collision test above silently stops testing anything." + ) diff --git a/uv.lock b/uv.lock index 723c7d7f..f2daef7c 100644 --- a/uv.lock +++ b/uv.lock @@ -983,6 +983,7 @@ dependencies = [ { name = "lxml" }, { name = "markdown" }, { name = "mozilla-django-oidc" }, + { name = "nats-py" }, { name = "nepali" }, { name = "openai" }, { name = "opensearch-py" }, @@ -1043,6 +1044,7 @@ requires-dist = [ { name = "markdown", specifier = ">=3.5" }, { name = "markitdown", extras = ["docx", "pdf", "pptx", "xlsx"], marker = "extra == 'bigo-enrichment'", specifier = ">=0.1.5" }, { name = "mozilla-django-oidc", specifier = ">=5.0.2" }, + { name = "nats-py", specifier = ">=2.9" }, { name = "nepali", specifier = ">=1.2.0" }, { name = "openai", specifier = ">=2.30,<3" }, { name = "opensearch-py", specifier = ">=2.4" }, @@ -1475,6 +1477,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/43/e3/7d92a15f894aa0c9c4b49b8ee9ac9850d6e63b03c9c32c0367a13ae62209/mpmath-1.3.0-py3-none-any.whl", hash = "sha256:a0b2b9fe80bbcd81a6647ff13108738cfb482d481d826cc0e02f5b35e5c88d2c", size = 536198, upload-time = "2023-03-07T16:47:09.197Z" }, ] +[[package]] +name = "nats-py" +version = "2.15.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/02/f0/fc5e93f2b0dd14a202590ad9d30eda1955ea872039b5204357348d0f4b1e/nats_py-2.15.0.tar.gz", hash = "sha256:6622c547d9a7d2313d9c147d46c386188f4ec2c7b5c9f9a0438a4d1b55f54a93", size = 75995, upload-time = "2026-06-05T07:34:03.904Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/db/a8/b55606c7c621fb813c8ec78baf201d2c78bf6051091ec0c7ada572999e95/nats_py-2.15.0-py3-none-any.whl", hash = "sha256:9f8d36aa52a9926a88b8f1d70cf1fdce0ad387941479b500ee9ab3e51073cefd", size = 90334, upload-time = "2026-06-05T07:34:02.81Z" }, +] + [[package]] name = "nepali" version = "1.2.0"