Born Ready

# mount() registers the event handler AND
# immediately runs setup if the tool is
# already there — because it might not be.
coordinator.hooks.register(
  "skills:discovered", on_skills_discovered
)
if coordinator.get_capability(
    TOOL_SKILLS_DISCOVERY_CAPABILITY
) is not None:
    # duplicate path — tool mounted first
    await _refresh_watched_skills(...)
        

async def _discover_events(coordinator):
    # Path 1: core canonical list
    discovered = set(ALL_EVENTS)
    # Path 2: contribution channels
    contributions = await coordinator
        .collect_contributions("obs.events")
    for ev in contributions:
        discovered.update(ev)
    # Path 3: legacy capability fallback
    cap = coordinator.get_capability(
        "observability.events")
    if cap is not None:
        discovered.update(cap() if callable(cap)
                           else cap)
    return discovered  # 3 paths for 1 answer
        

# tool-delegate/src/__init__.py
obs_events = coordinator.get_capability(
    "observability.events") or []
obs_events.extend([
    "delegate:agent_spawned",
    "delegate:agent_completed",
    ...3 more...
])
coordinator.register_capability(
    "observability.events", obs_events
)
# register_capability ≠ register_contributor
# collect_contributions() never sees this
        

# tool-delegate/src/__init__.py
coordinator.register_contributor(
    "observability.events",
    "tool-delegate",
    lambda: [
        "delegate:agent_spawned",
        "delegate:agent_resumed",
        "delegate:agent_completed",
        "delegate:agent_cancelled",
        "delegate:error",
    ],
)
# now visible to collect_contributions()
        

async def on_session_ready(coordinator) -> None:
    """Called after ALL modules have completed
    their mount() calls. The coordinator is
    fully composed. Return value is ignored.
    Exceptions are non-fatal and isolated.
    """
    # Tools are here. Providers are here.
    # Every hook is here. No guessing.
    contributions = await coordinator
        .collect_contributions("obs.events")
        

Design Decisions

• coordinator only, no config — config belongs in mount()
• async-only — sync implementations log a warning, are skipped
• non-fatal, isolated — one failure cannot block others
• fires before session:fork — children run independently
• 100% backward compatible — existing modules unchanged
• queue drains after dispatch — no double-dispatch risk

async def mount(coordinator, config=None):
    config = config or {}
    priority = int(config.get("priority", 100))
    session_logger = SessionLogger(...)
    
    # define handler closure...
    async def handler(event, data):
        session_logger.write(...)
    
    events = list(ALL_EVENTS)
    if auto_discover:
        # race condition: tools may not be mounted
        discovered = coordinator.get_capability(
            "observability.events") or []
        events.extend(discovered)
    
    for ev in events:
        coordinator.hooks.register(ev, handler)
        

_deferred_configs: list[dict] = []

async def mount(coordinator, config=None):
    _deferred_configs.append(config or {})

async def on_session_ready(coordinator):
    while _deferred_configs:
        config = _deferred_configs.pop(0)
        # ... build logger, closure ...

        # legacy path preserved
        if auto_discover:
            discovered = coordinator
                .get_capability("obs.events") or []
            events.extend(discovered)
        # new path — safe here, all tools mounted
        if auto_discover:
            contributions = await coordinator
                .collect_contributions("obs.events")
            for c in contributions:
                events.extend(c)
        for ev in events:
            coordinator.hooks.register(ev, handler)
        

Know Before You Ship

Failure Is Observable

The Pattern

• mount() — store config only
• on_session_ready() — do cross-module work
• collect_contributions() — safe only here
• DTU e2e profile — prove the full stack

The Stack