Topics

example/Actions.vala

@@ -84,6 +84,7 @@ namespace Example {
             yield state_service.update_global_typed<AnnouncementPublic>("announcement", a => {
                 a.text = text;
             });
+            yield topic_registry.trigger("announcement:latest");
 
             return directives().notify("info", "Announcement updated").navigate("/dashboard");
         }

example/HomeEntrypoint.vala

@@ -36,6 +36,42 @@ namespace Example {
         public string text { get; set; }
     }
 
+    /** Derived model — a short preview of the announcement, produced by the topic modifier. */
+    public class AnnouncementPreview : Object {
+        public string summary { get; set; }
+        public int chars { get; set; }
+    }
+
+    /** Topic: loads the current announcement from the global slot when triggered. */
+    public class AnnouncementTopic : Statum.Topic<AnnouncementPublic> {
+        private StateService state_service = Inversion.inject<StateService>();
+
+        public override async AnnouncementPublic load(string topic_key) throws GLib.Error {
+            var slot = state_service.get_slot(StateService.GLOBAL_KEY_PREFIX + "announcement");
+            var model = new AnnouncementPublic();
+            if (slot != null && ((!)slot).current_state != null) {
+                try {
+                    model = GObjectMapping.from_properties_typed<AnnouncementPublic>(
+                        typeof(AnnouncementPublic), ((!)slot).current_state.public_data);
+                } catch {}
+            }
+            return model;
+        }
+    }
+
+    /** Modifier: derives an AnnouncementPreview from the AnnouncementPublic payload. */
+    public class AnnouncementPreviewModifier : Statum.StateModifier {
+        public override Type state_type { get { return typeof(AnnouncementPreview); } }
+
+        public override GLib.Object? derive(GLib.Object payload, GLib.Object? current, HeldStates held) {
+            var announcement = (AnnouncementPublic) payload;
+            var preview = new AnnouncementPreview();
+            preview.chars = announcement.text.length;
+            preview.summary = announcement.text.length > 40 ? announcement.text.substring(0, 40) + "…" : announcement.text;
+            return preview;
+        }
+    }
+
     /**
      * Entrypoint for the home page: hydrates counter + page slots with typed
      * models, embedding action references as fields.
@@ -54,11 +90,26 @@ namespace Example {
             var page = new PagePublic();
             page.login = action_registry.author<LoginAction>();
 
+            // Create a derived "preview" slot and register a topic watcher so it
+            // updates live when the announcement changes.
+            var preview_model = new AnnouncementPreview();
+            preview_model.summary = "";
+            preview_model.chars = 0;
+            var preview_slot = state_service.new_slot(Scope.PAGE, new State() {
+                type_name = "preview",
+                public_data = GObjectMapping.to_properties(preview_model),
+                private_data = new PropertyDictionary()
+            });
+            topic_registry.watch("announcement:latest", preview_slot.id,
+                typeof(AnnouncementPreview).name(), request.held);
+
             return directives()
                 .set_global("announcement")
                 .subscribe_global("announcement")
                 .set_typed<CounterPublic>("counter", Scope.PAGE, counter)
-                .set_typed<PagePublic>("page", Scope.PAGE, page);
+                .set_typed<PagePublic>("page", Scope.PAGE, page)
+                .set(preview_slot)
+                .subscribe(preview_slot.id);
         }
     }
 

example/Main.vala

@@ -30,6 +30,11 @@ void main(string[] args) {
         announcement.text = "Welcome to the Statum demo!";
         statum.global<AnnouncementPublic>("announcement", announcement);
 
+        // Register a topic + modifier: when triggered, derives a short preview
+        // from the current announcement and pushes it to watching clients.
+        statum.topic<AnnouncementTopic>("announcement");
+        statum.topic_modifier<AnnouncementPreviewModifier, AnnouncementPreview>("announcement");
+
         // Background timer: cycle the announcement every 10 seconds. Every
         // subscribed tab receives the update via the SSE push — open two tabs
         // side by side to see the broadcast in action.
@@ -52,6 +57,10 @@ void main(string[] args) {
                     private_data = new PropertyDictionary()
                 };
                 statum.state_service.update_global.begin("announcement", state);
+                // Trigger the topic so derived "preview" slots update too.
+                statum.topic_registry.trigger.begin("announcement:latest", () => {
+                    stderr.printf("[topic] trigger completed\n");
+                });
             } catch (GLib.Error e) {
                 warning("Failed to update announcement: %s", e.message);
             }

example/home.html

@@ -6,6 +6,7 @@
 </head>
 <body>
     <p><strong>Announcement:</strong> <span stm-text="announcement.text">…</span></p>
+    <p><em>Preview:</em> <span stm-text="preview.summary">…</span> (<span stm-text="preview.chars">0</span> chars)</p>
 
     <!-- A component (fragment): its <head> merges into the page head and its
          body is materialised into the wrapping element. -->

getting-started.md

@@ -546,3 +546,57 @@ statum-genkeys > web-config.json
 
 Without these, the server generates random keys per restart (fine for
 development; a warning is logged).
+
+### Topic system
+
+The topic system lets external services (workers, webhooks, timers) push updates
+to clients watching a specific entity. A **topic** loads the entity from the
+database when triggered; **state modifiers** derive specific slot types from the
+payload and push them to subscribed clients via SSE.
+
+```vala
+// A typed topic: loads the entity once per trigger
+public class CatTopic : Topic<Cat> {
+    private Database db = Inversion.inject<Database>();
+    public override async Cat load(string topic_key) throws GLib.Error {
+        return yield db.load_cat(int.parse(topic_key.split(":")[1]));
+    }
+}
+
+// A state modifier: derives a slot type from the payload + held context
+public class CatDetailModifier : StateModifier<Cat, CatDetail> {
+    public override CatDetail derive(Cat cat, CatDetail current, HeldStates held) {
+        if (current != null && current.breed != cat.breed) return current; // skip
+        return new CatDetail() { breed = cat.breed, description = cat.description };
+    }
+}
+
+// Registration
+statum.topic<CatTopic>("cat");
+statum.topic_modifier<CatDetailModifier, CatDetail>("cat");
+
+// Entrypoint: watch a topic
+var slot = state_service.new_slot(Scope.PAGE, state);
+topics.watch("cat:42", slot.id, typeof(CatDetail).name(), request.held);
+return directives().set(slot).subscribe(slot.id);
+
+// Trigger from anywhere (worker, webhook, timer)
+yield topic_registry.trigger("cat:42");
+```
+
+`derive` return values: new instance → push via SSE; `current` (same ref) → skip; `null` → clear slot + remove watcher. Exceptions are treated as `null`.
+
+`HeldStates` provides a live lookup of the client's held slots at trigger time (reads current cache values):
+
+```vala
+var auth = held.get<AuthPublic>("auth");  // null if expired
+```
+
+| Component | Description |
+|---|---|
+| `Topic<TPayload>` | Base class. Override `load(topic_key)` to produce the payload. |
+| `StateModifier<TPayload, TState>` | Base class. Override `derive(payload, current, held)`. |
+| `HeldStates` | Live lookup of client-held state by type name. |
+| `TopicRegistry` | `watch(topic, slot_key, slot_type, request_held)`, `trigger(topic)`, `sweep()`. |
+| `statum.topic<T>(prefix)` | Registers a topic handler. |
+| `statum.topic_modifier<M, S>(prefix)` | Registers a state modifier. |

plans/topic-system-plan.md

@@ -0,0 +1,378 @@
+# Topic System Plan — Typed topics, state modifiers, and live held-state lookup
+
+## Overview
+
+A server-side pub/sub system that lets external services (workers, webhooks,
+other actions) push updates to clients watching a specific entity. Built on top
+of the existing slot/SSE pipeline — no new transport.
+
+A **topic** (`Topic<TPayload>`) loads the entity from the database when
+triggered. **State modifiers** (`StateModifier<TPayload, TState>`) receive the
+payload + the watcher's current slot state + a live lookup of all the client's
+held states, and decide whether to push an update (return new state), skip
+(return `current`), or revoke (return `null`).
+
+---
+
+## Design
+
+### `Topic<TPayload>` — loads the entity once per trigger
+
+```vala
+public abstract class Topic<TPayload> : Object {
+    /** Load the payload for `topic_key` (e.g. "cat:42" → Cat from the database). */
+    public abstract async TPayload load(string topic_key) throws GLib.Error;
+}
+```
+
+Runs once per trigger (not per watcher). DI-constructed — can inject databases,
+APIs, etc.
+
+### `StateModifier<TPayload, TState>` — derives slot state per watcher
+
+```vala
+public abstract class StateModifier<TPayload, TState> : Object {
+    /**
+     * @param payload  The trigger payload (loaded once by Topic<T>).
+     * @param current  The current state of the slot being modified, or null if
+     *                 it has expired from the cache.
+     * @param held     Live lookup of the client's currently-held states by
+     *                 type_name (reads CURRENT cache values at trigger time).
+     * @return New TState → push; `current` (same ref) → skip; null → clear + remove.
+     */
+    public abstract TState? derive(TPayload payload, TState? current, HeldStates held);
+}
+```
+
+### `HeldStates` — live lookup of client state at trigger time
+
+```vala
+public class HeldStates : Object {
+    /** Typed current state for a held type, or null if expired/dropped. */
+    public T? get<T>(string type_name);
+}
+```
+
+Internally holds a `type_name → slot_key` map (captured at watch time from
+`request.held`). On each `get<T>`, reads the CURRENT state from
+`state_service.get_slot(key)`. If the slot expired, returns null.
+
+### Return-value semantics
+
+| `derive` returns | Action |
+|---|---|
+| New `TState` instance | `state_service.update(slot_key, state)` → SSE push |
+| `current` (same reference) | No push (unchanged / not relevant) |
+| `null` | `state_service.clear_slot(slot_key)` → SSE clear + remove watcher |
+| Throws exception | Same as `null`: clear slot + remove watcher |
+
+---
+
+## Registration
+
+```vala
+// Startup
+statum.topic<CatTopic>("cat");
+statum.topic_modifier<CatBreedDetailModifier, Cat, CatBreedDetail>("cat");
+statum.topic_modifier<CatSummaryModifier, Cat, CatSummary>("cat");
+```
+
+`statum.topic<T>(prefix)` registers the topic handler for a key prefix (e.g.
+"cat" matches "cat:42", "cat:99").
+
+`statum.topic_modifier<M, TPayload, TState>(prefix)` registers a state modifier
+for the prefix. Multiple modifiers can share a prefix (one per slot type).
+
+## Entrypoint — watch with held context
+
+```vala
+public class CatDetailEntrypoint : StatumEntrypoint {
+    private TopicRegistry topics = inject<TopicRegistry>();
+
+    public override async DirectiveBuilder handle() throws GLib.Error {
+        var cat = yield db.find_cat_by_breed(request.route("breed"));
+
+        var detail_slot = state_service.new_slot(Scope.PAGE, to_detail_state(cat));
+        var summary_slot = state_service.new_slot(Scope.PAGE, to_summary_state(cat));
+
+        // Capture the full held context (type_name → slot_key) so modifiers
+        // can query auth, related entities, etc. at trigger time.
+        topics.watch(@"cat:$(cat.id)", detail_slot.id, typeof(CatBreedDetail).name(), request.held);
+        topics.watch(@"cat:$(cat.id)", summary_slot.id, typeof(CatSummary).name(), request.held);
+
+        return directives()
+            .set(detail_slot)
+            .set(summary_slot)
+            .subscribe(detail_slot.id)
+            .subscribe(summary_slot.id);
+    }
+}
+```
+
+## Trigger
+
+```vala
+// From any context (worker, webhook, admin action)
+yield topics.trigger("cat:42");
+```
+
+## What the registry does on trigger
+
+1. Resolve the topic handler by prefix ("cat" → `CatTopic`).
+2. Check if there are any watchers for `topic_key`. If none, **skip the load
+   entirely** — no database query, no DI construction. The trigger is a no-op.
+3. `topic.load("cat:42")` → `Cat` (one DB query).
+4. Find all modifiers for the prefix.
+5. For each watcher:
+   a. Find the modifier whose `state_type_name` matches the watcher's slot type.
+   b. Read `current` from `state_service.get_slot(watcher.slot_key)`.
+   c. Build `HeldStates` from `watcher.held_keys`.
+   d. Call `modifier.derive(payload, current, held)`.
+   e. If result is null OR the call throws → `clear_slot` + remove watcher.
+   f. If result == `current` (reference equality) → skip.
+   g. If result is a new instance → `update(slot_key, state)` → SSE push.
+
+## Cleanup
+
+| Event | Action |
+|---|---|
+| `derive` returns null | `clear_slot` + remove watcher |
+| `derive` throws | Same as null: `clear_slot` + remove watcher |
+| SSE disconnect (navigate away) | `update` push is no-op; periodic sweep removes stale watchers |
+| Server restart | Watchers lost from memory; re-registered on next entrypoint load |
+| Slot expired from cache | `trigger` catches `SLOT_NOT_FOUND` in `update` → removes watcher |
+| Periodic sweep (every 5 min) | Checks `get_slot(key)` for each watcher; removes if null |
+
+---
+
+## Tasks
+
+### V-1: `HeldStates` — `src/TopicSystem.vala` (or `src/HeldStates.vala`)
+
+```vala
+public class HeldStates : Object {
+    private Dictionary<string, string> slot_keys;
+    private StateService state_service;
+
+    internal HeldStates(Dictionary<string, string> slot_keys, StateService state_service) { ... }
+
+    public T? get<T>(string type_name) { ... }
+    public Properties? get_properties(string type_name) { ... }
+}
+```
+
+`get<T>` reads the current state from `state_service.get_slot`, maps to `T` via
+`GObjectMapping.from_properties_typed<T>`.
+
+### V-2: `Topic<TPayload>` base class — `src/TopicSystem.vala`
+
+```vala
+public abstract class Topic<TPayload> : Object {
+    public abstract async TPayload load(string topic_key) throws GLib.Error;
+}
+```
+
+### V-3: `StateModifier<TPayload, TState>` + non-generic bridge — `src/TopicSystem.vala`
+
+The generic class stores the payload/state types. A non-generic internal
+interface (`TopicStateModifier`) lets the registry store heterogeneous modifiers:
+
+```vala
+internal interface TopicStateModifier : Object {
+    public abstract string state_type_name { get; }
+    public abstract async ModifierOutcome apply(Object payload, string slot_key, HeldStates held) throws GLib.Error;
+}
+
+public enum ModifierOutcome { UPDATED, UNCHANGED, CLEARED }
+
+public abstract class StateModifier<TPayload, TState> : Object, TopicStateModifier {
+    protected StateService state_service = inject<StateService>();
+
+    public string state_type_name { get { return typeof(TState).name(); } }
+
+    public abstract TState? derive(TPayload payload, TState? current, HeldStates held);
+
+    public async ModifierOutcome apply(Object payload, string slot_key, HeldStates held) throws GLib.Error {
+        var slot = state_service.get_slot(slot_key);
+        TState? current = null;
+        if (slot != null && slot.current_state != null) {
+            try {
+                current = GObjectMapping.from_properties_typed<TState>(typeof(TState), slot.current_state.public_data);
+            } catch { current = null; }
+        }
+
+        var result = derive((TPayload) payload, current, held);
+
+        if (result == null) {
+            yield state_service.clear_slot(slot_key);
+            return ModifierOutcome.CLEARED;
+        }
+        if (result == current) {
+            return ModifierOutcome.UNCHANGED;
+        }
+
+        var new_state = new State() {
+            type_name = state_type_name,
+            public_data = GObjectMapping.to_properties((Object) result),
+            private_data = slot?.current_state?.private_data ?? new PropertyDictionary()
+        };
+        yield state_service.update(slot_key, new_state);
+        return ModifierOutcome.UPDATED;
+    }
+}
+```
+
+### V-4: `TopicRegistry` (singleton) — `src/TopicSystem.vala`
+
+```vala
+public class TopicRegistry : Object {
+    private StateService state_service = inject<StateService>();
+
+    // topic_key → watchers
+    private Dictionary<string, Series<Watcher>> topic_watchers;
+    // topic prefix → Topic handler (stored as a factory, resolved via DI)
+    private Dictionary<string, Type> topic_types;
+    // topic prefix → modifiers (by state_type_name)
+    private Dictionary<string, Dictionary<string, TopicStateModifier>> prefix_modifiers;
+
+    // Watcher record
+    private class Watcher : Object {
+        public string slot_key;
+        public string slot_type;
+        public Dictionary<string, string> held_keys; // type_name → slot_key
+    }
+
+    public void watch(string topic_key, string slot_key, string slot_type,
+                      ReadOnlyAssociative<string, HeldSlot> request_held) { ... }
+
+    public async void trigger(string topic_key) throws GLib.Error {
+        // Early exit: if nobody is watching, skip the load entirely (no DB query).
+        var watchers = topic_watchers.get_or_default(topic_key);
+        if (watchers == null || ((!)watchers).to_array().length == 0) return;
+
+        var prefix = extract_prefix(topic_key); // "cat:42" → "cat"
+        var topic = construct_topic(prefix);
+        var payload = yield topic.load(topic_key);
+
+        var to_remove = new Series<Watcher>();
+        foreach (var watcher in (!)watchers) {
+            var modifiers = prefix_modifiers.get_or_default(prefix);
+            if (modifiers == null) continue;
+
+            TopicStateModifier modifier;
+            if (!((!)modifiers).try_get(watcher.slot_type, out modifier)) continue;
+
+            var held = new HeldStates(watcher.held_keys, state_service);
+
+            try {
+                var outcome = yield modifier.apply(payload, watcher.slot_key, held);
+                if (outcome == ModifierOutcome.CLEARED) {
+                    to_remove.add(watcher);
+                }
+            } catch (GLib.Error e) {
+                // Exception in derive → clear + remove (same as null return)
+                try { yield state_service.clear_slot(watcher.slot_key); }
+                catch {}
+                to_remove.add(watcher);
+            }
+        }
+
+        foreach (var w in to_remove) {
+            ((!)watchers).remove(w);
+        }
+    }
+
+    // Periodic cleanup: remove watchers whose slots are no longer cached.
+    public void sweep() { ... }
+}
+```
+
+### V-5: `StatumConfigurator` — register topics + modifiers
+
+```vala
+public void topic<TTopic>(string prefix) throws Error {
+    container.register_transient<TTopic>();
+    // Store typeof(TTopic) for the prefix
+}
+
+public void topic_modifier<TModifier, TPayload, TState>(string prefix) throws Error {
+    container.register_transient<TModifier>();
+    // Store typeof(TModifier) for the prefix + typeof(TState).name()
+}
+```
+
+The registry resolves modifiers and topics from the container (DI-constructed).
+
+### V-6: `StatumModule` — register `TopicRegistry` singleton + sweep timer
+
+```vala
+container.register_singleton<TopicRegistry>();
+// Sweep timer runs every 5 minutes (background).
+```
+
+### V-7: Entrypoint/action access
+
+Both `StatumEntrypoint` and `StatumAction` gain:
+```vala
+protected TopicRegistry topics = inject<TopicRegistry>();
+```
+
+### V-8: Example
+
+Add a `CatTopic` + `CatBreedDetailModifier` to the example (or a simpler variant
+using the existing announcement/counter models to demonstrate the flow without a
+database).
+
+### D-1: Documentation
+
+- **`model.md`**: add a "Topic system" section.
+- **`getting-started.md`**: add `Topic<T>`, `StateModifier<T,S>`, `TopicRegistry`
+  to the Vala API reference.
+
+---
+
+## API sketch (end-to-end)
+
+```vala
+// Topic: loads the entity
+public class CatTopic : Topic<Cat> {
+    private Database db = inject<Database>();
+    public override async Cat load(string topic_key) throws GLib.Error {
+        return yield db.load_cat(int.parse(topic_key.split(":")[1]));
+    }
+}
+
+// Modifier: derives CatBreedDetail from Cat + held context
+public class CatBreedDetailModifier : StateModifier<Cat, CatBreedDetail> {
+    public override CatBreedDetail? derive(Cat cat, CatBreedDetail? current, HeldStates held) {
+        var auth = held.get<AuthPublic>("auth");
+        if (auth == null) return null;                       // expired → clear
+        if (current != null && current.breed != cat.breed)
+            return current;                                  // different breed → skip
+        return new CatBreedDetail() { ... };                 // push
+    }
+}
+
+// Registration
+statum.topic<CatTopic>("cat");
+statum.topic_modifier<CatBreedDetailModifier, Cat, CatBreedDetail>("cat");
+
+// Entrypoint
+topics.watch(@"cat:$(id)", slot.id, typeof(CatBreedDetail).name(), request.held);
+
+// Trigger (anywhere)
+yield topics.trigger("cat:42");
+```
+
+## Server resources
+
+- Each watcher: ~150 bytes (topic key + slot key + type name + held keys map).
+- 10K concurrent watchers ≈ 1.5 MB.
+- Trigger cost: one `load()` (DB) + per-watcher: one `get_slot` + one `derive` (in-memory) + optionally one `update` (sign + push).
+- Sweep: every 5 minutes, iterates watchers, checks `get_slot`. O(N) where N = watcher count.
+
+## Out of scope
+
+- Per-watcher personalised payloads (different state for different watchers of the same topic). Use per-user topic keys instead.
+- Cross-process pub/sub (Redis/NATS backend for multi-server). Each process has its own TopicRegistry.
+- Pattern-based topic matching (wildcards). Prefix matching ("cat" matches "cat:42") is sufficient.

src/Statum.vala

@@ -34,6 +34,7 @@ namespace Statum {
             container.register_singleton<HeldSlotResolver>();
             container.register_singleton<EntrypointRouteTable>();
             container.register_singleton<ActionRegistry>();
+            container.register_singleton<TopicRegistry>();
 
             // The realtime channel endpoint is the singleton ChannelService.
             container.register_singleton<ChannelEndpoint>()
@@ -134,7 +135,8 @@ namespace Statum {
         private Container container = inject<Container>();
         private EntrypointRouteTable route_table = inject<EntrypointRouteTable>();
         private ActionRegistry action_registry = inject<ActionRegistry>();
-        /** Exposed for background tasks (timers, webhooks) that need to update global slots. */
+        /** Exposed for background tasks (timers, webhooks) that need to trigger topics. */
+        public TopicRegistry topic_registry = inject<TopicRegistry>();
         public StateService state_service = inject<StateService>();
 
         /**
@@ -206,6 +208,16 @@ namespace Statum {
             action_registry.register<TAction>();
         }
 
+        /** Registers a typed topic handler for a key prefix (e.g. "cat" → "cat:42"). */
+        public void topic<TTopic>(string prefix) {
+            topic_registry.register_topic<TTopic>(prefix);
+        }
+
+        /** Registers a state modifier that derives a slot type from a topic payload. */
+        public void topic_modifier<TModifier, TState>(string prefix) {
+            topic_registry.register_modifier<TModifier, TState>(prefix);
+        }
+
     }
 
 }

src/StatumHandlers.vala

@@ -33,6 +33,9 @@ namespace Statum {
         /** The singleton action registry (for authoring action references). */
         protected ActionRegistry action_registry = inject<ActionRegistry>();
 
+        /** The singleton topic registry (for watching topics). */
+        protected TopicRegistry topic_registry = inject<TopicRegistry>();
+
         /**
          * Returns the directive builder that hydrates the page bound to this
          * entrypoint. The framework calls {@link DirectiveBuilder.build} on it.
@@ -67,6 +70,7 @@ namespace Statum {
         protected StateService state_service = inject<StateService>();
         protected ChannelService channel_service = inject<ChannelService>();
         protected ActionRegistry action_registry = inject<ActionRegistry>();
+        protected TopicRegistry topic_registry = inject<TopicRegistry>();
         protected Inversion.Scope scope = inject<Inversion.Scope>();
 
         /** The per-request Statum context (populated before {@link handle}). */

src/TopicSystem.vala

@@ -0,0 +1,280 @@
+using Invercargill;
+using Invercargill.DataStructures;
+using Inversion;
+using Astralis;
+
+namespace Statum {
+
+    public enum ModifierOutcome {
+        UPDATED,
+        UNCHANGED,
+        CLEARED,
+    }
+
+    // ------------------------------------------------------------------
+    // HeldStates
+    // ------------------------------------------------------------------
+
+    public class HeldStates : Object {
+
+        private Dictionary<string, string> slot_keys;
+        private StateService state_service;
+
+        internal HeldStates(Dictionary<string, string> slot_keys, StateService state_service) {
+            this.slot_keys = slot_keys;
+            this.state_service = state_service;
+        }
+
+        public T? get<T>(string type_name) {
+            string slot_key;
+            if (!slot_keys.try_get(type_name, out slot_key)) {
+                return null;
+            }
+            var slot = state_service.get_slot(slot_key);
+            if (slot == null || ((!)slot).current_state == null) {
+                return null;
+            }
+            try {
+                return GObjectMapping.from_properties_typed<T>(typeof(T), ((!)slot).current_state.public_data);
+            } catch (GLib.Error e) {
+                return null;
+            }
+        }
+
+        public Properties? get_properties(string type_name) {
+            string slot_key;
+            if (!slot_keys.try_get(type_name, out slot_key)) {
+                return null;
+            }
+            var slot = state_service.get_slot(slot_key);
+            if (slot == null || ((!)slot).current_state == null) {
+                return null;
+            }
+            return ((!)slot).current_state.public_data;
+        }
+    }
+
+    // ------------------------------------------------------------------
+    // TopicLoader (internal non-generic bridge)
+    // ------------------------------------------------------------------
+
+    public interface TopicLoader : Object {
+        public abstract async GLib.Object? load_payload(string topic_key) throws GLib.Error;
+    }
+
+    // ------------------------------------------------------------------
+    // Topic<TPayload>
+    // ------------------------------------------------------------------
+
+    public abstract class Topic<TPayload> : Object, TopicLoader {
+
+        public abstract async TPayload load(string topic_key) throws GLib.Error;
+
+        public async GLib.Object? load_payload(string topic_key) throws GLib.Error {
+            return (GLib.Object) yield load(topic_key);
+        }
+    }
+
+    // ------------------------------------------------------------------
+    // TopicStateModifier (internal non-generic bridge)
+    // ------------------------------------------------------------------
+
+    public interface TopicStateModifier : Object {
+        public abstract Type state_type { get; }
+        public abstract async ModifierOutcome apply(GLib.Object payload, string slot_key, HeldStates held) throws GLib.Error;
+    }
+
+    // ------------------------------------------------------------------
+    // StateModifier<TPayload, TState>
+    // ------------------------------------------------------------------
+
+    public abstract class StateModifier : Object, TopicStateModifier {
+
+        protected StateService state_service = inject<StateService>();
+
+        public abstract Type state_type { get; }
+
+        public string state_type_name {
+            get { return state_type.name(); }
+        }
+
+        /**
+         * Derive new state from the payload + context.
+         *
+         * @param payload  The trigger payload (cast to your expected type).
+         * @param current  The current state of the slot, or null if expired.
+         * @param held     Live lookup of the client's held states.
+         * @return New Object → push; same ref as {@code current} → skip; null → clear.
+         */
+        public abstract GLib.Object? derive(GLib.Object payload, GLib.Object? current, HeldStates held);
+
+        public async ModifierOutcome apply(GLib.Object payload, string slot_key, HeldStates held) throws GLib.Error {
+            var slot = state_service.get_slot(slot_key);
+            GLib.Object? current = null;
+            if (slot != null && ((!)slot).current_state != null) {
+                try {
+                    current = GObjectMapping.from_properties(state_type, ((!)slot).current_state.public_data);
+                } catch (GLib.Error e) {
+                    current = null;
+                }
+            }
+
+            var result = derive(payload, current, held);
+
+            if (result == null) {
+                yield state_service.clear_slot(slot_key);
+                return ModifierOutcome.CLEARED;
+            }
+            if ((void*) result == (void*) current) {
+                return ModifierOutcome.UNCHANGED;
+            }
+
+            var new_state = new State() {
+                type_name = (slot != null && ((!)slot).current_state != null)
+                    ? ((!)slot).current_state.type_name
+                    : state_type_name,
+                public_data = GObjectMapping.to_properties((!)result),
+                private_data = (slot != null && ((!)slot).current_state != null)
+                    ? ((!)slot).current_state.private_data ?? new PropertyDictionary()
+                    : new PropertyDictionary()
+            };
+            yield state_service.update(slot_key, new_state);
+            return ModifierOutcome.UPDATED;
+        }
+    }
+
+    // ------------------------------------------------------------------
+    // TopicRegistry
+    // ------------------------------------------------------------------
+
+    public class TopicRegistry : Object {
+
+        private StateService state_service = inject<StateService>();
+        private Container container = inject<Container>();
+
+        private Dictionary<string, Series<Watcher>> topic_watchers = new Dictionary<string, Series<Watcher>>();
+        private Dictionary<string, Type> topic_types = new Dictionary<string, Type>();
+        private Dictionary<string, Dictionary<string, Type>> prefix_modifier_types = new Dictionary<string, Dictionary<string, Type>>();
+
+        public void register_topic<TTopic>(string prefix) {
+            container.register_transient<TTopic>();
+            topic_types[prefix] = typeof(TTopic);
+        }
+
+        public void register_modifier<TModifier, TState>(string prefix) {
+            container.register_transient<TModifier>();
+            var stn = typeof(TState).name();
+
+            Dictionary<string, Type> modifiers;
+            if (!prefix_modifier_types.try_get(prefix, out modifiers)) {
+                modifiers = new Dictionary<string, Type>();
+                prefix_modifier_types[prefix] = modifiers;
+            }
+            modifiers[stn] = typeof(TModifier);
+        }
+
+        public void watch(string topic_key, string slot_key, string slot_type,
+                          ReadOnlyAssociative<string, HeldSlot> request_held) {
+            var held_keys = new Dictionary<string, string>();
+            foreach (var kvp in request_held) {
+                held_keys[kvp.key] = kvp.value.key;
+            }
+
+            var watcher = new Watcher() {
+                slot_key = slot_key,
+                slot_type = slot_type,
+                held_keys = held_keys
+            };
+
+            Series<Watcher> watchers;
+            if (!topic_watchers.try_get(topic_key, out watchers)) {
+                watchers = new Series<Watcher>();
+                topic_watchers[topic_key] = watchers;
+            }
+            watchers.add(watcher);
+        }
+
+        public async void trigger(string topic_key) throws GLib.Error {
+            // Early exit: no watchers → skip the load entirely.
+            Series<Watcher> watchers;
+            if (!topic_watchers.try_get(topic_key, out watchers) || watchers == null) {
+                return;
+            }
+            if (((!)watchers).to_array().length == 0) {
+                return;
+            }
+
+            var prefix = extract_prefix(topic_key);
+
+            Type topic_type;
+            if (!topic_types.try_get(prefix, out topic_type)) {
+                return;
+            }
+
+            var scope = container.create_transient_scope();
+            var topic_obj = scope.resolve_type(topic_type);
+            if (!(topic_obj is TopicLoader)) {
+                return;
+            }
+            var payload = yield ((TopicLoader)(!)topic_obj).load_payload(topic_key);
+
+            Dictionary<string, Type> modifier_types;
+            if (!prefix_modifier_types.try_get(prefix, out modifier_types) || modifier_types == null) {
+                return;
+            }
+
+            var to_remove = new Series<Watcher>();
+            foreach (var watcher in (!)watchers) {
+                Type modifier_type;
+                if (!((!)modifier_types).try_get(watcher.slot_type, out modifier_type)) {
+                    continue;
+                }
+
+                var held = new HeldStates(watcher.held_keys, state_service);
+
+                try {
+                    var modifier = (TopicStateModifier) scope.resolve_type(modifier_type);
+                    var outcome = yield modifier.apply(payload, watcher.slot_key, held);
+                    if (outcome == ModifierOutcome.CLEARED) {
+                        to_remove.add(watcher);
+                    }
+                } catch (GLib.Error e) {
+                    try {
+                        yield state_service.clear_slot(watcher.slot_key);
+                    } catch {}
+                    to_remove.add(watcher);
+                }
+            }
+
+            foreach (var w in to_remove) {
+                ((!)watchers).remove(w);
+            }
+        }
+
+        public void sweep() {
+            foreach (var entry in topic_watchers) {
+                var watchers = entry.value;
+                var to_remove = new Series<Watcher>();
+                foreach (var watcher in watchers) {
+                    if (state_service.get_slot(watcher.slot_key) == null) {
+                        to_remove.add(watcher);
+                    }
+                }
+                foreach (var w in to_remove) {
+                    watchers.remove(w);
+                }
+            }
+        }
+
+        private static string extract_prefix(string topic_key) {
+            var colon = topic_key.index_of(":");
+            return colon >= 0 ? topic_key.substring(0, colon) : topic_key;
+        }
+
+        private class Watcher : Object {
+            public string slot_key;
+            public string slot_type;
+            public Dictionary<string, string> held_keys;
+        }
+    }
+}

src/meson.build

@@ -46,6 +46,7 @@ sources = files(
     'ActionEndpoint.vala',
     'DirectiveBuilder.vala',
     'GObjectMapping.vala',
+    'TopicSystem.vala',
     'Model/Scope.vala',
     'Model/SlotDto.vala',
     'Model/FrameDto.vala',