ultanio/cobot

Fork 4

Architecture: unify hooks and extension points into a single dispatch mechanism #33

New issue

Closed

opened 2026-02-20 14:13:05 +00:00 by nazim · 1 comment

nazim commented

2026-02-20 14:13:05 +00:00

Contributor

Problem

Cobot currently has two distinct dispatch mechanisms for plugin communication:

1. Hooks (lifecycle pipeline)

11 fixed methods in base.py: on_message_received, transform_system_prompt, etc.
Every plugin can override them
registry.run_hook() calls all overriders in priority order, flowing ctx through the chain
Adding new hooks requires changing base.py + HOOK_METHODS list

2. Extension Points (service contracts)

Declared in PluginMeta.extension_points (what a plugin defines)
Fulfilled via PluginMeta.implements dict (what a plugin provides)
Consumers call registry.get_implementations("context.system_prompt") and iterate manually
Any plugin can define new ones without touching core

The confusion

Both mechanisms achieve similar goals but work differently:

	Hooks	Extension Points
Who defines	Core (`base.py`)	Any plugin
Direction	Broadcast pipeline	Service lookup
Cardinality	All plugins run	Only implementers
Adding new	Requires core change	Any plugin can define
Error handling	Built into `run_hook`	Manual per-consumer
Registry access	Automatic via `run_hook`	Manual `get_registry()` + iteration

Concrete overlap example: transform_system_prompt is a hook that compaction uses, but context.system_prompt is an extension point that soul/memory/workspace use. Both contribute to the system prompt. A new plugin author must know which to use.

Current `get_registry()` Coupling

Three different patterns exist for registry access:

Global singleton: session.start() calls get_registry() directly
Manual injection: agent.py calls tools.set_registry(registry) and compaction.set_registry(registry)
Unset: context._registry = None — hopes someone sets it

Every plugin that defines extension points needs the registry to call get_implementations(). The current approach doesn't scale — every new extension-point-defining plugin needs custom wiring in agent.py.

Proposal: Unify the Mechanism

Core idea

Make extension points and hooks use the same dispatch system. Extension points are just hooks that any plugin can define, where only explicit implementers get called.

Changes

1. Auto-inject registry into all plugins (in start_all()):

async def start_all(self):
    for plugin_id in self._load_order:
        plugin = self._plugins[plugin_id]
        plugin._registry = self  # every plugin gets it automatically
        await plugin.start()

This eliminates all set_registry() methods and get_registry() calls.

2. Add call_extension() to base Plugin class:

class Plugin(ABC):
    _registry = None  # injected by registry.start_all()

    async def call_extension(self, point: str, ctx: dict = None) -> list:
        """Call all implementations of an extension point.
        
        Like run_hook, but only calls plugins that explicitly
        implement the point via meta.implements.
        """
        if not self._registry:
            return []
        results = []
        for pid, plugin, method_name in self._registry.get_implementations(point):
            try:
                method = getattr(plugin, method_name)
                if asyncio.iscoroutinefunction(method):
                    result = await method(ctx) if ctx else await method()
                else:
                    result = method(ctx) if ctx else method()
                if result is not None:
                    results.append(result)
            except Exception as e:
                print(f"[{self.meta.id}] Error calling {pid}.{method_name}: {e}", file=sys.stderr)
        return results

3. Rewrite consumers to use call_extension:

Before (context plugin):

if self._registry:
    implementations = self._registry.get_implementations("context.system_prompt")
    for plugin_id, plugin, method_name in implementations:
        try:
            method = getattr(plugin, method_name)
            contribution = method()
            if contribution:
                parts.append(contribution)
        except Exception as e:
            print(f"Error: {e}")

After:

parts = await self.call_extension("context.system_prompt")

4. Make core hooks extension points of the agent itself (optional, longer term):

