feat(state-hub): CUST-WP-0040 — NATS lifecycle event publishing for activity-core

Makes the state hub an event publisher so activity-core can drive
maintenance automation declaratively via ActivityDefinitions, rather
than the hub creating tasks itself.

- api/events/: lazy JetStream publisher + EventEnvelope mirroring
  activity-core's contract; no-op when NATS_URL unset, fire-and-forget
  with logged failures so publishing never breaks an API request.
- Wired publishers on the five v1.0 lifecycle events:
    org.statehub.repo.registered        (POST /repos/)
    org.statehub.workstream.completed   (PATCH /workstreams/* on transition)
    org.statehub.decision.resolved      (POST /decisions/*/resolve)
    org.statehub.domain.goal.activated  (POST /domain-goals/*/activate)
    org.statehub.task.stale             (scripts/cleanup_stale_tasks.py)
- docs/nats-event-subjects.md: subject naming convention + catalog.
- docs/cron-migration.md: design stub for replacing custodian-sync
  systemd timer and cleanup-stale cron with ActivityDefinitions
  (depends on activity-core WP-0003).
- docs/activity-core-delegation.md: protocol, invariants, cutover plan.
- SCOPE.md: declares activity-core as downstream event consumer and
  restates that the state hub stays a read model, not a task factory.

Workplan: workplans/CUST-WP-0040-state-hub-nats-activity-core-integration.md
242 tests pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

state-hub/api/events/__init__.py

@@ -0,0 +1,13 @@
+from api.events.envelope import EventEnvelope
+from api.events.nats_publisher import (
+    publish_event,
+    publish_event_sync,
+    shutdown_publisher,
+)
+
+__all__ = [
+    "EventEnvelope",
+    "publish_event",
+    "publish_event_sync",
+    "shutdown_publisher",
+]

state-hub/api/events/envelope.py

@@ -0,0 +1,55 @@
+"""EventEnvelope — schema for state-hub lifecycle events published to NATS.
+
+Mirrors the EventEnvelope contract defined in activity-core
+(`src/activity_core/models.py`). The state-hub publishes; activity-core
+consumes and routes to ActivityDefinitions.
+
+Subject naming convention (see docs/nats-event-subjects.md):
+
+    org.statehub.{noun}.{verb}
+
+Examples:
+    org.statehub.repo.registered
+    org.statehub.workstream.completed
+    org.statehub.decision.resolved
+    org.statehub.domain.goal.activated
+    org.statehub.task.stale
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Any
+
+from pydantic import BaseModel, Field
+
+PUBLISHER = "the-custodian/state-hub"
+
+
+class EventEnvelope(BaseModel):
+    """Standard envelope shared with activity-core. Do not break compatibility.
+
+    All inbound events on activity-core's side are normalised into this shape.
+    """
+
+    id: str = Field(description="UUID v4 — stable unique ID for deduplication.")
+    type: str = Field(description="Dot-namespaced event type, e.g. 'org.statehub.repo.registered'.")
+    version: str = Field(default="1.0", description="Schema version string.")
+    timestamp: datetime = Field(description="When the event occurred (UTC).")
+    publisher: str = Field(default=PUBLISHER, description="Originating service.")
+    attributes: dict[str, Any] = Field(
+        default_factory=dict,
+        description="Event-specific attributes; structure varies by event type.",
+    )
+
+    @classmethod
+    def new(cls, event_type: str, attributes: dict[str, Any] | None = None) -> "EventEnvelope":
+        """Construct an envelope with a fresh UUID and current UTC timestamp."""
+        return cls(
+            id=str(uuid.uuid4()),
+            type=event_type,
+            timestamp=datetime.now(tz=timezone.utc),
+            publisher=PUBLISHER,
+            attributes=attributes or {},
+        )

state-hub/api/events/nats_publisher.py

@@ -0,0 +1,139 @@
+"""NATS JetStream publisher for state-hub lifecycle events.
+
+Design:
+    - One process-wide publisher (`_Publisher` singleton).
+    - Connects lazily on first publish; reuses the connection thereafter.
+    - When ``NATS_URL`` is unset or empty, every publish is a logged no-op
+      so the state hub remains usable in environments without NATS.
+    - All publishes are fire-and-forget from the caller's perspective.
+      Failures are logged but never raise — losing a lifecycle event must
+      never break the API request that triggered it.
+
+Stream + subject conventions live in ``docs/nats-event-subjects.md``.
+Envelope schema lives in :mod:`api.events.envelope`.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import logging
+import os
+from typing import TYPE_CHECKING
+
+from api.events.envelope import EventEnvelope
+
+if TYPE_CHECKING:  # pragma: no cover — import-only for typing
+    from nats.aio.client import Client as NATSClient
+    from nats.js.client import JetStreamContext
+
+logger = logging.getLogger("state_hub.events.nats")
+
+_STREAM_NAME = "ACTIVITY_EVENTS"
+_STREAM_SUBJECT_PATTERN = "org.>"
+
+
+def _nats_url() -> str | None:
+    """Resolve NATS_URL at call time so tests / configs can override it."""
+    url = os.environ.get("NATS_URL", "").strip()
+    return url or None
+
+
+class _Publisher:
+    """Singleton holding the live NATS connection + JetStream context."""
+
+    def __init__(self) -> None:
+        self._nc: "NATSClient | None" = None
+        self._js: "JetStreamContext | None" = None
+        self._connect_lock = asyncio.Lock()
+        self._ensured_stream = False
+
+    async def _connect(self, url: str) -> None:
+        # Imported inside the method so module import works without the dep.
+        import nats
+        import nats.js.api
+
+        async with self._connect_lock:
+            if self._nc is not None and self._nc.is_connected:
+                return
+            self._nc = await nats.connect(url, connect_timeout=2)
+            self._js = self._nc.jetstream()
+            logger.info("nats: connected to %s", url)
+
+            if not self._ensured_stream:
+                try:
+                    await self._js.find_stream_name_by_subject("org.statehub.repo.registered")
+                    self._ensured_stream = True
+                except Exception:
+                    try:
+                        await self._js.add_stream(
+                            nats.js.api.StreamConfig(
+                                name=_STREAM_NAME,
+                                subjects=[_STREAM_SUBJECT_PATTERN],
+                            )
+                        )
+                        logger.info("nats: created JetStream stream %r", _STREAM_NAME)
+                    except Exception as exc:  # pragma: no cover — defensive
+                        logger.warning("nats: could not ensure stream %r: %s", _STREAM_NAME, exc)
+                    self._ensured_stream = True
+
+    async def publish(self, subject: str, envelope: EventEnvelope) -> None:
+        url = _nats_url()
+        if url is None:
+            logger.debug("nats: NATS_URL unset — skipping publish %s (id=%s)", subject, envelope.id)
+            return
+
+        try:
+            if self._nc is None or not self._nc.is_connected:
+                await self._connect(url)
+            assert self._js is not None
+            payload = envelope.model_dump_json().encode()
+            ack = await self._js.publish(subject, payload)
+            logger.info(
+                "nats: published %s id=%s stream=%s seq=%s",
+                subject,
+                envelope.id,
+                getattr(ack, "stream", "?"),
+                getattr(ack, "seq", "?"),
+            )
+        except Exception as exc:
+            logger.warning("nats: publish failed %s id=%s err=%s", subject, envelope.id, exc)
+
+    async def shutdown(self) -> None:
+        if self._nc is not None:
+            try:
+                await self._nc.drain()
+            except Exception:  # pragma: no cover — defensive
+                pass
+            self._nc = None
+            self._js = None
+            self._ensured_stream = False
+
+
+_PUBLISHER = _Publisher()
+
+
+async def publish_event(subject: str, envelope: EventEnvelope) -> None:
+    """Publish ``envelope`` on ``subject``. Logs but never raises on failure.
+
+    No-op when ``NATS_URL`` is not configured.
+    """
+    await _PUBLISHER.publish(subject, envelope)
+
+
+def publish_event_sync(subject: str, envelope: EventEnvelope) -> None:
+    """Fire-and-forget variant for sync callers (scripts, cron jobs).
+
+    Runs the publish in a short-lived event loop. Intended for one-shot CLI
+    callers that aren't already inside an async context. Server code should
+    prefer :func:`publish_event` with ``asyncio.create_task``.
+    """
+    try:
+        asyncio.run(publish_event(subject, envelope))
+    except RuntimeError:
+        # Already inside a running loop — schedule and forget.
+        asyncio.get_event_loop().create_task(publish_event(subject, envelope))
+
+
+async def shutdown_publisher() -> None:
+    """Drain the NATS connection on app shutdown."""
+    await _PUBLISHER.shutdown()

state-hub/api/main.py

@@ -9,6 +9,7 @@ from starlette.requests import Request
 from starlette.responses import Response as StarletteResponse
 
 from api.database import engine
+from api.events import shutdown_publisher
 from api.routers import decisions, extension_points, progress, state, tasks, technical_debt, topics, workstreams, workstream_dependencies
 from api.routers import domains, repos, contributions, sbom, policy, domain_goals, repo_goals, messages, capability_requests, tpsc
 from api.routers import token_events
@@ -53,6 +54,7 @@ class ETagMiddleware(BaseHTTPMiddleware):
 @asynccontextmanager
 async def lifespan(app: FastAPI):
     yield
+    await shutdown_publisher()
     await engine.dispose()
 
 

state-hub/api/routers/decisions.py

@@ -1,3 +1,4 @@
+import asyncio
 import logging
 import uuid
 from datetime import datetime, timezone
@@ -10,6 +11,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.database import get_session
+from api.events import EventEnvelope, publish_event
 from api.models.decision import Decision, DecisionStatus, DecisionType
 from api.models.progress_event import ProgressEvent
 from api.schemas.decision import DecisionCreate, DecisionRead, DecisionResolve, DecisionUpdate
@@ -150,6 +152,20 @@ async def resolve_decision_action(
     if body.write_log:
         await _write_project_log(decision, body.rationale, body.decided_by, session)
 
+    subject = "org.statehub.decision.resolved"
+    envelope = EventEnvelope.new(
+        subject,
+        attributes={
+            "decision_id": str(decision.id),
+            "title": decision.title,
+            "topic_id": str(decision.topic_id) if decision.topic_id else None,
+            "workstream_id": str(decision.workstream_id) if decision.workstream_id else None,
+            "decided_by": body.decided_by,
+            "rationale_snippet": (body.rationale or "")[:240],
+        },
+    )
+    asyncio.create_task(publish_event(subject, envelope))
+
     return decision
 
 

state-hub/api/routers/domain_goals.py

@@ -1,3 +1,4 @@
+import asyncio
 import uuid
 
 from fastapi import APIRouter, Depends, HTTPException, status
@@ -5,6 +6,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.database import get_session
+from api.events import EventEnvelope, publish_event
 from api.models.domain import Domain
 from api.models.domain_goal import DomainGoal, DomainGoalStatus  # noqa: F401 (DomainGoalStatus used in activate)
 from api.schemas.domain_goal import DomainGoalCreate, DomainGoalRead, DomainGoalUpdate
@@ -96,6 +98,7 @@ async def activate_domain_goal(
     goal = await session.get(DomainGoal, goal_id)
     if goal is None:
         raise HTTPException(status_code=404, detail="Domain goal not found")
+    was_active = goal.status == DomainGoalStatus.active
 
     # Supersede any other active goal for this domain
     existing = await session.execute(
@@ -105,10 +108,28 @@ async def activate_domain_goal(
             DomainGoal.id != goal_id,
         )
     )
+    superseded_ids: list[str] = []
     for old in existing.scalars().all():
         old.status = DomainGoalStatus.superseded
+        superseded_ids.append(str(old.id))
 
     goal.status = DomainGoalStatus.active
     await session.commit()
     await session.refresh(goal)
+
+    if not was_active:
+        domain = await session.get(Domain, goal.domain_id)
+        subject = "org.statehub.domain.goal.activated"
+        envelope = EventEnvelope.new(
+            subject,
+            attributes={
+                "goal_id": str(goal.id),
+                "domain_id": str(goal.domain_id),
+                "domain_slug": domain.slug if domain else None,
+                "title": goal.title,
+                "superseded_goal_ids": superseded_ids,
+            },
+        )
+        asyncio.create_task(publish_event(subject, envelope))
+
     return goal

state-hub/api/routers/repos.py

@@ -16,6 +16,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.config import settings
 from api.database import get_session
+from api.events import EventEnvelope, publish_event
 from api.doi_engine import (
     compute_fingerprint,
     evaluate as _doi_evaluate,
@@ -94,6 +95,20 @@ async def register_repo(
     session.add(repo)
     await session.commit()
     await session.refresh(repo)
+
+    subject = "org.statehub.repo.registered"
+    envelope = EventEnvelope.new(
+        subject,
+        attributes={
+            "repo_id": str(repo.id),
+            "repo_slug": repo.slug,
+            "domain_slug": body.domain_slug,
+            "remote_url": repo.remote_url,
+            "local_path": repo.local_path,
+        },
+    )
+    asyncio.create_task(publish_event(subject, envelope))
+
     return repo
 
 

state-hub/api/routers/workstreams.py

@@ -1,3 +1,4 @@
+import asyncio
 import uuid
 import socket
 import time
@@ -9,6 +10,7 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from api.database import get_session
+from api.events import EventEnvelope, publish_event
 from api.models.managed_repo import ManagedRepo
 from api.models.workstream import Workstream
 from api.schemas.workstream import (
@@ -168,10 +170,27 @@ async def update_workstream(
     ws = await session.get(Workstream, workstream_id)
     if ws is None:
         raise HTTPException(status_code=404, detail="Workstream not found")
+    prev_status = ws.status
     for field, value in body.model_dump(exclude_unset=True).items():
         setattr(ws, field, value)
     await session.commit()
     await session.refresh(ws)
+
+    if prev_status != "completed" and ws.status == "completed":
+        subject = "org.statehub.workstream.completed"
+        envelope = EventEnvelope.new(
+            subject,
+            attributes={
+                "workstream_id": str(ws.id),
+                "slug": ws.slug,
+                "title": ws.title,
+                "topic_id": str(ws.topic_id),
+                "repo_id": str(ws.repo_id) if ws.repo_id else None,
+                "repo_goal_id": str(ws.repo_goal_id) if ws.repo_goal_id else None,
+            },
+        )
+        asyncio.create_task(publish_event(subject, envelope))
+
     return ws