Per-model OpSpec registry with change notifications.
- Stores specs keyed by (alias, target).
- Adds/sets/removes specs and notifies listeners with the changed keys.
- Binder should call subscribe(...)to rebuild a model's namespaces when the
  registry changes (partial rebuild is possible based on changed keys).
Thread-safe via an instance-level RLock.
                  
                    Source code in tigrbl/op/model_registry.py
                    | 91
92
93
94
95
96
97
98
99 | def __init__(self, table: type) -> None:
    self._table: type = table
    self._items: Dict[Tuple[str, TargetOp], OpSpec] = {}
    self._lock = RLock()
    # store weakrefs to listener callables where possible; fallback to strong refs
    self._listeners: List[
        Callable[[OpspecRegistry, set[Tuple[str, TargetOp]]], None]
    ] = []
    self._version: int = 0
 | 
 
  
    
            
              Source code in tigrbl/op/model_registry.py
              |  | def keys(self) -> Iterator[Tuple[str, TargetOp]]:
    with self._lock:
        return iter(tuple(self._items.keys()))
 | 
 
     
 
    
            
              Source code in tigrbl/op/model_registry.py
              |  | def items(self) -> Iterator[Tuple[Tuple[str, TargetOp], OpSpec]]:
    with self._lock:
        return iter(tuple(self._items.items()))
 | 
 
     
 
    
            
              Source code in tigrbl/op/model_registry.py
              |  | def values(self) -> Iterator[OpSpec]:
    with self._lock:
        return iter(tuple(self._items.values()))
 | 
 
     
 
    
        Stable snapshot of all specs.
            
              Source code in tigrbl/op/model_registry.py
              |  | def get_all(self) -> Tuple[OpSpec, ...]:
    """Stable snapshot of all specs."""
    with self._lock:
        return tuple(self._items.values())
 | 
 
     
 
    
        Register a listener to be called on changes.
NOTE: The listener should be idempotent. It receives (registry, changed_keys).
            
              Source code in tigrbl/op/model_registry.py
              | 130
131
132
133
134
135
136
137
138 | def subscribe(self, fn: Listener) -> None:
    """
    Register a listener to be called on changes.
    NOTE: The listener should be idempotent. It receives (registry, changed_keys).
    """
    with self._lock:
        # Avoid duplicate subscriptions
        if fn not in self._listeners:
            self._listeners.append(fn)
 | 
 
     
 
    
            
              Source code in tigrbl/op/model_registry.py
              |  | def unsubscribe(self, fn: Listener) -> None:
    with self._lock:
        try:
            self._listeners.remove(fn)
        except ValueError:
            pass
 | 
 
     
 
    
        Add or overwrite one or more specs.
Returns the set of changed keys.
            
              Source code in tigrbl/op/model_registry.py
              | 161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182 | def add(self, specs: Iterable[OpSpec] | OpSpec) -> set[Tuple[str, TargetOp]]:
    """
    Add or overwrite one or more specs.
    Returns the set of changed keys.
    """
    if isinstance(specs, OpSpec):
        specs = (specs,)
    changed: set[Tuple[str, TargetOp]] = set()
    with self._lock:
        for sp in specs:
            sp = _ensure_table(sp, self._table)
            k = _spec_key(sp)
            if self._items.get(k) is sp:
                continue  # exact object already present
            self._items[k] = sp
            changed.add(k)
        if changed:
            self._version += 1
    if changed:
        self._notify(changed)
    return changed
 | 
 
     
 
    
        Replace all specs with the provided iterable.
Returns the set of changed keys (union of removed + added/updated).
            
              Source code in tigrbl/op/model_registry.py
              | 184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210 | def set(self, specs: Iterable[OpSpec]) -> set[Tuple[str, TargetOp]]:
    """
    Replace all specs with the provided iterable.
    Returns the set of changed keys (union of removed + added/updated).
    """
    new_map: Dict[Tuple[str, TargetOp], OpSpec] = {}
    for sp in specs:
        sp = _ensure_table(sp, self._table)
        new_map[_spec_key(sp)] = sp
    with self._lock:
        old_keys = set(self._items.keys())
        new_keys = set(new_map.keys())
        removed = old_keys - new_keys
        added_or_updated = {
            k for k in new_keys if self._items.get(k) is not new_map[k]
        }
        changed = removed | added_or_updated
        self._items = new_map
        if changed:
            self._version += 1
    if changed:
        self._notify(changed)
    return changed
 | 
 
     
 
remove(alias, target=None)
        Remove specs by alias (optionally constrain to a specific target).
Returns the set of removed keys.
            
              Source code in tigrbl/op/model_registry.py
              | 212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238 | def remove(
    self, alias: str, target: TargetOp | None = None
) -> set[Tuple[str, TargetOp]]:
    """
    Remove specs by alias (optionally constrain to a specific target).
    Returns the set of removed keys.
    """
    removed: set[Tuple[str, TargetOp]] = set()
    with self._lock:
        if target is None:
            # remove all targets under this alias
            for k in list(self._items.keys()):
                if k[0] == alias:
                    self._items.pop(k, None)
                    removed.add(k)
        else:
            k = (alias, target)
            if k in self._items:
                self._items.pop(k, None)
                removed.add(k)
        if removed:
            self._version += 1
    if removed:
        self._notify(removed)
    return removed
 | 
 
     
 
    
            
              Source code in tigrbl/op/model_registry.py
              | 240
241
242
243
244
245
246 | def clear(self) -> None:
    with self._lock:
        if not self._items:
            return
        self._items.clear()
        self._version += 1
    self._notify(set())
 |