Add SSE events to API v2

AGENTS.md

@@ -26,15 +26,20 @@ mypy simplyblock_web simplyblock_cli simplyblock_core  # Type check (or: tox -e
 
 ## Architecture
 
-Three packages, one entry point:
+Three packages, two front-ends (CLI and Web API) over a shared core:
 
 | Package | Role |
 |---------|------|
 | `simplyblock_cli/` | `sbctl` command-line interface (auto-generated entry point) |
 | `simplyblock_core/` | Business logic, data models, background services, FDB access |
 | `simplyblock_web/` | REST API — FastAPI (v2) + Flask (v1) hybrid on a single uvicorn process |
 
-Data flows: **CLI → Web API → Core controllers → FoundationDB**. Storage nodes are reached via JSON-RPC (`rpc_client.py`).
+Two front-ends sit on top of the core and **both call it in-process** — the CLI does *not* go through the Web API:
+
+- **`sbctl` CLI → Core controllers → FoundationDB.** `clibase.py` imports `simplyblock_core` directly (`cluster_ops`, `storage_node_ops`, the `controllers` package, `DBController`) and instantiates `DBController()` itself.
+- **Web API → Core controllers → FoundationDB.** `simplyblock_web` is a separate entry point serving remote/programmatic clients over the same core.
+
+Storage nodes are reached via JSON-RPC (`rpc_client.py`).
 
 ## Coding Conventions
 
@@ -109,3 +114,13 @@ Edit only `AGENTS.md` files and `.agents/skills/` contents. Never edit the symli
 ### Local overrides
 
 At every level where an `AGENTS.md` exists, also check for a sibling `AGENTS.local.md`. If present, load it in addition to `AGENTS.md` — its contents extend or override the checked-in instructions. `AGENTS.local.md` is gitignored and intended for per-developer notes that should not be committed.
+
+## graphify
+
+This project has a knowledge graph at graphify-out/ with god nodes, community structure, and cross-file relationships.
+
+Rules:
+- For codebase questions, first run `graphify query "<question>"` when graphify-out/graph.json exists. Use `graphify path "<A>" "<B>"` for relationships and `graphify explain "<concept>"` for focused concepts. These return a scoped subgraph, usually much smaller than GRAPH_REPORT.md or raw grep output.
+- If graphify-out/wiki/index.md exists, use it for broad navigation instead of raw source browsing.
+- Read graphify-out/GRAPH_REPORT.md only for broad architecture review or when query/path/explain do not surface enough context.
+- After modifying code, run `graphify update .` to keep the graph current (AST-only, no API cost).

requirements.txt

@@ -23,6 +23,7 @@ sentry-sdk[flask]>=1.40
 flask-openapi3
 jsonschema
 fastapi
+sse-starlette
 uvicorn
 prometheus_api_client
 paramiko

simplyblock_core/cluster_ops.py

@@ -35,6 +35,19 @@
 
 db_controller = DBController()
 
+
+def watch_clusters():
+    """Stream changes across all clusters."""
+    return db_controller.watch(Cluster)
+
+
+def watch_cluster(cluster_id):
+    """Stream changes for a single cluster."""
+    return db_controller.watch(
+        Cluster,
+        select=lambda models: [c for c in models if c.get_id() == cluster_id],
+    )
+
 def _create_update_user(cluster_id, grafana_url, grafana_secret: SecretStr, user_secret: SecretStr, update_secret=False):
     session = requests.session()
     session.auth = ("admin", grafana_secret.get_secret_value())

simplyblock_core/controllers/device_controller.py

@@ -5,6 +5,7 @@
 from simplyblock_core import distr_controller, utils, storage_node_ops, constants
 from simplyblock_core.controllers import device_events, tasks_controller
 from simplyblock_core.db_controller import DBController
+from simplyblock_core.models.cluster import Cluster
 from simplyblock_core.models.nvme_device import NVMeDevice, JMDevice
 from simplyblock_core.models.storage_node import StorageNode
 from simplyblock_core.prom_client import PromClient
@@ -19,6 +20,39 @@
 logger = logging.getLogger()
 
 
