FusionAGI/fusionagi/reasoning/insight_bus.py

"""Cross-head insight bus — shared learning channel between heads.

Heads can publish observations (insights) to the bus, and other heads
can subscribe to learn from them.  This enables the Safety head to
learn from Logic's contradiction detections, Research's source quality
assessments, and so on — breaking the head-isolation barrier.

Usage:

    bus = InsightBus()
    bus.publish("logic", Insight(source="logic", message="Contradiction found", ...))
    recent = bus.get_insights(subscriber="safety", limit=10)
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from typing import Any

from fusionagi._logger import logger


@dataclass
class Insight:
    """A single observation published by a head."""

    source: str
    message: str
    domain: str = ""
    confidence: float = 0.5
    metadata: dict[str, Any] = field(default_factory=dict)
    timestamp: float = field(default_factory=time.monotonic)


class InsightBus:
    """Shared bus for cross-head learning.

    Heads publish observations; other heads consume them to enrich
    their own reasoning.  The bus maintains a rolling window of
    insights and supports filtered retrieval.

    Args:
        max_insights: Maximum insights retained (oldest dropped first).
    """

    def __init__(self, max_insights: int = 1000) -> None:
        self._insights: list[Insight] = []
        self._max = max_insights
        self._subscribers: dict[str, list[str]] = {}

    def publish(self, publisher: str, insight: Insight) -> None:
        """Publish an insight from a head.

        Args:
            publisher: Head ID of the publisher.
            insight: The observation to share.
        """
        self._insights.append(insight)
        if len(self._insights) > self._max:
            self._insights = self._insights[-self._max:]

        logger.debug(
            "InsightBus: insight published",
            extra={
                "publisher": publisher,
                "domain": insight.domain,
                "message": insight.message[:80],
            },
        )

    def subscribe(self, subscriber: str, domains: list[str] | None = None) -> None:
        """Register a head's interest in certain domains.

        Args:
            subscriber: Head ID subscribing.
            domains: Domains of interest (None = all).
        """
        self._subscribers[subscriber] = domains or []

    def get_insights(
        self,
        subscriber: str | None = None,
        domain: str | None = None,
        limit: int = 20,
        since: float | None = None,
    ) -> list[Insight]:
        """Retrieve recent insights, optionally filtered.

        Args:
            subscriber: If given, filter by subscriber's registered domains.
            domain: Explicit domain filter.
            limit: Max results.
            since: Only insights after this timestamp.

        Returns:
            List of matching insights, most recent first.
        """
        results = self._insights

        if since is not None:
            results = [i for i in results if i.timestamp >= since]

        if domain:
            results = [i for i in results if i.domain == domain]
        elif subscriber and subscriber in self._subscribers:
            domains = self._subscribers[subscriber]
            if domains:
                results = [i for i in results if i.domain in domains]

        return list(reversed(results[-limit:]))

    def get_summary(self) -> dict[str, Any]:
        """Return bus statistics."""
        by_source: dict[str, int] = {}
        by_domain: dict[str, int] = {}
        for i in self._insights:
            by_source[i.source] = by_source.get(i.source, 0) + 1
            if i.domain:
                by_domain[i.domain] = by_domain.get(i.domain, 0) + 1
        return {
            "total_insights": len(self._insights),
            "subscribers": list(self._subscribers.keys()),
            "by_source": by_source,
            "by_domain": by_domain,
        }


__all__ = ["Insight", "InsightBus"]