Merge pull request #980 from jupyter-naas/claude/eloquent-cannon-6b0db4

feat(core): add EventService for durable, queryable events

Author: Dr0p42

config.local.yaml

@@ -252,6 +252,16 @@ services:
       config:
         rabbitmq_url: "amqp://{{ secret.RABBITMQ_USER }}:{{ secret.RABBITMQ_PASSWORD }}@rabbitmq:5672"
 
+  # Q: Where is the durable event log stored, and how do live listeners find each other?
+  # A: SQLite holds every published event for query and replay; the bus above carries
+  #    live broadcasts to subscribers. Two systems, one contract: SQLite owns durability,
+  #    the bus owns delivery. Late or restarted consumers catch up with iter_query_for_consumer.
+  event:
+    event_adapter:
+      adapter: "sqlite"
+      config:
+        db_path: "storage/events/events.sqlite"
+
   # Q: Where is temporary data kept (sessions, caches, locks)?
   # A: Redis handles it. Swap to "python" for an in-memory dict with no Redis container.
   kv:

docs/reference/naas_abi_core/services/event/EventFactory.md

@@ -0,0 +1,28 @@
+# EventFactory
+
+## What it is
+Factory for building a default `EventService` backed by `EventSQLiteAdapter`. The database file is placed under the project's `storage/` folder (created if absent).
+
+## Public API
+
+### `class EventFactory`
+- `EventSQLite_find_storage(bus: BusService | None = None, subpath: str = "events.sqlite", needle: str = "storage") -> EventService`
+  - Locates (or creates) the `storage/` folder via `naas_abi_core.utils.Storage.find_storage_folder`.
+  - Places the SQLite file under `<storage>/events/<subpath>` (subpath is prefixed with `events/` if not already).
+  - Returns an `EventService` wired to that adapter and the provided bus.
+
+## Configuration/Dependencies
+- `EventSQLiteAdapter` (sibling adapter).
+- `BusService` (optional).
+- `naas_abi_core.utils.Storage.find_storage_folder`.
+
+## Usage
+
+```python
+from naas_abi_core.services.event.EventFactory import EventFactory
+
+events = EventFactory.EventSQLite_find_storage(bus=bus_service)
+```
+
+## Caveats
+- If no `bus` is passed, `events.subscribe(...)` raises `RuntimeError`. `publish` and `query` still work — the event log remains durable.

docs/reference/naas_abi_core/services/event/EventPort.md