+def _explode_devices(nodes, node_id):
+    # Devices are embedded in their StorageNode record; expose them as per-device
+    # entries so the diff yields device-level events. The one place that knows
+    # devices live inside nodes.
+    for node in nodes:
+        if node.get_id() == node_id:
+            return list(node.nvme_devices)
+    return []
+
+
+def watch_devices(cluster_id, node_id):
+    """Stream device changes for one storage node."""
+    db = DBController()
+    return db.watch(
+        StorageNode,
+        select=lambda nodes: _explode_devices(nodes, node_id),
+        ancestors=[(Cluster, cluster_id), (StorageNode, node_id)],
+    )
+
+
+def watch_device(cluster_id, node_id, device_id):
+    """Stream changes for a single device."""
+    db = DBController()
+    return db.watch(
+        StorageNode,
+        select=lambda nodes: [
+            device for device in _explode_devices(nodes, node_id)
+            if device.get_id() == device_id
+        ],
+        ancestors=[(Cluster, cluster_id), (StorageNode, node_id)],
+    )
+
+
 def get_storage_node_by_jm_device(db_controller: DBController, id) -> StorageNode:
     try:
         return next(

simplyblock_core/controllers/lvol_controller.py

@@ -28,6 +28,29 @@
 logger = utils.get_logger(__name__)
 
 
+def watch_volumes(cluster_id, pool_id):
+    """Stream volume changes for one pool (same scope as get_lvols_by_pool_id)."""
+    db = DBController()
+    return db.watch(
+        LVol,
+        select=lambda models: db.get_lvols_by_pool_id(pool_id, source=models),
+        ancestors=[(Cluster, cluster_id), (Pool, pool_id)],
+    )
+
+
+def watch_volume(cluster_id, pool_id, volume_id):
+    """Stream changes for a single volume."""
+    db = DBController()
+    return db.watch(
+        LVol,
+        select=lambda models: [
+            lvol for lvol in db.get_lvols_by_pool_id(pool_id, source=models)
+            if lvol.get_id() == volume_id
+        ],
+        ancestors=[(Cluster, cluster_id), (Pool, pool_id)],
+    )
+
+
 def _create_crypto_lvol(rpc_client, lvol, cluster):
     name = lvol.crypto_bdev
     base_name = f"{lvol.lvs_name}/{lvol.lvol_bdev}"

simplyblock_core/controllers/pool_controller.py

@@ -13,12 +13,33 @@
 from simplyblock_core.controllers import pool_events, lvol_controller
 from simplyblock_core.db_controller import DBController
 from simplyblock_core.kms import KMSException, create_kms_connection, pool_kek_name
+from simplyblock_core.models.cluster import Cluster
 from simplyblock_core.models.pool import Pool
 from simplyblock_core.prom_client import PromClient
 
 logger = lg.getLogger()
 
 
+def watch_pools(cluster_id):
+    """Stream pool changes for one cluster (same scope as get_pools)."""
+    db = DBController()
+    return db.watch(
+        Pool,
+        select=lambda models: db.get_pools(cluster_id, source=models),
+        ancestors=[(Cluster, cluster_id)],
+    )
+
+
+def watch_pool(cluster_id, pool_id):
+    """Stream changes for a single pool."""
+    db = DBController()
+    return db.watch(
+        Pool,
+        select=lambda models: [pool for pool in models if pool.get_id() == pool_id],
+        ancestors=[(Cluster, cluster_id)],
+    )
+
+
 def _generate_string(length):
     return ''.join(random.SystemRandom().choice(
         string.ascii_letters + string.digits) for _ in range(length))

simplyblock_core/controllers/snapshot_controller.py

@@ -19,6 +19,7 @@
 from simplyblock_core.kms._exceptions import KMSException
 from simplyblock_core.db_controller import DBController
 from simplyblock_core.models.job_schedule import JobSchedule
+from simplyblock_core.models.cluster import Cluster
 from simplyblock_core.models.pool import Pool
 from simplyblock_core.models.snapshot import SnapShot
 from simplyblock_core.models.lvol_model import LVol
@@ -27,6 +28,29 @@
 
 logger = lg.getLogger()
 
+
+def watch_snapshots(cluster_id, pool_id):
+    """Stream snapshot changes for one pool (same scope as get_snapshots_by_pool_id)."""
+    db = DBController()
+    return db.watch(
+        SnapShot,
+        select=lambda models: db.get_snapshots_by_pool_id(pool_id, source=models),
+        ancestors=[(Cluster, cluster_id), (Pool, pool_id)],
+    )
+
+
+def watch_snapshot(cluster_id, pool_id, snapshot_id):
+    """Stream changes for a single snapshot."""
+    db = DBController()
+    return db.watch(
+        SnapShot,
+        select=lambda models: [
+            snap for snap in db.get_snapshots_by_pool_id(pool_id, source=models)
+            if snap.get_id() == snapshot_id
+        ],
+        ancestors=[(Cluster, cluster_id), (Pool, pool_id)],
+    )
+
 db_controller = DBController()
 
 

simplyblock_core/controllers/tasks_controller.py

@@ -14,6 +14,28 @@
 logger = logging.getLogger()
 db = db_controller.DBController()
 
+
+def watch_tasks(cluster_id):
+    """Stream task changes for one cluster (excludes device-migration tasks,
+    matching the task list endpoint)."""
+    return db.watch(
+        JobSchedule,
+        select=lambda models: [
+            task for task in db.get_job_tasks(cluster_id, source=models)
+            if task.function_name != JobSchedule.FN_DEV_MIG
+        ],
+        ancestors=[(Cluster, cluster_id)],
+    )
+
+
+def watch_task(cluster_id, task_id):
+    """Stream changes for a single task."""
+    return db.watch(
+        JobSchedule,
+        select=lambda models: [task for task in models if task.uuid == task_id],
+        ancestors=[(Cluster, cluster_id)],
+    )
+
 # Identity used for task leases. Hostname (not pid) so a runner that crashes
 # and restarts on the same host re-claims its own in-flight tasks immediately.
 _RUNNER_HOST = socket.gethostname()

simplyblock_core/db_controller.py

@@ -7,7 +7,7 @@
 import fdb
 from typing import Any, List, Optional
 
-from simplyblock_core import constants
+from simplyblock_core import constants, watches
 from simplyblock_core.models.cluster import Cluster, ClusterAddNodeLock, PortReservation
 from simplyblock_core.models.events import EventObj
 from simplyblock_core.models.job_schedule import JobSchedule
@@ -56,13 +56,22 @@ def __init__(self):
         except Exception:
             logger.exception("FDB initialization failed")
 
+    def watch(self, model_cls, *, select=None, ancestors=()):
+        """Async stream of watch.ChangeEvent batches for a watched model class.
+
+        Thin pass-through to :func:`simplyblock_core.watch.watch` so controllers
+        reach the watch primitive the same way they reach every other DB op.
+        """
+        from simplyblock_core import watch as _watch
+        return _watch.watch(model_cls, select=select, ancestors=ancestors)
+
     def get_storage_nodes(self) -> List[StorageNode]:
         ret = StorageNode().read_from_db(self.kv_store)
         ret = sorted(ret, key=lambda x: x.create_dt)
         return ret
 
-    def get_storage_nodes_by_cluster_id(self, cluster_id: str) -> List[StorageNode]:
-        ret = StorageNode().read_from_db(self.kv_store)
+    def get_storage_nodes_by_cluster_id(self, cluster_id: str, *, source=None) -> List[StorageNode]:
+        ret = source if source is not None else StorageNode().read_from_db(self.kv_store)
         nodes = []
         for n in ret:
             if n.cluster_id == cluster_id:
@@ -101,15 +110,11 @@ def get_storage_device_by_id(self, id: str) -> NVMeDevice:
         return device
 
 
-    def get_pools(self, cluster_id: Optional[str] = None) -> List[Pool]:
-        pools = []
+    def get_pools(self, cluster_id: Optional[str] = None, *, source=None) -> List[Pool]:
+        all_pools = source if source is not None else Pool().read_from_db(self.kv_store)
         if cluster_id:
-            for pool in Pool().read_from_db(self.kv_store):
-                if pool.cluster_id == cluster_id:
-                    pools.append(pool)
-        else:
-            pools = Pool().read_from_db(self.kv_store)
-        return pools
+            return [pool for pool in all_pools if pool.cluster_id == cluster_id]
+        return all_pools
 
     def get_pool_by_id(self, id: str) -> Pool:
         pool = single_or_none(Pool().read_from_db(self.kv_store, id))
@@ -123,8 +128,8 @@ def get_pool_by_name(self, name: str) -> Pool:
             raise KeyError(f'Pool {name} not found')
         return pool
 
-    def get_lvols(self, cluster_id: Optional[str] = None) -> List[LVol]:
-        lvols = self.get_all_lvols()
+    def get_lvols(self, cluster_id: Optional[str] = None, *, source=None) -> List[LVol]:
+        lvols = source if source is not None else self.get_all_lvols()
         lvols = [lvol for lvol in lvols if lvol.status != LVol.STATUS_DELETED]
         if not cluster_id:
             return lvols
@@ -155,9 +160,9 @@ def get_lvols_by_node_id(self, node_id: str) -> List[LVol]:
                 lvols.append(lvol)
         return sorted(lvols, key=lambda x: x.create_dt)
 
-    def get_lvols_by_pool_id(self, pool_id: str) -> List[LVol]:
+    def get_lvols_by_pool_id(self, pool_id: str, *, source=None) -> List[LVol]:
         lvols = []
-        for lvol in self.get_lvols():
+        for lvol in self.get_lvols(source=source):
             if lvol.pool_uuid == pool_id:
                 lvols.append(lvol)
         return sorted(lvols, key=lambda x: x.create_dt)
@@ -170,9 +175,9 @@ def get_hostnames_by_pool_id(self, pool_id: str) -> List[str]:
                 hostnames.append(lv.hostname)
         return hostnames
 
-    def get_snapshots(self, cluster_id: Optional[str] = None) -> List[SnapShot]:
+    def get_snapshots(self, cluster_id: Optional[str] = None, *, source=None) -> List[SnapShot]:
         start_time = time.time()
-        snaps = SnapShot().read_from_db(self.kv_store)
+        snaps = source if source is not None else SnapShot().read_from_db(self.kv_store)
         if cluster_id:
             snaps = [n for n in snaps if n.cluster_id == cluster_id]
         ret = sorted(snaps, key=lambda x: x.created_at)
@@ -283,7 +288,9 @@ def get_device_capacity(self, device, limit=1) -> List[DeviceStatObject]:
             self.kv_store, id="%s/%s" % (device.cluster_id, device.get_id()), limit=limit, reverse=True)
         return stats
 
