feat(correlation/attribution): wire bus handler, persist state (Phase 4)

attribution_worker.handle_observation_event now executes the full
end-to-end path:

* ensure stub identity (Phase 1)
* observations_for_identity_primitive() — new repo helper joining
  observations through attackers.identity_id, so v1's clusterer
  gets cross-attacker rollup for free
* aggregate_observations() with ValueKind dispatched off the BEHAVE
  PRIMITIVE_REGISTRY; unknown primitives default to categorical
* upsert_attribution_state() — last_change_ts locked when state is
  unchanged so the dashboard can render "stable since X"
* publish attribution.profile.state_changed only on transition;
  idempotent re-runs over the same observation set fire nothing
  (loop-prevention invariant matching ttp.tagged)

Tests:
* 5 end-to-end attribution scenarios over in-memory SQLite + FakeBus.
* test_base_repo's DummyRepo + coverage body now stub every abstract
  surface BaseRepository declares — the 6 added by this branch plus
  the 12 left un-stubbed by earlier work (BEHAVE Phase 1, TTP
  rollups, iter helpers). The coverage test could not previously
  even instantiate.
* test_aggregate_categorical's dispatcher rejection updated for the
  Phase 3 + 4 contract — ValueError on unknown kinds, not
  NotImplementedError.

decnet/correlation/attribution_worker.py

@@ -26,12 +26,25 @@ from decnet.bus import topics as _topics
 from decnet.bus.base import BaseBus
 from decnet.bus.factory import get_bus
 from decnet.bus.publish import (
+    publish_safely,
     run_control_listener_signal as _run_control_listener_signal,
     run_health_heartbeat as _run_health_heartbeat,
 )
+from decnet.correlation.attribution.aggregate import aggregate_observations
 from decnet.logging import get_logger
 from decnet.web.db.repository import BaseRepository
 
+try:
+    from decnet_behave_shell.spec import (
+        PRIMITIVE_REGISTRY,
+        ValueKind,
+    )
+    _BEHAVE_REGISTRY_AVAILABLE = True
+except ImportError:  # pragma: no cover
+    PRIMITIVE_REGISTRY = {}
+    ValueKind = None
+    _BEHAVE_REGISTRY_AVAILABLE = False
+
 log = get_logger("correlation.attribution_worker")
 
 _WORKER_NAME = "attribution"
@@ -156,13 +169,103 @@ async def handle_observation_event(
             attacker_uuid,
         )
         return
