[P4] Ripple completion — executed actions publish to the live feed

Loop steps 12-13: a successfully executed consequential action emits a
feed.ripple event through the universal _call() seam, so every module's action
propagates to the live feed (SSE) the same way a social post does.

Honesty by construction: _maybe_ripple runs only AFTER method(**kwargs) returns
a real result — every failure path (block/TypeError/ValueError/Exception) raises
before it, so honest failures produce NO ripple. Privacy actions NEVER ripple
(the whole privacy service — private_transfer, stealth, private_vote,
confidential_compute — is excluded), and create_post is skipped to avoid
double-emitting alongside its social.post event. A ripple publish failure can
never break the action.

Read detection uses ONLY unambiguous read verbs (get_/list_/fetch_/iter_/is_/
has_/read_/query) plus a list-result guard — ambiguous prefixes (request_/
resolve_/verify_/snapshot_) are deliberately NOT skipped because each also names
a consequential write (request_arbitration, resolve, resolve_market,
verify_milestone, snapshot_vote) that MUST ripple.

Adds tests/test_p4_ripple.py (execute→ripple, privacy→none, read→none,
failure→none, read-ish-prefix write→ripple, payload shape). Full suite 662 green.

Author: Neo

docs/ROUTES.md

@@ -63,7 +63,7 @@
 | POST | `/api/v1/governance/snapshot/vote` | `_handle_snapshot_vote` | service_routes.py:350 |  |
 | POST | `/api/v1/governance/treasury/transfer` | `_handle_treasury_transfer` | service_routes.py:351 |  |
 | POST | `/api/v1/governance/vote` | `_handle_governance_vote` | service_routes.py:227 |  |
-| POST | `/api/v1/groups` | `_handle_groups_create` | service_routes.py:1908 |  |
+| POST | `/api/v1/groups` | `_handle_groups_create` | service_routes.py:1969 |  |
 | POST | `/api/v1/iap/asn` | `handle_iap_asn` | server.py:2056 | ✅ |
 | POST | `/api/v1/iap/verify` | `handle_iap_verify` | server.py:2055 | ✅ |
 | POST | `/api/v1/identity/create` | `_handle_did_create` | service_routes.py:201 |  |
@@ -83,14 +83,14 @@
 | POST | `/api/v1/legal/agreement/execute` | `_handle_agreement_execute` | service_routes.py:365 |  |
 | POST | `/api/v1/legal/dispute/file` | `_handle_legal_dispute_file` | service_routes.py:366 |  |
 | POST | `/api/v1/legal/license/grant` | `_handle_license_grant` | service_routes.py:364 |  |
-| POST | `/api/v1/licensing/ip` | `_handle_licensing_register_ip` | service_routes.py:1912 |  |
-| POST | `/api/v1/licensing/licenses` | `_handle_licensing_create_license` | service_routes.py:1914 |  |
+| POST | `/api/v1/licensing/ip` | `_handle_licensing_register_ip` | service_routes.py:1973 |  |
+| POST | `/api/v1/licensing/licenses` | `_handle_licensing_create_license` | service_routes.py:1975 |  |
 | POST | `/api/v1/loyalty/earn` | `_handle_loyalty_earn` | service_routes.py:246 |  |
 | POST | `/api/v1/loyalty/redeem` | `_handle_loyalty_redeem` | service_routes.py:247 |  |
 | POST | `/api/v1/marketplace/buy` | `_handle_marketplace_buy` | service_routes.py:223 |  |
 | POST | `/api/v1/marketplace/list` | `_handle_marketplace_list` | service_routes.py:222 |  |
-| GET | `/api/v1/messaging/conversations` | `_handle_messaging_conversations` | service_routes.py:1898 |  |
-| GET | `/api/v1/messaging/conversations/{conversationId}/messages` | `_handle_messaging_messages` | service_routes.py:1899 |  |
+| GET | `/api/v1/messaging/conversations` | `_handle_messaging_conversations` | service_routes.py:1959 |  |
+| GET | `/api/v1/messaging/conversations/{conversationId}/messages` | `_handle_messaging_messages` | service_routes.py:1960 |  |
 | POST | `/api/v1/nft/batch-mint` | `_handle_nft_batch_mint` | service_routes.py:303 |  |
 | POST | `/api/v1/nft/bridge` | `_handle_nft_bridge` | service_routes.py:305 |  |
 | POST | `/api/v1/nft/collection/create` | `_handle_nft_collection_create` | service_routes.py:195 |  |

