Skip to content

manager

CartridgeManager

Singleton that owns the dynamic load/unload lifecycle of cartridges.

Cartridges live in :mod:wintermute.cartridges. Each cartridge module exposes one primary class — either named after the module itself (tpm20 → tpm20) or carrying the Cartridge suffix (firmware_analysis → FirmwareAnalysisCartridge). The manager discovers, instantiates, and registers each cartridge's public methods as AI tools so the LLM can call them by name.

Singleton: a single instance is shared across the console, tests, and any other caller that constructs CartridgeManager(). Use :meth:reset_for_tests to clear state between unit tests.

Source code in wintermute/cartridges/manager.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
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
211
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
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
class CartridgeManager:
    """Singleton that owns the dynamic load/unload lifecycle of cartridges.

    Cartridges live in :mod:`wintermute.cartridges`. Each cartridge module
    exposes one primary class — either named after the module itself
    (``tpm20`` → ``tpm20``) or carrying the ``Cartridge`` suffix
    (``firmware_analysis`` → ``FirmwareAnalysisCartridge``). The manager
    discovers, instantiates, and registers each cartridge's public methods
    as AI tools so the LLM can call them by name.

    Singleton: a single instance is shared across the console, tests, and
    any other caller that constructs ``CartridgeManager()``. Use
    :meth:`reset_for_tests` to clear state between unit tests.
    """

    _instance: ClassVar[Optional["CartridgeManager"]] = None
    _initialized: bool

    def __new__(cls) -> "CartridgeManager":
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self) -> None:
        # Idempotent because ``__new__`` returns the same instance every time.
        if getattr(self, "_initialized", False):
            return
        self._initialized = True
        self.loaded_cartridges: Dict[str, Any] = {}
        # cartridge name -> tool names registered on its behalf
        self._tool_names: Dict[str, List[str]] = {}
        self.cartridges_path: Path = Path(__file__).resolve().parent
        # Observer pattern: callbacks fired whenever a successful load() or
        # unload() mutates the global tool registry. The MCP server uses
        # this hook to dynamically refresh its tool surface and emit a
        # `notifications/tools/list_changed` to connected clients.
        self._callbacks: List[Callable[[], None]] = []

    # ------------------------------------------------------------------
    # Test helper — explicit reset; do not use from production code.
    # ------------------------------------------------------------------

    @classmethod
    def reset_for_tests(cls) -> None:
        """Drop all loaded cartridges, unregister their tools, and clear
        any registered observer callbacks.

        Designed for test isolation. Safe to call when no instance exists.
        """
        if cls._instance is None:
            return
        instance = cls._instance
        for name in list(instance.loaded_cartridges.keys()):
            try:
                instance.unload(name)
            except Exception:
                log.exception("reset_for_tests: failed to unload %r", name)
        instance.loaded_cartridges.clear()
        instance._tool_names.clear()
        instance._callbacks.clear()

    # ------------------------------------------------------------------
    # Observer pattern — callbacks fire on every load/unload that mutates
    # the global tool registry. Observers are invoked synchronously in
    # registration order; exceptions in one callback do not prevent the
    # others from running.
    # ------------------------------------------------------------------

    def register_callback(self, fn: Callable[[], None]) -> None:
        """Subscribe ``fn`` to load/unload notifications.

        Callbacks are sync. If a subscriber needs to bridge into an
        asyncio event loop (e.g. the MCP server posting a
        ``notifications/tools/list_changed``) it is responsible for
        capturing its own loop and dispatching via
        :func:`asyncio.run_coroutine_threadsafe` or equivalent.
        """
        if fn not in self._callbacks:
            self._callbacks.append(fn)

    def unregister_callback(self, fn: Callable[[], None]) -> bool:
        """Drop ``fn`` from the observer list. Returns ``True`` if removed."""
        try:
            self._callbacks.remove(fn)
        except ValueError:
            return False
        return True

    def _fire_callbacks(self) -> None:
        for cb in list(self._callbacks):
            try:
                cb()
            except Exception:
                log.exception(
                    "CartridgeManager observer callback %r raised — continuing",
                    cb,
                )

    # ------------------------------------------------------------------
    # Discovery
    # ------------------------------------------------------------------

    def list_available(self) -> List[str]:
        """Scan ``wintermute/cartridges/`` for cartridge module names.

        Excludes the package's plumbing files (``__init__.py``,
        ``manager.py``, ``__main__.py``) so the result only contains
        loadable cartridge module stems.
        """
        if not self.cartridges_path.is_dir():
            return []
        names: List[str] = []
        for entry in sorted(self.cartridges_path.iterdir()):
            if entry.suffix != ".py":
                continue
            stem = entry.stem
            if stem in _NON_CARTRIDGE_MODULES:
                continue
            names.append(stem)
        return names

    def list_loaded(self) -> List[str]:
        """Return names of currently loaded cartridges (load order)."""
        return list(self.loaded_cartridges.keys())

    # ------------------------------------------------------------------
    # Load / unload
    # ------------------------------------------------------------------

    def load(self, name: str) -> bool:
        """Import the cartridge module, instantiate its primary class, and
        register every public method as an AI tool.

        Args:
            name: Cartridge module stem, e.g. ``"tpm20"`` or
                ``"firmware_analysis"``. Must be present in
                :meth:`list_available`.

        Returns:
            ``True`` if the cartridge transitioned from "not loaded" to
            "loaded" and at least one method was registered. Returns
            ``True`` even when zero methods were registered, as long as
            the instance itself was successfully created.

        Raises:
            ModuleNotFoundError: If the module cannot be imported.
            RuntimeError: If no primary class can be located, or if the
                instance constructor raises (e.g., ``JTAGCartridge`` when
                ``openocd`` is missing). The original exception is wrapped.
        """
        if name in self.loaded_cartridges:
            log.info("Cartridge %r is already loaded.", name)
            return False

        module = importlib.import_module(f"wintermute.cartridges.{name}")
        cls = self._find_primary_class(module, name)
        if cls is None:
            raise RuntimeError(
                f"Cartridge {name!r}: no primary class found "
                f"(expected a class matching the module name or ending in "
                f"{_CARTRIDGE_CLASS_SUFFIX!r})."
            )
        try:
            instance = cls()
        except Exception as exc:
            raise RuntimeError(
                f"Cartridge {name!r}: failed to instantiate {cls.__name__}: {exc}"
            ) from exc

        registered_names = self._register_instance_methods(name, instance)
        self.loaded_cartridges[name] = instance
        self._tool_names[name] = registered_names
        log.info(
            "Loaded cartridge %r (%s) — %d tool(s) registered.",
            name,
            cls.__name__,
            len(registered_names),
        )
        # Observer broadcast: fire AFTER the manager state is fully
        # consistent so callbacks observe the post-load registry.
        self._fire_callbacks()
        return True

    def unload(self, name: str) -> bool:
        """Remove the cartridge instance and unregister every tool that
        was registered on its behalf.

        Returns ``True`` if the cartridge was loaded prior to the call.
        """
        if name not in self.loaded_cartridges:
            return False
        tool_names = self._tool_names.pop(name, [])
        unregister_tools(tool_names)
        del self.loaded_cartridges[name]
        log.info(
            "Unloaded cartridge %r — %d tool(s) unregistered.",
            name,
            len(tool_names),
        )
        # Observer broadcast: fire AFTER the manager state is fully
        # consistent so callbacks observe the post-unload registry.
        self._fire_callbacks()
        return True

    def get(self, name: str) -> Any:
        """Return the live cartridge instance, or raise ``KeyError``."""
        if name not in self.loaded_cartridges:
            raise KeyError(f"Cartridge {name!r} is not loaded.")
        return self.loaded_cartridges[name]

    def tool_names_for(self, name: str) -> List[str]:
        """Return the AI tool names that belong to the given cartridge."""
        return list(self._tool_names.get(name, []))

    # ------------------------------------------------------------------
    # Internal helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _find_primary_class(module: Any, module_stem: str) -> Optional[type]:
        """Locate the primary class of a cartridge module.

        Resolution order:
            1. Class whose name equals the module stem (case-insensitive).
            2. First class whose name ends in ``Cartridge`` (case-insens.).
            3. First class actually defined in the module (i.e. not imported).

        ``None`` is returned if nothing matches — the caller decides how
        to surface the failure.
        """
        candidates: List[type] = []
        target_lower = module_stem.lower()
        suffix_match: Optional[type] = None
        own_module: Optional[type] = None

        for member_name, member in inspect.getmembers(module, inspect.isclass):
            if getattr(member, "__module__", "") != module.__name__:
                # Imported into the module, not defined in it.
                continue
            candidates.append(member)
            lower_name = member_name.lower()
            if lower_name == target_lower:
                return member
            if suffix_match is None and lower_name.endswith(_CARTRIDGE_CLASS_SUFFIX):
                suffix_match = member
            if own_module is None:
                own_module = member

        if suffix_match is not None:
            return suffix_match
        return own_module

    def _register_instance_methods(
        self, cartridge_name: str, instance: Any
    ) -> List[str]:
        """Walk the instance's public methods and feed them to the
        :func:`register_tools` adapter.

        Methods that fail conversion (unusual signatures, unsupported
        annotations) are skipped with a warning so a single bad method
        cannot block the rest of the cartridge from loading.
        """
        callables: List[Callable[..., Any]] = []
        for attr_name in dir(instance):
            if attr_name.startswith("_"):
                continue
            try:
                member = getattr(instance, attr_name)
            except Exception:
                continue
            if not inspect.ismethod(member):
                continue
            callables.append(member)

        registered: List[str] = []
        for fn in callables:
            try:
                [tool] = register_tools([fn])
            except Exception as exc:
                log.warning(
                    "Cartridge %r: skipping %s — %s",
                    cartridge_name,
                    fn.__name__,
                    exc,
                )
                continue
            try:
                global_tool_registry.register(tool)
            except Exception as exc:
                log.warning(
                    "Cartridge %r: failed to register %s: %s",
                    cartridge_name,
                    tool.name,
                    exc,
                )
                continue
            registered.append(tool.name)
        return registered