-    # Phase 4 will run the merger here and emit
-    # ``attribution.profile.state_changed`` on transition. Phase 1
-    # ends with stub materialisation only.
-    log.debug(
-        "attribution worker: stub identity=%s for attacker=%s primitive=%s",
-        identity_uuid, attacker_uuid, primitive,
+    primitive_str = str(primitive)
+
+    # Load the full per-(identity, primitive) observation series.
+    # v0 with 1:1 stub identities, this is the single attacker's
+    # series; v1's clusterer makes it a cross-attacker union.
+    observations = await repo.observations_for_identity_primitive(
+        identity_uuid, primitive_str,
     )
+    if not observations:
+        log.debug(
+            "attribution worker: no observations yet for identity=%s "
+            "primitive=%s (race with upsert)",
+            identity_uuid, primitive_str,
+        )
+        return
+
+    # Run merger.
+    value_kind = _value_kind_for(primitive_str)
+    new_state = aggregate_observations(observations, value_kind=value_kind)
+
+    # Load prior state to detect transitions.
+    prior = await repo.get_attribution_state(identity_uuid, primitive_str)
+    state_changed = prior is None or prior.get("state") != new_state.state
+
+    # Persist. last_change_ts is locked to the prior row when state is
+    # unchanged so the dashboard's "stable since" timestamp doesn't
+    # reset on every observation.
+    if prior is not None and not state_changed:
+        last_change_ts = float(prior.get("last_change_ts", new_state.last_observation_ts))
+    else:
+        last_change_ts = new_state.last_observation_ts
+    await repo.upsert_attribution_state({
+        "identity_uuid": identity_uuid,
+        "primitive": primitive_str,
+        "current_value": new_state.current_value,
+        "state": new_state.state,
+        "confidence": new_state.confidence,
+        "observation_count": new_state.observation_count,
+        "last_change_ts": last_change_ts,
+        "last_observation_ts": new_state.last_observation_ts,
+    })
+
+    # Emit state_changed only on transition. Idempotent re-runs (same
+    # observations, same merger output) produce no event — matches
+    # the loop-prevention invariant that ttp.tagged uses.
+    if state_changed and bus is not None:
+        await publish_safely(
+            bus,
+            _topics.attribution(_topics.ATTRIBUTION_PROFILE_STATE_CHANGED),
+            {
+                "identity_uuid": identity_uuid,
+                "primitive": primitive_str,
+                "old_state": prior.get("state") if prior else None,
+                "new_state": new_state.state,
+                "current_value": new_state.current_value,
+                "confidence": new_state.confidence,
+                "observation_count": new_state.observation_count,
+                "ts": new_state.last_observation_ts,
+            },
+            event_type=_topics.ATTRIBUTION_PROFILE_STATE_CHANGED,
+        )
+        log.info(
+            "attribution worker: identity=%s primitive=%s %s -> %s confidence=%.2f",
+            identity_uuid, primitive_str,
+            (prior or {}).get("state") or "<new>", new_state.state,
+            new_state.confidence,
+        )
+
+
+def _value_kind_for(primitive: str) -> str:
+    """Resolve a BEHAVE primitive name to the merger's ValueKind tag.
+
+    Maps the BEHAVE registry's ``ValueKind`` enum onto the three
+    mergers the engine ships:
+
+    * ``CATEGORICAL`` / ``BOOL`` / ``FREE_STRING`` / ``ARRAY`` →
+      ``"categorical"`` (BOOL is a 2-cardinality categorical;
+      FREE_STRING and ARRAY collapse to opaque-token categorical
+      until a v1 specialised merger lands)
+    * ``NUMERIC`` → ``"numeric"``
+    * ``HASH``    → ``"hash"``
+
+    Unknown primitives (registry miss) default to categorical — the
+    safest fallback because the categorical merger is one-outlier-
+    tolerant and won't lie about confidence on noisy categorical
+    data the way a numeric merger would on non-numeric values.
+    """
+    if not _BEHAVE_REGISTRY_AVAILABLE:
+        return "categorical"
+    spec = PRIMITIVE_REGISTRY.get(primitive)
+    if spec is None or ValueKind is None:
+        return "categorical"
+    if spec.kind is ValueKind.NUMERIC:
+        return "numeric"
+    if spec.kind is ValueKind.HASH:
+        return "hash"
+    return "categorical"
 
 
 def _payload_of(event: Any) -> dict[str, Any]:

decnet/web/db/repository.py

@@ -341,6 +341,20 @@ class BaseRepository(ABC):
         ordered by ``ts`` ASC. Empty list when none."""
         raise NotImplementedError
 
+    @abstractmethod
+    async def observations_for_identity_primitive(
+        self, identity_uuid: str, primitive: str,
+    ) -> list[dict[str, Any]]:
+        """Every observation of ``primitive`` across all attackers
+        rolling up to ``identity_uuid``, ordered by ``ts`` ASC.
+
+        Empty list when the identity has no observations of this
+        primitive. v0 with 1:1 stub identities returns the same set
+        as ``observations_time_series(attacker_uuid, primitive)``;
+        v1's clusterer makes the union meaningful.
+        """
+        raise NotImplementedError
+
     @abstractmethod
     async def has_observations_for_evidence(self, evidence_ref: str) -> bool:
         """True iff any observation row carries this ``evidence_ref``.

decnet/web/db/sqlmodel_repo/observations.py

@@ -25,7 +25,7 @@ from typing import Any, Optional
 from sqlalchemy import desc, func, select
 from sqlmodel import col
 
-from decnet.web.db.models import ObservationRow
+from decnet.web.db.models import Attacker, ObservationRow
 from decnet.web.db.sqlmodel_repo._helpers import _MixinBase
 
 
@@ -164,6 +164,34 @@ class ObservationsMixin(_MixinBase):
                 return None
             return row.model_dump(mode="json")
 
+    async def observations_for_identity_primitive(
+        self, identity_uuid: str, primitive: str,
+    ) -> list[dict[str, Any]]:
+        """Union of every observation of *primitive* across the
+        attackers rolling up to *identity_uuid*, ordered ``ts`` ASC.
+
+        v0 with 1:1 stub identities returns the same set as
+        ``observations_time_series(attacker_uuid, primitive)``.
+        v1's clusterer makes the union load-bearing — multiple
+        attackers point at the same identity_id and this query is
+        what gives the merger a cross-attacker view.
+        """
+        async with self._session() as session:
+            stmt = (
+                select(ObservationRow)
+                .join(Attacker, ObservationRow.attacker_uuid == Attacker.uuid)
+                .where(
+                    Attacker.identity_id == identity_uuid,
+                    ObservationRow.primitive == primitive,
+                )
+                .order_by(ObservationRow.ts)
+            )
+            rows = (await session.execute(stmt)).scalars().all()
+            return [
+                {"ts": row.ts, "value": row.value, "confidence": row.confidence}
+                for row in rows
+            ]
+
     async def has_observations_for_evidence(
         self, evidence_ref: str,
     ) -> bool: