feat: SSE AgentSignals for agent events

.github/workflows/test.yaml

@@ -14,7 +14,7 @@ env:
   # deliberate and reviewable instead of being buried inside individual steps.
   PYTHON_VERSION: '3.12.6'
   KERIPY_REF: '4ee02c0213770d25a0114fe7ebd7ab4ab5500cde'
-  KERIA_REF: '9e2461550f373ad7bdbe7eebeaceac689cb15397'
+  KERIA_REF: '5b703bd8a60fab68a6476819626b22784317bf14'
   VLEI_REF: 'f514b9431c5f965b5f7f64a8693e19df2f181564'
 
 jobs:
@@ -109,7 +109,7 @@ jobs:
       - name: Stage 1 - Check out KERIA
         uses: actions/checkout@v4.2.2
         with:
-          repository: WebOfTrust/keria
+          repository: kentbull/keria
           ref: ${{ env.KERIA_REF }}
           path: keria
 

scripts/sync_integration_deps.py

@@ -25,8 +25,10 @@ def sync_repo(root: Path, repo: str, ref: str) -> None:
     if not root.exists():
         root.parent.mkdir(parents=True, exist_ok=True)
         run("git", "clone", repo, str(root))
+    else:
+        run("git", "-C", str(root), "remote", "set-url", "origin", repo)
 
-    run("git", "-C", str(root), "fetch", "--tags", "origin")
+    run("git", "-C", str(root), "fetch", "--tags", "--force", "origin")
     run("git", "-C", str(root), "fetch", "origin")
     run("git", "-C", str(root), "checkout", "--detach", ref)
 

src/signify/app/clienting.py

@@ -419,6 +419,11 @@ def schemas(self):
         from signify.app.schemas import Schemas
         return Schemas(client=self)
 
+    def signals(self):
+        """Return the generic KERIA agent signaling resource wrapper."""
+        from signify.app.signaling import AgentSignals
+        return AgentSignals(client=self)
+
     def config(self):
         """Return the agent-configuration read resource wrapper."""
         from signify.app.coring import Config

src/signify/app/signaling.py

@@ -0,0 +1,39 @@
+# -*- encoding: utf-8 -*-
+"""Generic KERIA agent signaling helpers."""
+
+from keri import kering
+from keri.core import indexing, serdering
+
+
+class AgentSignals:
+    """Generic signed event stream for one connected KERIA agent."""
+
+    def __init__(self, client):
+        self.client = client
+
+    def stream(self):
+        """Open the authenticated generic agent SSE stream."""
+        return self.client.stream(
+            "/signals/stream",
+            headers={"Accept": "text/event-stream"},
+        )
+
+    def verifyReplyEnvelope(self, envelope, route=None):
+        """Verify one KERIA agent-signed KERI ``rpy`` envelope."""
+        if self.client.agent is None:
+            raise kering.ConfigurationError("client must be connected before verification")
+
+        rserder = serdering.SerderKERI(sad=envelope["rpy"])
+        if route is not None and rserder.ked.get("r") != route:
+            return False
+
+        data = rserder.ked.get("a", {})
+        if data.get("agent") != self.client.agent.pre:
+            return False
+
+        sigs = envelope.get("sigs") or []
+        if not sigs:
+            return False
+
+        siger = indexing.Siger(qb64=sigs[0])
+        return self.client.agent.verfer.verify(sig=siger.raw, ser=rserder.raw)

tests/app/test_clienting.py

@@ -837,6 +837,16 @@ def test_signify_client_schemas(make_signify_client):
     assert out.client == client
 
 
+def test_signify_client_signals(make_signify_client):
+    client = make_signify_client()
+
+    out = client.signals()
+
+    from signify.app.signaling import AgentSignals
+    assert type(out) is AgentSignals
+    assert out.client == client
+
+
 def test_signify_client_config(make_signify_client):
     client = make_signify_client()
 

tests/app/test_signaling.py

@@ -0,0 +1,82 @@
+# -*- encoding: utf-8 -*-
+"""
+SIGNIFY generic agent signaling helper tests.
+"""
+
+import pytest
+from keri import kering
+from keri.core import eventing, signing
+from mockito import expect, mock, verifyNoUnwantedInteractions, unstub
+
+from signify.app.signaling import AgentSignals
+
+
+def signed_envelope(route="/test/signals/request", agent="agent-aid", signer=None):
+    signer = signer if signer is not None else signing.Salter(raw=b"0123456789abcdef").signer()
+    rpy = eventing.reply(route=route, data={"agent": agent, "payload": "value"})
+    sig = signer.sign(ser=rpy.raw, index=0)
+    return signer, {"rpy": rpy.ked, "sigs": [sig.qb64]}
+
+
+def test_agent_signals_stream_uses_generic_endpoint():
+    client = mock(strict=True)
+    stream = iter([])
+    expect(client, times=1).stream(
+        "/signals/stream",
+        headers={"Accept": "text/event-stream"},
+    ).thenReturn(stream)
+
+    assert AgentSignals(client).stream() is stream
+    verifyNoUnwantedInteractions()
+    unstub()
+
+
+def test_agent_signals_verify_agent_signed_reply_envelope():
+    signer, envelope = signed_envelope()
+    client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})})
+
+    assert AgentSignals(client).verifyReplyEnvelope(
+        envelope,
+        route="/test/signals/request",
+    ) is True
+
+
+def test_agent_signals_verify_rejects_wrong_route():
+    signer, envelope = signed_envelope(route="/wrong")
+    client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})})
+
+    assert AgentSignals(client).verifyReplyEnvelope(
+        envelope,
+        route="/test/signals/request",
+    ) is False
+
+
+def test_agent_signals_verify_rejects_wrong_agent_payload():
+    signer, envelope = signed_envelope(agent="wrong-agent")
+    client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})})
+
+    assert AgentSignals(client).verifyReplyEnvelope(envelope) is False
+
+
+def test_agent_signals_verify_rejects_missing_signatures():
+    signer, envelope = signed_envelope()
+    envelope["sigs"] = []
+    client = mock({"agent": mock({"pre": "agent-aid", "verfer": signer.verfer})})
+
+    assert AgentSignals(client).verifyReplyEnvelope(envelope) is False
+
+
+def test_agent_signals_verify_rejects_bad_signature():
+    signer, envelope = signed_envelope()
+    other = signing.Salter(raw=b"fedcba9876543210").signer()
+    client = mock({"agent": mock({"pre": "agent-aid", "verfer": other.verfer})})
+
+    assert AgentSignals(client).verifyReplyEnvelope(envelope) is False
+
+
+def test_agent_signals_verify_requires_connected_agent():
+    _, envelope = signed_envelope()
+    client = mock({"agent": None})
+
+    with pytest.raises(kering.ConfigurationError, match="client must be connected"):
+        AgentSignals(client).verifyReplyEnvelope(envelope)

tests/integration/README.md

@@ -56,7 +56,7 @@ The current stack pins those source dependencies to explicit compatibility
 SHAs:
 
 - `keripy`: `4ee02c0213770d25a0114fe7ebd7ab4ab5500cde` (tag `1.2.12`)
-- `keria`: `9e2461550f373ad7bdbe7eebeaceac689cb15397`
+- `keria`: `5b703bd8a60fab68a6476819626b22784317bf14` from `kentbull/keria`
 - `vLEI`: `f514b9431c5f965b5f7f64a8693e19df2f181564` (tag `1.0.2`)
 
 The CI runtime also constrains `hio` to `0.6.14` across the live stack.

tests/integration/_services/keria_server.py

@@ -6,6 +6,8 @@
 from pathlib import Path
 import signal
 
+import falcon
+
 
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
@@ -26,11 +28,52 @@ def configure_temp_log_root(config_dir: str) -> None:
     ogling.Ogler.TempHeadDir = str(temp_head_dir)
 
 
+class TestSignalsEnd:
+    """Harness-only route for exercising KERIA's generic SSE signal channel."""
+
+    def __init__(self, streaming):
+        self.streaming = streaming
+
+    def on_get(self, req, rep):
+        agent = req.context.agent
+        rep.status = falcon.HTTP_200
+        rep.media = {"subscribers": len(agent.sseBroadcaster.subscribers)}
+
+    def on_post(self, req, rep):
+        agent = req.context.agent
+        body = req.media or {}
+        self.streaming.enqueueSignedReplyCue(
+            agent.signalCues,
+            event=body.get("event", "agent.signal.test"),
+            route=body.get("route", "/test/signals/request"),
+            payload=body.get("payload", {}),
+            event_id=body.get("event_id", "test-signal"),
+        )
+        rep.status = falcon.HTTP_202
+        rep.media = {"queued": True}
+
+
+def install_test_signal_route(agenting):
+    """Register an authenticated test-only signal trigger route on KERIA admin."""
+    original_create_admin_server_doer = agenting.createAdminServerDoer
+
+    def create_admin_server_doer(config, agency):
+        admin_app, admin_server_doer = original_create_admin_server_doer(config, agency)
+        from keria.app import streaming
+
+        admin_app.add_route("/test/signals", TestSignalsEnd(streaming))
+        return admin_app, admin_server_doer
+
+    agenting.createAdminServerDoer = create_admin_server_doer
+
+
 def main() -> None:
     args = parse_args()
     configure_temp_log_root(args.config_dir)
     from keria.app import agenting
 