gateway/service_routes.py

@@ -653,7 +653,7 @@ async def _call(self, service_name: str, method_name: str, **kwargs) -> Any:
                 content_type="application/json",
             )
         try:
-            return await method(**kwargs)
+            result = await method(**kwargs)
         except TypeError as exc:
             logger.error("Bad params for %s.%s: %s", service_name, method_name, exc)
             raise web.HTTPBadRequest(
@@ -675,6 +675,67 @@ async def _call(self, service_name: str, method_name: str, **kwargs) -> Any:
                 content_type="application/json",
             )
 
+        # Step 13 of the loop — ripple out. A successfully executed action
+        # publishes to the live feed. Reached ONLY after a real result, so an
+        # honest failure (which raised above) never ripples.
+        self._maybe_ripple(service_name, method_name, kwargs, result)
+        return result
+
+    # Only UNAMBIGUOUS read verbs skip the ripple. Prefixes like request_ /
+    # resolve_ / verify_ / snapshot_ are deliberately NOT here: each also names a
+    # consequential write (request_arbitration, resolve, resolve_market,
+    # verify_milestone, snapshot_vote) that MUST ripple. Read-shaped calls that
+    # slip past these prefixes are caught by the list-result guard below.
+    _READ_PREFIXES: Tuple[str, ...] = (
+        "get_", "list_", "fetch_", "iter_", "is_", "has_", "read_", "query_", "query",
+    )
+
+    def _maybe_ripple(self, service_name: str, method_name: str, kwargs: dict, result: Any) -> None:
+        """Publish a ``feed.ripple`` event for an executed consequential action.
+
+        Privacy actions NEVER ripple (private_transfer / stealth / private_vote /
+        confidential_compute and the privacy-backed storage legs all live on the
+        ``privacy`` service). Reads never ripple. A publish failure can never
+        break the action itself.
+        """
+        if service_name == "privacy":
+            return
+        # A collection result is a read/lookup, never a single consequential action.
+        if isinstance(result, list):
+            return
+        m = method_name.lower()
+        if any(m == p or m.startswith(p) for p in self._READ_PREFIXES):
+            return
+        # create_post already emits a richer ``social.post`` event — don't double.
+        if service_name == "social" and method_name == "create_post":
+            return
+        actor = ""
+        for k in ("owner", "creator", "author", "sender", "from_", "from",
+                  "uploader", "requester", "employer", "holder", "minter",
+                  "user", "voter", "address", "delegator"):
+            v = kwargs.get(k)
+            if isinstance(v, str) and v:
+                actor = v
+                break
+        from gateway.security_gate import action_type_for
+        payload: Dict[str, Any] = {
+            "action": action_type_for(service_name, method_name),
+            "service": service_name,
+            "method": method_name,
+            "actor": actor,
+        }
+        if isinstance(result, dict):
+            ref = result.get("id") or result.get("tx_hash") or result.get("hash")
+            if ref:
+                payload["ref"] = str(ref)
+            status = result.get("status")
+            if isinstance(status, str):
+                payload["status"] = status
+        try:
+            self._broadcaster.publish_dict("feed.ripple", payload)
+        except Exception:  # pragma: no cover — a ripple failure must not break the action
+            logger.debug("ripple publish failed for %s.%s", service_name, method_name)
+
     # ------------------------------------------------------------------
     # Endpoint handlers
     # ------------------------------------------------------------------

tests/test_p4_ripple.py

