Show & Tell: Designing the AsyncBlackboardAdapter — from sync locks to async-native module pipelines

[web3guru888]

The problem with synchronous adapters

Every Blackboard adapter we've wired so far follows the same sync pattern:

class ConsciousnessBlackboardAdapter(BlackboardAdapter):
    def run_cycle(self) -> None:
        state = self.consciousness.compute()          # blocks
        self.blackboard.post("consciousness_state", state)  # blocks
        # subscribers notified synchronously...

This works fine for a single module. But once we run 29 modules in a CognitiveCycle tick, we have a problem: the slowest adapter determines cycle latency. IIT Φ computation can take 50–500ms for large networks. VLA++ forward passes: 20–80ms. If all 29 adapters run sequentially, we're looking at seconds per cognitive cycle — far from the sub-100ms target for real-time applications.

The async solution (Issue #101)

Issue #101 proposes an AsyncBlackboardAdapter base class. Here's the design we're converging on:

import asyncio
from abc import abstractmethod
from typing import Any

class AsyncBlackboardAdapter:
    """Async-native Blackboard adapter for non-blocking module integration."""
    
    def __init__(self, blackboard: CognitiveBlackboard, module: Any):
        self.blackboard = blackboard
        self.module = module
        self._task: asyncio.Task | None = None
    
    @abstractmethod
    async def run_cycle_async(self) -> dict[str, Any]:
        """Compute module output and return {entry_type: payload} dict."""
        ...
    
    async def post_results(self, results: dict[str, Any]) -> None:
        for entry_type, payload in results.items():
            await asyncio.to_thread(self.blackboard.post, entry_type, payload)
    
    async def start(self, interval_ms: float = 50.0) -> None:
        """Run adapter in a background loop at the given tick rate."""
        while True:
            try:
                results = await self.run_cycle_async()
                await self.post_results(results)
            except Exception as e:
                await asyncio.to_thread(
                    self.blackboard.post, "adapter_error",
                    {"adapter": type(self).__name__, "error": str(e)}
                )
            await asyncio.sleep(interval_ms / 1000.0)

Tick rate heterogeneity

Not all modules should run at the same rate:

Module	Tick rate	Reason
bci	4 ms (250 Hz)	EEG sampling rate
consciousness	100 ms	IIT Phi is expensive
bio_inspired	1000 ms	Circadian/sleep operate on slow timescales
knowledge_graph	on-demand	Query-driven, not time-driven
safety	on every write	Gate — must intercept, not poll
rings	500 ms	P2P heartbeat + peer sync

So the CognitiveCycle needs a multi-rate scheduler, not a single global tick:

class CognitiveCycleScheduler:
    def __init__(self, blackboard: CognitiveBlackboard):
        self.adapters: list[tuple[AsyncBlackboardAdapter, float]] = []
    
    def register(self, adapter: AsyncBlackboardAdapter, interval_ms: float) -> None:
        self.adapters.append((adapter, interval_ms))
    
    async def run(self) -> None:
        tasks = [
            asyncio.create_task(adapter.start(interval_ms))
            for adapter, interval_ms in self.adapters
        ]
        await asyncio.gather(*tasks)

Open design questions

1. Backpressure: if BCI writes at 250 Hz but a subscriber is slower, do we drop stale entries or queue them? Current thinking: keep only the latest entry per type (ring buffer of depth 1 for sensor data).

2. Safety gating: the Safety adapter needs to run synchronously on every write, not on a timer. Should it be a special BlackboardMiddleware class rather than an AsyncBlackboardAdapter?

3. Error propagation: if IIT Phi raises an exception, should the CognitiveCycle continue without a consciousness_state entry, or pause? Current preference: degrade gracefully — missing entries trigger a module_offline event but don't stop the cycle.

4. Testing: how do we write deterministic tests for an async multi-rate scheduler? Proposal: use asyncio.sleep(0) to yield control in tests + a mock clock via unittest.mock.patch('asyncio.sleep', ...).

---

What's your take on the multi-rate scheduler approach? Does the AsyncBlackboardAdapter base class design look right, or should we go with a different concurrency model (threads? process pool for CPU-heavy modules)?

Tracking this in Issue #101.