-    def get_clusters(self) -> List[Cluster]:
+    def get_clusters(self, *, source=None) -> List[Cluster]:
+        if source is not None:
+            return source
         return Cluster().read_from_db(self.kv_store)
 
     def get_cluster_by_id(self, cluster_id: str) -> Cluster:
@@ -299,8 +306,11 @@ def get_port_stats(self, node_id: str, port_id: str, limit: int = 20) -> List[Po
     def get_events(self, event_id: str = " ", limit: int = 0, reverse: bool = False) -> List[EventObj]:
         return EventObj().read_from_db(self.kv_store, id=event_id, limit=limit, reverse=reverse)
 
-    def get_job_tasks(self, cluster_id: str, reverse: bool = True, limit: int = 0) -> List[JobSchedule]:
-        ret = JobSchedule().read_from_db(self.kv_store, id=cluster_id, reverse=reverse, limit=limit)
+    def get_job_tasks(self, cluster_id: str, reverse: bool = True, limit: int = 0, *, source=None) -> List[JobSchedule]:
+        if source is not None:
+            ret = [t for t in source if t.cluster_id == cluster_id]
+        else:
+            ret = JobSchedule().read_from_db(self.kv_store, id=cluster_id, reverse=reverse, limit=limit)
         return sorted(ret, key=lambda x: x.date)
 
 
@@ -326,9 +336,9 @@ def get_snapshots_by_node_id(self, node_id: str) -> List[SnapShot]:
                 ret.append(snap)
         return sorted(ret, key=lambda x: x.create_dt)
 
-    def get_snapshots_by_pool_id(self, pool_id: str) -> List[SnapShot]:
+    def get_snapshots_by_pool_id(self, pool_id: str, *, source=None) -> List[SnapShot]:
         ret = []
-        snaps = self.get_snapshots()
+        snaps = self.get_snapshots(source=source)
         for snap in snaps:
             if snap.pool_uuid == pool_id:
                 ret.append(snap)
@@ -675,6 +685,8 @@ def _atomic_update_tx(self, tr, key, model_cls, mutate_fn):
         if mutate_fn(obj) is False:
             return obj
         tr[key] = json.dumps(obj.to_dict(unwrap_secrets=True)).encode()
+        if getattr(model_cls, '_WATCHED', False):
+            tr.add(watches.watch_counter_key(model_cls), watches.ONE_LE64)
         return obj
 
     def atomic_update(self, obj, mutate_fn):
@@ -900,6 +912,7 @@ def _try_set_node_restarting_tx(self, tr, cluster_id, node_id):
             prefix = target.get_db_id()
             data = json.dumps(target.get_clean_dict(unwrap_secrets=True))
             tr[prefix.encode()] = data.encode()
+            tr.add(watches.watch_counter_key(StorageNode), watches.ONE_LE64)
 
         return True, None