get(name)

Return the live cartridge instance, or raise KeyError.

Source code in wintermute/cartridges/manager.py 
251
252
253
254
255
def get(self, name: str) -> Any:
    """Return the live cartridge instance, or raise ``KeyError``."""
    if name not in self.loaded_cartridges:
        raise KeyError(f"Cartridge {name!r} is not loaded.")
    return self.loaded_cartridges[name]

list_available()

Scan wintermute/cartridges/ for cartridge module names.

Excludes the package's plumbing files (__init__.py, manager.py, __main__.py) so the result only contains loadable cartridge module stems.

Source code in wintermute/cartridges/manager.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
def list_available(self) -> List[str]:
    """Scan ``wintermute/cartridges/`` for cartridge module names.

    Excludes the package's plumbing files (``__init__.py``,
    ``manager.py``, ``__main__.py``) so the result only contains
    loadable cartridge module stems.
    """
    if not self.cartridges_path.is_dir():
        return []
    names: List[str] = []
    for entry in sorted(self.cartridges_path.iterdir()):
        if entry.suffix != ".py":
            continue
        stem = entry.stem
        if stem in _NON_CARTRIDGE_MODULES:
            continue
        names.append(stem)
    return names

list_loaded()

Return names of currently loaded cartridges (load order).

Source code in wintermute/cartridges/manager.py 
168
169
170
def list_loaded(self) -> List[str]:
    """Return names of currently loaded cartridges (load order)."""
    return list(self.loaded_cartridges.keys())