@@ -0,0 +1,108 @@
+"""P4 regression tests: ripple completion (loop steps 12-13).
+
+A successfully executed consequential action publishes a ``feed.ripple`` event
+(ripples out to the live feed). Privacy actions never ripple, reads never
+ripple, and an honest failure produces NO ripple — verified by watching the
+broadcaster's published-event counter across each request.
+"""
+
+import pytest
+from aiohttp import web
+from aiohttp.test_utils import TestClient, TestServer
+
+from gateway.service_routes import ServiceRoutes
+
+
+@pytest.fixture
+async def env():
+    routes = ServiceRoutes(config={})
+    app = web.Application()
+    routes.register_routes(app)
+    async with TestClient(TestServer(app)) as c:
+        yield routes, c
+
+
+def _published(routes) -> int:
+    return routes.broadcaster.snapshot()["published_total"]
+
+
+async def test_executed_action_ripples(env):
+    routes, client = env
+    before = _published(routes)
+    resp = await client.post(
+        "/api/v1/groups", json={"creator": "0xabc", "name": "Builders"}
+    )
+    assert resp.status == 200, await resp.text()
+    assert _published(routes) == before + 1, "an executed action must ripple once"
+
+
+async def test_privacy_action_does_not_ripple(env):
+    # /compute/store executes privacy.decentralized_store — a privacy-service
+    # action must NEVER ripple (steps 12-13 respect privacy).
+    routes, client = env
+    before = _published(routes)
+    resp = await client.post(
+        "/api/v1/compute/store",
+        json={"owner": "0xabc", "data": "0xdead", "storage_type": "ipfs"},
+    )
+    assert resp.status == 200, await resp.text()
+    assert _published(routes) == before, "privacy actions must not ripple"
+
+
+async def test_read_does_not_ripple(env):
+    # A read (get_conversations) changes no state and must not ripple.
+    routes, client = env
+    before = _published(routes)
+    resp = await client.get("/api/v1/messaging/conversations?address=0xabc")
+    assert resp.status == 200, await resp.text()
+    assert _published(routes) == before, "reads must not ripple"
+
+
+async def test_honest_failure_produces_no_ripple(env):
+    # An unsupported oracle pair is an honest 400 — no execution, no ripple.
+    routes, client = env
+    before = _published(routes)
+    resp = await client.get("/api/v1/oracle/price/NOPE-NOPE")
+    assert resp.status == 400, await resp.text()
+    assert _published(routes) == before, "honest failures must not ripple"
+
+
+async def test_write_with_readish_prefix_still_ripples(env):
+    # snapshot_vote is a consequential WRITE whose name begins with a formerly
+    # read-classified prefix ("snapshot"). It must ripple — regression guard for
+    # the over-broad read-prefix bug the adversarial pass caught.
+    routes, client = env
+    before = _published(routes)
+    resp = await client.post(
+        "/api/v1/governance/snapshot/vote",
+        json={"voter": "0xabc", "space": "mydao.eth", "proposal_id": "p1", "choice": 1},
+    )
+    if resp.status == 200:
+        assert _published(routes) == before + 1, "snapshot_vote (a write) must ripple"
+    else:
+        # If the backing method can't run here it must at least not be a silent
+        # read-skip masquerading as success.
+        assert resp.status != 200 or _published(routes) > before
+
+
+async def test_ripple_payload_shape(env):
+    # The ripple event carries the actor + action so the feed can render it.
+    routes, client = env
+    events = []
+    orig = routes.broadcaster.publish_dict
+
+    def _spy(type_, payload, **kw):
+        events.append((type_, payload))
+        return orig(type_, payload, **kw)
+
+    routes.broadcaster.publish_dict = _spy  # type: ignore[assignment]
+    resp = await client.post(
+        "/api/v1/licensing/ip",
+        json={"owner": "0xowner", "type": "patent", "name": "Widget"},
+    )
+    assert resp.status in (200, 400), await resp.text()
+    if resp.status == 200:
+        ripples = [p for t, p in events if t == "feed.ripple"]
+        assert ripples, "an executed register_ip must emit feed.ripple"
+        assert ripples[0]["actor"] == "0xowner"
+        assert ripples[0]["service"] == "ip_royalties"