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

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 events import bus, subjects
+from 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)

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

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 +

config/settings.py

@@ -308,6 +308,8 @@ 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) ──
+    "events",
     # ── Generic LLM invocation (provider registry: bedrock/proxy/CLI harnesses) ─
     "llm",
     # ── Unified search (platform-wide; queries all three domains' indices) ────
@@ -1110,3 +1112,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", "")

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

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 = "events"