+    install_test_signal_route(agenting)
+
     config = agenting.KERIAServerConfig(
         name="keria",
         base="",

tests/integration/_services/witness_server.py

@@ -46,7 +46,7 @@ def create_loopback_http_server(
     indirecting.createHttpServer = create_loopback_http_server
 
     class NoopQueryEnd:
-        def __init__(self, hab):
+        def __init__(self, hab, **_kwa):
             self.hab = hab
 
         def on_get(self, req, rep):

tests/integration/dependencies.py

@@ -26,8 +26,8 @@ class IntegrationDependency:
 
 KERIA = IntegrationDependency(
     name="KERIA",
-    repo="https://github.com/WebOfTrust/keria.git",
-    ref="9e2461550f373ad7bdbe7eebeaceac689cb15397",
+    repo="https://github.com/kentbull/keria.git",
+    ref="5b703bd8a60fab68a6476819626b22784317bf14",
     path_name="keria",
     env_root="SIGNIFYPY_INTEGRATION_KERIA_ROOT",
     env_ref="SIGNIFYPY_INTEGRATION_KERIA_REF",

tests/integration/test_signaling.py

@@ -0,0 +1,102 @@
+"""Live integration coverage for generic KERIA AgentSignals SSE."""
+
+from __future__ import annotations
+
+import json
+import queue
+import threading
+import time
+
+import pytest
+
+from signify.app.clienting import SignifyClient
+from tests.integration.helpers import POLL_INTERVAL, poll_until
+
+
+pytestmark = pytest.mark.integration
+
+
+def _same_agent_client(client: SignifyClient) -> SignifyClient:
+    """Connect a second HTTP session to the same already-booted KERIA agent."""
+    control = SignifyClient(
+        passcode=client.bran,
+        url=client.url,
+        boot_url=client.boot_url,
+    )
+    control.connect()
+    return control
+
+
+def _wait_for_event(events: queue.Queue, errors: queue.Queue, *, timeout=30):
+    deadline = time.monotonic() + timeout
+    while time.monotonic() < deadline:
+        if not errors.empty():
+            raise errors.get()
+        try:
+            return events.get(timeout=0.1)
+        except queue.Empty:
+            continue
+
+    if not errors.empty():
+        raise errors.get()
+    raise TimeoutError("timed out waiting for SSE signal event")
+
+
+def test_agent_signals_stream_receives_signed_agent_event(client_factory):
+    stream_client = client_factory()
+    control_client = _same_agent_client(stream_client)
+    assert control_client.agent.pre == stream_client.agent.pre
+
+    stream = stream_client.signals().stream()
+    events = queue.Queue()
+    errors = queue.Queue()
+
+    def read_first_data_event():
+        try:
+            for event in stream:
+                if event.data:
+                    events.put(event)
+                    return
+        except Exception as err:  # pragma: no cover - surfaced by main thread
+            errors.put(err)
+
+    reader = threading.Thread(target=read_first_data_event, daemon=True)
+    reader.start()
+
+    try:
+        poll_until(
+            lambda: control_client.get("/test/signals").json(),
+            ready=lambda body: body["subscribers"] >= 1,
+            timeout=20,
+            interval=POLL_INTERVAL,
+            describe="SSE subscription registration",
+        )
+
+        signal = {
+            "event": "agent.signal.test",
+            "event_id": "test-signal-1",
+            "route": "/test/signals/request",
+            "payload": {"subject": "sse-integration"},
+        }
+        response = control_client.post("/test/signals", json=signal)
+        assert response.status_code == 202
+
+        event = _wait_for_event(events, errors)
+        assert event.event == signal["event"]
+        assert event.id == signal["event_id"]
+
+        envelope = json.loads(event.data)
+        rpy = envelope["rpy"]
+        assert rpy["r"] == signal["route"]
+        assert rpy["a"]["subject"] == signal["payload"]["subject"]
+        assert rpy["a"]["agent"] == stream_client.agent.pre
+        assert stream_client.signals().verifyReplyEnvelope(
+            envelope,
+            route=signal["route"],
+        ) is True
+    finally:
+        try:
+            stream.close()
+        except Exception:
+            pass
+        reader.join(timeout=5)