The 11 lifecycle hooks could become extension points defined by a core lifecycle plugin. Then HOOK_METHODS in base.py becomes unnecessary — plugins declare what lifecycle events they care about via implements, and new lifecycle events can be added by any plugin.

What stays the same

PluginMeta structure (extension_points, implements, capabilities, dependencies)
Plugin load order by priority
ctx dict flowing through chains
abort pattern for short-circuiting

Migration path

Add _registry auto-injection in start_all() → removes all set_registry() / get_registry()
Add call_extension() helper → reduces boilerplate in context, session, etc.
Gradually move hooks to extension points as plugins are updated
Eventually, HOOK_METHODS and the fixed hook methods in base.py become optional sugar

Discussion

Should the unified mechanism support both sync and async methods? (Current hooks are async, some extension point methods are sync)
Should call_extension flow ctx through a chain (like hooks) or collect results (like current extension points)? Or support both modes?
Is the implements dict in PluginMeta the right way to declare, or should plugins use decorators?

Analysis by @nazim based on reading base.py, registry.py, interfaces.py, agent.py, and the context/session/tools/compaction plugins.

## Problem Cobot currently has **two distinct dispatch mechanisms** for plugin communication: ### 1. Hooks (lifecycle pipeline) - 11 fixed methods in `base.py`: `on_message_received`, `transform_system_prompt`, etc. - Every plugin can override them - `registry.run_hook()` calls all overriders in priority order, flowing `ctx` through the chain - Adding new hooks requires changing `base.py` + `HOOK_METHODS` list ### 2. Extension Points (service contracts) - Declared in `PluginMeta.extension_points` (what a plugin defines) - Fulfilled via `PluginMeta.implements` dict (what a plugin provides) - Consumers call `registry.get_implementations("context.system_prompt")` and iterate manually - Any plugin can define new ones without touching core ### The confusion Both mechanisms achieve similar goals but work differently: | | Hooks | Extension Points | |---|---|---| | Who defines | Core (`base.py`) | Any plugin | | Direction | Broadcast pipeline | Service lookup | | Cardinality | All plugins run | Only implementers | | Adding new | Requires core change | Any plugin can define | | Error handling | Built into `run_hook` | Manual per-consumer | | Registry access | Automatic via `run_hook` | Manual `get_registry()` + iteration | **Concrete overlap example:** `transform_system_prompt` is a hook that compaction uses, but `context.system_prompt` is an extension point that soul/memory/workspace use. Both contribute to the system prompt. A new plugin author must know which to use. ## Current `get_registry()` Coupling Three different patterns exist for registry access: 1. **Global singleton:** `session.start()` calls `get_registry()` directly 2. **Manual injection:** `agent.py` calls `tools.set_registry(registry)` and `compaction.set_registry(registry)` 3. **Unset:** `context._registry = None` — hopes someone sets it Every plugin that defines extension points needs the registry to call `get_implementations()`. The current approach doesn't scale — every new extension-point-defining plugin needs custom wiring in `agent.py`. ## Proposal: Unify the Mechanism ### Core idea Make extension points and hooks use the **same dispatch system**. Extension points are just hooks that any plugin can define, where only explicit implementers get called. ### Changes **1. Auto-inject registry into all plugins** (in `start_all()`): ```python async def start_all(self): for plugin_id in self._load_order: plugin = self._plugins[plugin_id] plugin._registry = self # every plugin gets it automatically await plugin.start() ``` This eliminates all `set_registry()` methods and `get_registry()` calls. **2. Add `call_extension()` to base Plugin class:** ```python class Plugin(ABC): _registry = None # injected by registry.start_all() async def call_extension(self, point: str, ctx: dict = None) -> list: """Call all implementations of an extension point. Like run_hook, but only calls plugins that explicitly implement the point via meta.implements. """ if not self._registry: return [] results = [] for pid, plugin, method_name in self._registry.get_implementations(point): try: method = getattr(plugin, method_name) if asyncio.iscoroutinefunction(method): result = await method(ctx) if ctx else await method() else: result = method(ctx) if ctx else method() if result is not None: results.append(result) except Exception as e: print(f"[{self.meta.id}] Error calling {pid}.{method_name}: {e}", file=sys.stderr) return results ``` **3. Rewrite consumers to use `call_extension`:** Before (context plugin): ```python if self._registry: implementations = self._registry.get_implementations("context.system_prompt") for plugin_id, plugin, method_name in implementations: try: method = getattr(plugin, method_name) contribution = method() if contribution: parts.append(contribution) except Exception as e: print(f"Error: {e}") ``` After: ```python parts = await self.call_extension("context.system_prompt") ``` **4. Make core hooks extension points of the agent itself** (optional, longer term): The 11 lifecycle hooks could become extension points defined by a core `lifecycle` plugin. Then `HOOK_METHODS` in base.py becomes unnecessary — plugins declare what lifecycle events they care about via `implements`, and new lifecycle events can be added by any plugin. ### What stays the same - `PluginMeta` structure (extension_points, implements, capabilities, dependencies) - Plugin load order by priority - `ctx` dict flowing through chains - `abort` pattern for short-circuiting ### Migration path 1. Add `_registry` auto-injection in `start_all()` → removes all `set_registry()` / `get_registry()` 2. Add `call_extension()` helper → reduces boilerplate in context, session, etc. 3. Gradually move hooks to extension points as plugins are updated 4. Eventually, `HOOK_METHODS` and the fixed hook methods in `base.py` become optional sugar ## Discussion - Should the unified mechanism support both sync and async methods? (Current hooks are async, some extension point methods are sync) - Should `call_extension` flow ctx through a chain (like hooks) or collect results (like current extension points)? Or support both modes? - Is the `implements` dict in PluginMeta the right way to declare, or should plugins use decorators? --- *Analysis by @nazim based on reading base.py, registry.py, interfaces.py, agent.py, and the context/session/tools/compaction plugins.*