load(name)

Import the cartridge module, instantiate its primary class, and register every public method as an AI tool.

Parameters:

Name Type Description Default
name str

Cartridge module stem, e.g. "tpm20" or "firmware_analysis". Must be present in :meth:list_available.

 required

Returns:

Type Description
bool

True if the cartridge transitioned from "not loaded" to
bool

"loaded" and at least one method was registered. Returns
bool

True even when zero methods were registered, as long as
bool

the instance itself was successfully created.

Raises:

Type Description
ModuleNotFoundError

If the module cannot be imported.
RuntimeError

If no primary class can be located, or if the instance constructor raises (e.g., JTAGCartridge when openocd is missing). The original exception is wrapped.
Source code in wintermute/cartridges/manager.py 
176
177
178
179
180
181
182
183
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
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
def load(self, name: str) -> bool:
    """Import the cartridge module, instantiate its primary class, and
    register every public method as an AI tool.

    Args:
        name: Cartridge module stem, e.g. ``"tpm20"`` or
            ``"firmware_analysis"``. Must be present in
            :meth:`list_available`.

    Returns:
        ``True`` if the cartridge transitioned from "not loaded" to
        "loaded" and at least one method was registered. Returns
        ``True`` even when zero methods were registered, as long as
        the instance itself was successfully created.

    Raises:
        ModuleNotFoundError: If the module cannot be imported.
        RuntimeError: If no primary class can be located, or if the
            instance constructor raises (e.g., ``JTAGCartridge`` when
            ``openocd`` is missing). The original exception is wrapped.
    """
    if name in self.loaded_cartridges:
        log.info("Cartridge %r is already loaded.", name)
        return False

    module = importlib.import_module(f"wintermute.cartridges.{name}")
    cls = self._find_primary_class(module, name)
    if cls is None:
        raise RuntimeError(
            f"Cartridge {name!r}: no primary class found "
            f"(expected a class matching the module name or ending in "
            f"{_CARTRIDGE_CLASS_SUFFIX!r})."
        )
    try:
        instance = cls()
    except Exception as exc:
        raise RuntimeError(
            f"Cartridge {name!r}: failed to instantiate {cls.__name__}: {exc}"
        ) from exc

    registered_names = self._register_instance_methods(name, instance)
    self.loaded_cartridges[name] = instance
    self._tool_names[name] = registered_names
    log.info(
        "Loaded cartridge %r (%s) — %d tool(s) registered.",
        name,
        cls.__name__,
        len(registered_names),
    )
    # Observer broadcast: fire AFTER the manager state is fully
    # consistent so callbacks observe the post-load registry.
    self._fire_callbacks()
    return True