@@ -0,0 +1,60 @@
+# EventPort
+
+## What it is
+Interface (port) definitions for a durable event log with live broadcast:
+- A `StoredEvent` record returned by the durable log.
+- An adapter interface (`IEventAdapter`) for storage backends.
+- A service interface (`IEventService`) for higher-level event operations (publish, query, query-for-consumer, subscribe).
+- Event-related exceptions (`EventNotFoundError`, `InvalidEventError`).
+
+Events themselves are not modeled in this port: a published event must be an instance of a `LogProcess` subclass (an `onto2py`-generated Pydantic class whose class IRI identifies the event type).
+
+## Public API
+
+### Exceptions
+- `EventNotFoundError`: missing event lookup.
+- `InvalidEventError`: raised when `publish()` receives an object that is not a `LogProcess` subclass instance.
+
+### Models
+- `StoredEvent(dataclass, frozen=True)`
+  - Fields:
+    - `id: str` — instance IRI (the `LogProcess` instance's `_uri`).
+    - `event_type: str` — class IRI (the `LogProcess` subclass's `_class_uri`).
+    - `seq: int` — global monotonic sequence assigned by the adapter on append.
+    - `timestamp: str` — ISO 8601 timestamp.
+    - `payload: bytes` — serialized RDF graph (n-triples).
+
+### Interfaces
+
+#### `IEventAdapter` (secondary port — durable log)
+- `append(event_id, event_type, timestamp, payload) -> StoredEvent` — persist one event; returns the stored record with its assigned `seq`.
+- `query(event_type=None, since_seq=None, until_seq=None, since_timestamp=None, until_timestamp=None, limit=None) -> list[StoredEvent]` — filtered read, ordered by `seq` ascending.
+- `max_seq(event_type=None) -> int` — highest stored `seq` (0 if empty). Used by iterators to capture a snapshot upper bound.
+- `get_cursor(consumer_id, event_type) -> int` — return last delivered `seq` for `(consumer_id, event_type)`; `0` if unset.
+- `query_for_consumer(consumer_id, event_type, limit=None) -> list[StoredEvent]` — return events with `seq > cursor` and advance the cursor atomically in the same transaction.
+
+#### `IEventService` (primary port)
+- `publish(event) -> StoredEvent` — persist a `LogProcess` subclass instance and broadcast it on the bus. Raises `InvalidEventError` if `event` is not a `LogProcess` subclass instance.
+- `query(event_class=None, since_seq=None, until_seq=None, since_timestamp=None, until_timestamp=None, limit=None) -> list` — eager read. Return reconstructed event instances matching the filters.
+- `iter_query(event_class=None, since_seq=None, since_timestamp=None, until_timestamp=None, batch_size=500) -> Iterator` — streaming read with snapshot semantics; caller doesn't manage pagination.
+- `query_for_consumer(consumer_id, event_class, limit=None) -> list` — eager. Return undelivered events for `consumer_id` and advance the cursor.
+- `iter_query_for_consumer(consumer_id, event_class, batch_size=500) -> Iterator` — drain all pending events for `consumer_id` across batches, advancing the cursor each batch.
+- `subscribe(event_class, callback, routing_key="#") -> Thread` — subscribe to live events via the bus. **Live-only — does not replay history.** Use `query_for_consumer` / `iter_query_for_consumer` for catch-up.
+
+## Configuration/Dependencies
+- Standard library: `abc`, `dataclasses`, `threading`, `typing`.
+- No runtime configuration is defined in this module.
+
+## Usage
+
+```python
+from naas_abi_core.services.event.EventPort import IEventService
+from naas_abi_core.services.event.ontologies.modules.EventOntology import LogProcess
+
+def emit(svc: IEventService, evt: LogProcess) -> None:
+    svc.publish(evt)
+```
+
+## Caveats
+- `subscribe(...)` is **live-only**: subscribers who are not connected at publish time will not receive the event. Late or restarted subscribers must catch up via `query_for_consumer`.
+- The cursor in `query_for_consumer` advances **on read**, before processing. A consumer that crashes after the call returns but before processing the events will not see those events again — this is at-most-once semantics by design (v1).

docs/reference/naas_abi_core/services/event/EventService.md

@@ -0,0 +1,117 @@
+# EventService
+
+## What it is
+A durable event log with live broadcast. Each `publish(event)` is:
+1. **Persisted** to the secondary adapter (durability contract).
+2. **Broadcast** on the `BusService` topic derived from the event's class IRI (live signaling).
+
+Events are instances of `LogProcess` subclasses — Pydantic classes generated from TTL via `onto2py`. The class IRI (`_class_uri`) acts as the event type; the instance IRI (`_uri`) is the event id.
+
+Subscriptions are live-only. Historical reads go through `query()` (filter-based) or `query_for_consumer()` (per-consumer cursor with auto-advance).
+
+---
+
+## Public API
+
+### `class EventService(ServiceBase, IEventService)`
+
+- `__init__(adapter: IEventAdapter, bus: BusService | None = None)`
+  - Requires an adapter. A bus is optional but required for `subscribe()`.
+- `publish(event: LogProcess) -> StoredEvent`
+  - Persists via the adapter, then broadcasts on the bus topic `class_iri_to_topic(event._class_uri)`. If the bus broadcast fails, the event is still persisted (warning logged).
+  - **Populates `event.created_at = datetime.now()` if the caller didn't set it**, before serializing — so the timestamp round-trips through `query`/`iter_query`.
+  - Raises `InvalidEventError` if `event` is not a `LogProcess` subclass instance.
+- `query(event_class=None, since_seq=None, until_seq=None, since_timestamp=None, until_timestamp=None, limit=None) -> list[LogProcess]`
+  - Eager read. Reconstructs each row back into an instance of `event_class` (or `LogProcess` if no class hint is given). Use `since_seq` for stable seq-based pagination.
+- `iter_query(event_class=None, since_seq=None, since_timestamp=None, until_timestamp=None, limit=None, batch_size=500) -> Iterator[LogProcess]`
+  - Streaming read with **snapshot semantics**: captures the current `max(seq)` at the first call and stops once it has yielded everything up to that point. Events appended during iteration are not included — call again to pick them up. Caller does not have to manage pagination.
+  - `limit` caps the total number of events yielded; `batch_size` is the per-round-trip SQL fetch size (defaults to 500).
+- `query_for_consumer(consumer_id: str, event_class: type, limit=None) -> list[LogProcess]`
+  - Eager. Returns undelivered events for `(consumer_id, event_class._class_uri)` and advances the cursor in the same transaction.
+- `iter_query_for_consumer(consumer_id, event_class, limit=None, batch_size=500) -> Iterator[LogProcess]`
+  - Drains pending events for `consumer_id` in batches, advancing the cursor per batch. Stops when caught up or when `limit` events have been yielded. The cursor only advances over events actually returned, so a stopped iteration leaves the remainder pending for the next call.
+- `subscribe(event_class: type, callback: Callable[[LogProcess], None], routing_key: str = "#") -> Thread`
+  - Subscribes to the bus topic for `event_class._class_uri`. The callback receives a reconstructed instance of `event_class`. Requires a `BusService`; raises `RuntimeError` otherwise.
+
+### Module functions
+- `class_iri_to_topic(class_iri: str) -> str`
+  - Maps a class IRI to a bus-safe topic name: `"evt." + sha256(class_iri)[:32]`. Deterministic; avoids characters that RabbitMQ topic names dislike (`://`, `#`).
+
+---
+
+## Configuration/Dependencies
+- `IEventAdapter` implementation (defaults: `EventSQLiteAdapter`).
+- `BusService` for live broadcast and subscription (optional but required for `subscribe()`).
+- `naas_abi_core.services.event.event_ontology.LogProcess` (generated from `services/event/event.ttl` via `onto2py`).
+- Logging via `naas_abi_core.logger`.
+
+---
+
+## Usage
+
+### Define an event type (TTL → onto2py)
+```turtle
+# my_events.ttl
+@prefix abi: <http://ontology.naas.ai/abi/> .
+@prefix ex:  <http://example.org/> .
+
+ex:UserAuthenticated a owl:Class ;
+    rdfs:subClassOf abi:LogProcess ;
+    rdfs:label "user authenticated"@en .
+```
+
+Generate the Python module:
+```bash
+python -m naas_abi_core.utils.onto2py my_events.ttl my_events.py
+```
+
+### Publish, query, subscribe
+```python
+from naas_abi_core.services.event.EventFactory import EventFactory
+from my_events import UserAuthenticated
+
+events = EventFactory.EventSQLite_find_storage(bus=bus_service)
+
+# Live subscriber (live-only, no history replay)
+def on_user_auth(evt: UserAuthenticated) -> None:
+    print("auth:", evt.user_id)
+events.subscribe(UserAuthenticated, on_user_auth)
+
+# Publish
+events.publish(UserAuthenticated(user_id="alice"))
+
+# Historical query (eager)
+recent = events.query(event_class=UserAuthenticated, limit=100)
+
+# Streaming over all history without managing pagination (snapshot)
+for evt in events.iter_query(event_class=UserAuthenticated, batch_size=500):
+    process(evt)
+
+# Catch-up for a durable consumer (eager, one batch)
+batch = events.query_for_consumer("worker-1", UserAuthenticated)
+
+# Catch-up streaming until the consumer is up to date
+for evt in events.iter_query_for_consumer("worker-1", UserAuthenticated):
+    process(evt)
+```
+
+---
+
+## Reconstructed events
+
+`query`, `iter_query`, `query_for_consumer`, and `iter_query_for_consumer` all return **instances of the requested `event_class`** (a Pydantic model), not a wrapper. Domain fields you set at publish-time are populated; storage metadata is attached as private attributes:
+
+```python
+for evt in events.iter_query(event_class=UserAuthenticated):
+    evt.user_id        # domain field set at publish-time
+    evt.created_at     # always populated (auto-set by publish if unset)
+    evt._uri           # original instance IRI
+    evt._seq           # global monotonic sequence (storage metadata)
+    evt._stored_at     # ISO timestamp persisted by the adapter
+```
+
+## Caveats
+- **Live-only subscribe**: events published before a subscriber connects are not delivered to it. Use `query_for_consumer` for catch-up.
+- **At-most-once cursor**: `query_for_consumer` advances on read, before the caller processes. If the caller crashes after return, those events will not be returned again.
+- **Bus failure does not lose events**: persistence happens first; bus failures are logged and swallowed.
+- **Reconstruction without hint class**: `query()` without an `event_class` returns instances of the most-specific Python class found by walking `LogProcess.__subclasses__()`. If the event's class is not imported anywhere in the running process, the row is reconstructed as a bare `LogProcess`.

docs/reference/naas_abi_core/services/event/adapters/secondary/EventSQLiteAdapter.md

@@ -0,0 +1,48 @@
+# EventSQLiteAdapter
+
+## What it is
+SQLite-backed `IEventAdapter`. Single file, WAL mode, two tables:
+- `events` — durable append-only log with a global monotonic `seq` (autoincrement primary key).
+- `consumer_cursors` — per-(consumer_id, event_type) read position used by `query_for_consumer`.
+
+## Schema
+```sql
+CREATE TABLE events (
+    seq        INTEGER PRIMARY KEY AUTOINCREMENT,
+    id         TEXT NOT NULL UNIQUE,
+    event_type TEXT NOT NULL,
+    timestamp  TEXT NOT NULL,
+    payload    BLOB NOT NULL
+);
+CREATE INDEX idx_events_type_seq  ON events(event_type, seq);
+CREATE INDEX idx_events_timestamp ON events(timestamp);
+
+CREATE TABLE consumer_cursors (
+    consumer_id TEXT NOT NULL,
+    event_type  TEXT NOT NULL,
+    last_seq    INTEGER NOT NULL DEFAULT 0,
+    updated_at  TEXT NOT NULL,
+    PRIMARY KEY (consumer_id, event_type)
+);
+```
+
+## Public API
+
+### `class EventSQLiteAdapter(IEventAdapter)`
+- `__init__(db_path: str)`
+  - Opens (and creates if missing) the SQLite file at `db_path`. Parent directories are created.
+  - Sets `journal_mode=WAL` and `synchronous=NORMAL`.
+- `close() -> None`
+- `append(event_id, event_type, timestamp, payload) -> StoredEvent`
+- `query(event_type=None, since_seq=None, since_timestamp=None, until_timestamp=None, limit=None) -> list[StoredEvent]`
+- `get_cursor(consumer_id, event_type) -> int`
+- `query_for_consumer(consumer_id, event_type, limit=None) -> list[StoredEvent]`
+  - Atomic: read + cursor advance happen inside `BEGIN IMMEDIATE`.
+
+## Configuration/Dependencies
+- Standard library: `sqlite3`, `threading`, `os`, `datetime`.
+
+## Caveats
+- One file, one writer at a time (WAL allows concurrent readers). Heavy multi-process write contention may require splitting (not v1).
+- `append` raises `sqlite3.IntegrityError` if `event_id` already exists. Caller (the service layer) is expected to use unique instance IRIs.
+- `query_for_consumer` is **at-most-once on read**: the cursor advances before the caller processes. There is no ack/nack in v1.