nazim commented

2026-02-20 14:26:09 +00:00

Author

Contributor

Revised Proposal: Loop Plugin + Minimal Agent

After discussion with @k9ert, a cleaner design emerged:

The hooks are legacy — they belong to the agent loop

The 11 hooks in base.py are all called by agent.py. They're not a separate mechanism — they're extension points of the agent loop that were never ported to the extension point system.

Solution: Extract the loop into a plugin

agent.py becomes trivially simple:

class Cobot:
    def __init__(self, registry):
        self.registry = registry

    async def run(self):
        await self.registry.start_all()
        loops = self.registry.all_with_capability("loop")
        await asyncio.gather(*[loop.run() for loop in loops])

The agent is a loop runner. Nothing else. One extension point: "start all loops."

The current agent logic moves to a loop plugin:

class AgentLoopPlugin(Plugin):
    meta = PluginMeta(
        id="agent-loop",
        version="1.0.0",
        capabilities=["loop"],
        dependencies=["session", "context"],
        extension_points=[
            "loop.on_message",
            "loop.before_llm",
            "loop.after_llm",
            "loop.before_tool",
            "loop.after_tool",
            "loop.before_send",
            "loop.after_send",
            "loop.on_error",
        ],
        priority=50,
    )

    async def run(self):
        """The main poll → context → LLM → tools → send cycle."""
        while True:
            messages = await self.call_extension("session.receive")
            for msg in messages:
                await self._handle(msg)
            await asyncio.sleep(self._interval)

All 11 current hooks become extension points on the loop plugin. Existing plugins switch from overriding on_message_received() to implements: {"loop.on_message": "my_handler"}.

Multiple loops run concurrently

Any plugin with capabilities=["loop"] gets run by the agent. Examples:

agent-loop — the main poll → LLM → respond cycle
heartbeat-loop — periodic background tasks
dvm-loop — NIP-90 job listener
cron-loop — scheduled task execution