register_callback(fn)

Subscribe fn to load/unload notifications.

Callbacks are sync. If a subscriber needs to bridge into an asyncio event loop (e.g. the MCP server posting a notifications/tools/list_changed) it is responsible for capturing its own loop and dispatching via :func:asyncio.run_coroutine_threadsafe or equivalent.

Source code in wintermute/cartridges/manager.py 
115
116
117
118
119
120
121
122
123
124
125
def register_callback(self, fn: Callable[[], None]) -> None:
    """Subscribe ``fn`` to load/unload notifications.

    Callbacks are sync. If a subscriber needs to bridge into an
    asyncio event loop (e.g. the MCP server posting a
    ``notifications/tools/list_changed``) it is responsible for
    capturing its own loop and dispatching via
    :func:`asyncio.run_coroutine_threadsafe` or equivalent.
    """
    if fn not in self._callbacks:
        self._callbacks.append(fn)

reset_for_tests() classmethod

Drop all loaded cartridges, unregister their tools, and clear any registered observer callbacks.

Designed for test isolation. Safe to call when no instance exists.

Source code in wintermute/cartridges/manager.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
@classmethod
def reset_for_tests(cls) -> None:
    """Drop all loaded cartridges, unregister their tools, and clear
    any registered observer callbacks.

    Designed for test isolation. Safe to call when no instance exists.
    """
    if cls._instance is None:
        return
    instance = cls._instance
    for name in list(instance.loaded_cartridges.keys()):
        try:
            instance.unload(name)
        except Exception:
            log.exception("reset_for_tests: failed to unload %r", name)
    instance.loaded_cartridges.clear()
    instance._tool_names.clear()
    instance._callbacks.clear()

tool_names_for(name)

Return the AI tool names that belong to the given cartridge.

Source code in wintermute/cartridges/manager.py 
257
258
259
def tool_names_for(self, name: str) -> List[str]:
    """Return the AI tool names that belong to the given cartridge."""
    return list(self._tool_names.get(name, []))

unload(name)

Remove the cartridge instance and unregister every tool that was registered on its behalf.

Returns True if the cartridge was loaded prior to the call.

Source code in wintermute/cartridges/manager.py 
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def unload(self, name: str) -> bool:
    """Remove the cartridge instance and unregister every tool that
    was registered on its behalf.

    Returns ``True`` if the cartridge was loaded prior to the call.
    """
    if name not in self.loaded_cartridges:
        return False
    tool_names = self._tool_names.pop(name, [])
    unregister_tools(tool_names)
    del self.loaded_cartridges[name]
    log.info(
        "Unloaded cartridge %r — %d tool(s) unregistered.",
        name,
        len(tool_names),
    )
    # Observer broadcast: fire AFTER the manager state is fully
    # consistent so callbacks observe the post-unload registry.
    self._fire_callbacks()
    return True

unregister_callback(fn)

Drop fn from the observer list. Returns True if removed.

Source code in wintermute/cartridges/manager.py 
127
128
129
130
131
132
133
def unregister_callback(self, fn: Callable[[], None]) -> bool:
    """Drop ``fn`` from the observer list. Returns ``True`` if removed."""
    try:
        self._callbacks.remove(fn)
    except ValueError:
        return False
    return True