feat(events): NATS publish plumbing + a versioned prompt registry

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/

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.
+"""

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"

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()

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:<name>"`` for a consumer, ``"producer:<name>"`` 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,
+    }