asyncio.gather runs them all concurrently. Each loop defines its own extension points for other plugins to hook into.

What gets deleted

HOOK_METHODS list in base.py
All 11 no-op hook methods in Plugin base class
run_hook() in registry (replaced by call_extension())
run() helper in __init__.py
200+ lines of loop logic in agent.py
All set_registry() / get_registry() patterns (auto-inject instead)

Migration path

Auto-inject _registry into all plugins in start_all()
Add call_extension() helper to base Plugin
Create agent-loop plugin with the current agent.py logic
Port existing hook overrides to implements declarations
Simplify agent.py to just the loop runner
Remove legacy hook infrastructure from base.py/registry.py

Benefits

One dispatch mechanism — extension points only, no hooks
Swappable loops — want a different agent architecture? Write a different loop plugin
Composable — heartbeat, DVM, cron all as independent loop plugins
agent.py is ~10 lines — pure bootstrap
Testable — mock loops, test plugins in isolation

## Revised Proposal: Loop Plugin + Minimal Agent After discussion with @k9ert, a cleaner design emerged: ### The hooks are legacy — they belong to the agent loop The 11 hooks in `base.py` are all called by `agent.py`. They're not a separate mechanism — they're extension points of the agent loop that were never ported to the extension point system. ### Solution: Extract the loop into a plugin **agent.py becomes trivially simple:** ```python class Cobot: def __init__(self, registry): self.registry = registry async def run(self): await self.registry.start_all() loops = self.registry.all_with_capability("loop") await asyncio.gather(*[loop.run() for loop in loops]) ``` The agent is a loop runner. Nothing else. One extension point: "start all loops." **The current agent logic moves to a loop plugin:** ```python class AgentLoopPlugin(Plugin): meta = PluginMeta( id="agent-loop", version="1.0.0", capabilities=["loop"], dependencies=["session", "context"], extension_points=[ "loop.on_message", "loop.before_llm", "loop.after_llm", "loop.before_tool", "loop.after_tool", "loop.before_send", "loop.after_send", "loop.on_error", ], priority=50, ) async def run(self): """The main poll → context → LLM → tools → send cycle.""" while True: messages = await self.call_extension("session.receive") for msg in messages: await self._handle(msg) await asyncio.sleep(self._interval) ``` All 11 current hooks become extension points on the loop plugin. Existing plugins switch from overriding `on_message_received()` to `implements: {"loop.on_message": "my_handler"}`. ### Multiple loops run concurrently Any plugin with `capabilities=["loop"]` gets run by the agent. Examples: - **agent-loop** — the main poll → LLM → respond cycle - **heartbeat-loop** — periodic background tasks - **dvm-loop** — NIP-90 job listener - **cron-loop** — scheduled task execution `asyncio.gather` runs them all concurrently. Each loop defines its own extension points for other plugins to hook into. ### What gets deleted - `HOOK_METHODS` list in `base.py` - All 11 no-op hook methods in `Plugin` base class - `run_hook()` in registry (replaced by `call_extension()`) - `run()` helper in `__init__.py` - 200+ lines of loop logic in `agent.py` - All `set_registry()` / `get_registry()` patterns (auto-inject instead) ### Migration path 1. Auto-inject `_registry` into all plugins in `start_all()` 2. Add `call_extension()` helper to base `Plugin` 3. Create `agent-loop` plugin with the current agent.py logic 4. Port existing hook overrides to `implements` declarations 5. Simplify `agent.py` to just the loop runner 6. Remove legacy hook infrastructure from base.py/registry.py ### Benefits - **One dispatch mechanism** — extension points only, no hooks - **Swappable loops** — want a different agent architecture? Write a different loop plugin - **Composable** — heartbeat, DVM, cron all as independent loop plugins - **agent.py is ~10 lines** — pure bootstrap - **Testable** — mock loops, test plugins in isolation