"""Intent bus — broadcast intents, agents self-select, collect results."""

from __future__ import annotations

import asyncio
import time
import logging
from collections import defaultdict
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import TYPE_CHECKING, Any, Callable, Awaitable

from probos.types import IntentMessage, IntentResult, Priority
from probos.mesh.signal import SignalManager

if TYPE_CHECKING:
    from probos.mesh.nats_bus import NATSBus

logger = logging.getLogger(__name__)

# Type for subscriber callbacks
IntentHandler = Callable[[IntentMessage], Awaitable[IntentResult | None]]


@dataclass
class IntentMetrics:
    """Tracks intent broadcast statistics (AD-470)."""

    broadcast_count: int = 0
    send_count: int = 0
    total_results: int = 0
    type_counts: dict[str, int] = field(default_factory=lambda: defaultdict(int))
    type_durations_ms: dict[str, list[float]] = field(default_factory=lambda: defaultdict(list))

    def record_broadcast(self, intent_type: str, result_count: int, duration_ms: float) -> None:
        """Record a broadcast completion."""
        self.broadcast_count += 1
        self.total_results += result_count
        self.type_counts[intent_type] += 1
        durations = self.type_durations_ms[intent_type]
        durations.append(duration_ms)
        if len(durations) > 200:
            self.type_durations_ms[intent_type] = durations[-200:]

    def record_send(self, intent_type: str, duration_ms: float) -> None:
        """Record a directed send completion."""
        self.send_count += 1
        self.type_counts[intent_type] += 1
        durations = self.type_durations_ms[intent_type]
        durations.append(duration_ms)
        if len(durations) > 200:
            self.type_durations_ms[intent_type] = durations[-200:]

    def get_summary(self) -> dict[str, Any]:
        """Return a summary of intent metrics."""
        type_stats: dict[str, dict[str, Any]] = {}
        for intent_type, durations in self.type_durations_ms.items():
            if durations:
                type_stats[intent_type] = {
                    "count": self.type_counts[intent_type],
                    "mean_ms": round(sum(durations) / len(durations), 2),
                    "max_ms": round(max(durations), 2),
                }
        return {
            "broadcast_count": self.broadcast_count,
            "send_count": self.send_count,
            "total_results": self.total_results,
            "types": type_stats,
        }


class IntentBus:
    """Async pub/sub for intent broadcasting.

    Agents subscribe with a handler. When an intent is broadcast,
    all subscribers are notified concurrently. Each subscriber decides
    whether to respond (self-selection). Results are collected with
    a configurable timeout.
    """

    def __init__(self, signal_manager: SignalManager) -> None:
        self._signal_manager = signal_manager
        self._subscribers: dict[str, IntentHandler] = {}  # agent_id -> handler
        self._intent_index: dict[str, set[str]] = {}  # intent_name -> set of agent_ids
        self._pending_results: dict[str, list[IntentResult]] = {}  # intent_id -> results
        self._result_events: dict[str, asyncio.Event] = {}
        self._broadcast_timestamps: list[tuple[float, str]] = []  # (monotonic_time, intent_name)
        self._window_seconds: float = 60.0
        self._federation_fn: Callable[[IntentMessage], Awaitable[list[IntentResult]]] | None = None
        self._nats_bus: Any = None  # AD-637b: wired via set_nats_bus()
        self._pending_sub_tasks: set[asyncio.Task] = set()  # AD-637z: tracked NATS sub tasks
        # BF-223: Defer JetStream dispatch consumer creation until after ship
        # commissioning sets the correct NATS subject prefix (DID-based).
        # During startup, subscribe() skips dispatch consumers; finalize.py
        # calls create_dispatch_consumers() after prefix is stable.
        self._defer_dispatch_consumers: bool = True
        # AD-654b: Per-agent cognitive queues
        self._agent_queues: dict[str, Any] = {}  # agent_id -> AgentCognitiveQueue
        # AD-654b: Injected callback for response recording (replaces handler.__self__ reach-through)
        self._record_response: Callable[[str, str], None] | None = None  # (agent_id, thread_id)

        # BF-234: Consumer-side dedup — tracks recently-seen intent IDs to
        # suppress transport-layer duplicates (JetStream redelivery, js_publish
        # timeout-then-succeed). Keyed by intent_id, value is monotonic timestamp.
        self._seen_intents: dict[str, float] = {}
        self._last_seen_eviction: float = time.monotonic()
        self._duplicate_suppressed_count: int = 0

        # BF-234: Injected event emitter for duplicate-suppressed telemetry.
        # Wired from finalize.py via set_emit_event().
        self._emit_event_fn: Callable[[str, dict[str, Any]], None] | None = None
        # AD-470: Intent metrics
        self._metrics = IntentMetrics()

    def subscribe(self, agent_id: str, handler: IntentHandler, intent_names: list[str] | None = None) -> None:
        """Register an agent's intent handler.

        If intent_names is provided, the agent is indexed for those intents
        and will only be invoked when a matching intent is broadcast.
        Agents subscribed without intent_names receive all broadcasts (fallback).
        """
        self._subscribers[agent_id] = handler
        if intent_names:
            for name in intent_names:
                if name not in self._intent_index:
                    self._intent_index[name] = set()
                self._intent_index[name].add(agent_id)

        # AD-637b/z: Create NATS subscription for targeted send()
        if self._nats_bus and self._nats_bus.connected:
            try:
                loop = asyncio.get_running_loop()
                task = loop.create_task(
                    self._nats_subscribe_agent(agent_id, handler),
                    name=f"nats-sub-{agent_id[:12]}",
                )
                self._pending_sub_tasks.add(task)
                task.add_done_callback(self._pending_sub_tasks.discard)
                task.add_done_callback(self._on_nats_task_done)
                # AD-654a: Also subscribe to JetStream dispatch subject
                # BF-223: Skip during startup — deferred until prefix is stable
                if not self._defer_dispatch_consumers:
                    dispatch_task = loop.create_task(
                        self._js_subscribe_agent_dispatch(agent_id, handler),
                        name=f"js-dispatch-sub-{agent_id[:12]}",
                    )
                    self._pending_sub_tasks.add(dispatch_task)
                    dispatch_task.add_done_callback(self._pending_sub_tasks.discard)
                    dispatch_task.add_done_callback(self._on_nats_task_done)
            except RuntimeError:
                pass

    async def create_dispatch_consumers(self) -> None:
        """Create JetStream dispatch consumers for all subscribed agents.

        BF-223: Called from finalize.py after ship commissioning sets the
        correct DID-based NATS prefix. This ensures durable consumers are
        created with the stable prefix, eliminating the prefix race.

        Also clears _defer_dispatch_consumers so that agents created after
        startup (e.g., via self-modification) get immediate dispatch consumers.
        """
        self._defer_dispatch_consumers = False
        if not self._nats_bus or not self._nats_bus.connected:
            logger.debug("BF-223: NATS not connected, skipping dispatch consumer creation")
            return

        count = 0
        for agent_id, handler in self._subscribers.items():
            try:
                await self._js_subscribe_agent_dispatch(agent_id, handler)
                count += 1
            except Exception as e:
                logger.warning(
                    "BF-223: Dispatch consumer creation failed for %s: %s",
                    agent_id[:12], e,
                )
        logger.info("BF-223: Created %d JetStream dispatch consumers (prefix-stable)", count)

    async def _nats_subscribe_agent(self, agent_id: str, handler: IntentHandler) -> None:
        """Subscribe an agent to their NATS intent subject for send() delivery."""
        subject = f"intent.{agent_id}"

        async def _on_nats_intent(msg: Any) -> None:
            """NATS message adapter: deserialize → handler → serialize reply."""
            try:
                intent = self._deserialize_intent(msg.data)
                result = await handler(intent)
                if msg.reply:
                    if result is not None:
                        await msg.respond(self._serialize_result(result))
                    else:
                        # Agent declined — send empty success response
                        await msg.respond({"declined": True})
            except Exception as e:
                logger.warning("NATS intent handler error for %s: %s", agent_id[:8], e)
                if msg.reply:
                    error_result = IntentResult(
                        intent_id=msg.data.get("id", "") if isinstance(msg.data, dict) else "",
                        agent_id=agent_id,
                        success=False,
                        error=str(e),
                        confidence=0.0,
                    )
                    await msg.respond(self._serialize_result(error_result))

        sub = await self._nats_bus.subscribe(subject, _on_nats_intent)

    async def _js_subscribe_agent_dispatch(self, agent_id: str, handler: IntentHandler) -> None:
        """Subscribe agent to their JetStream dispatch subject (AD-654a).

        Creates a durable consumer on intent.dispatch.{agent_id} within
        the INTENT_DISPATCH stream. Messages queue while agent is busy
        and are processed sequentially (max_ack_pending=1).

        Uses manual_ack=True because cognitive chains need msg.term() on
        error (not msg.nak()) — LLM calls that already ran must not retry.
        """
        subject = f"intent.dispatch.{agent_id}"

        async def _on_dispatch(msg: Any) -> None:
            """JetStream dispatch callback — deserialize and enqueue.

            AD-654b: Enqueues to cognitive queue instead of inline processing.
            The queue manages ack/term, priority ordering, and handler dispatch.
            """
            try:
                intent_msg = self._deserialize_intent(msg.data)

                # BF-234: Consumer-side dedup gate — suppress transport-layer
                # duplicates (JetStream redelivery, js_publish timeout-then-succeed).
                # Only ward_room_notification intents need this; other intent types
                # are idempotent or use request/reply (not fire-and-forget).
                # NOTE: This is structural dedup (same intent.id delivered twice).
                # BF-198 record_agent_response/has_agent_responded is semantic
                # round-tracking (agent spoke in this thread) — different invariant.
                if intent_msg.intent == "ward_room_notification":
                    if self._is_duplicate_intent(intent_msg.id):
                        _first_seen_ts = self._seen_intents[intent_msg.id]
                        _age_ms = (time.monotonic() - _first_seen_ts) * 1000
                        self._duplicate_suppressed_count += 1
                        logger.warning(
                            "BF-234: Suppressed duplicate ward_room_notification "
                            "for %s (intent=%s, age=%.0fms, total_suppressed=%d)",
                            agent_id[:12], intent_msg.id[:8], _age_ms,
                            self._duplicate_suppressed_count,
                        )
                        if self._emit_event_fn:
                            self._emit_event_fn(
                                "wardroom.dispatch.duplicate_suppressed",
                                {
                                    "agent_id": agent_id,
                                    "thread_id": intent_msg.params.get("thread_id", ""),
                                    "intent_id": intent_msg.id,
                                    "age_ms": round(_age_ms, 1),
                                },
                            )
                        await msg.ack()
                        return
                    self._record_seen_intent(intent_msg.id)
                    self._maybe_evict_seen_intents()  # periodic sweep — after gate, not before

                # AD-654a/BF-198: Record response BEFORE handler runs to close
                # the proactive-loop race window.
                # Uses injected callback instead of handler.__self__ reach-through.
                if self._record_response:
                    _thread_id = intent_msg.params.get("thread_id", "")
                    if _thread_id:
                        self._record_response(intent_msg.target_agent_id, _thread_id)

                # AD-654b: Enqueue with priority classification
                queue = self._get_agent_queue(agent_id)
                if queue:
                    priority = Priority.classify(
                        intent=intent_msg.intent,
                        is_captain=intent_msg.params.get("is_captain", False),
                        was_mentioned=intent_msg.params.get("was_mentioned", False),
                    )
                    accepted = queue.enqueue(intent_msg, priority, js_msg=msg)
                    if not accepted:
                        # Queue rejected it (full + lower priority). term() it.
                        await msg.term()
                else:
                    # No queue — fall back to direct handler.
                    # This is normal for substrate agents (IntrospectAgent, VitalsMonitor, etc.)
                    # which don't have cognitive queues. Log at debug, not warning.
                    logger.debug("AD-654b: No queue for %s, direct dispatch", agent_id[:12])
                    await handler(intent_msg)
                    await msg.ack()
            except Exception as e:
                logger.warning(
                    "AD-654b: Dispatch callback error for %s: %s",
                    agent_id[:8], e,
                )
                await msg.term()

        # Durable name must be NATS-safe (alphanumeric + dash).
        durable_name = f"agent-dispatch-{agent_id}"

        # AD-654b: max_deliver=10 bounds nak() redelivery loops.
        # With circuit breaker nak(delay=60) + max_deliver=10, a stuck breaker
        # causes at most 10 redeliveries (~10 min) before JetStream auto-discards.
        # Without this, nak loops are unbounded.
        sub = await self._nats_bus.js_subscribe(
            subject,
            _on_dispatch,
            durable=durable_name,
            stream="INTENT_DISPATCH",
            max_ack_pending=1,
            ack_wait=300,
            manual_ack=True,
            max_deliver=10,
        )
        if sub:
            logger.debug("AD-654b: JetStream dispatch consumer for %s", agent_id[:12])

    def unsubscribe(self, agent_id: str) -> None:
        """Remove an agent's subscription and intent index entries."""
        self._subscribers.pop(agent_id, None)
        self.unregister_queue(agent_id)  # AD-654b: clean up cognitive queue
        for agent_set in self._intent_index.values():
            agent_set.discard(agent_id)
        # AD-637z: Clean up NATS subscription via NATSBus lifecycle management
        if self._nats_bus:
            subject = f"intent.{agent_id}"
            try:
                loop = asyncio.get_running_loop()
                task = loop.create_task(
                    self._nats_bus.remove_tracked_subscription(subject),
                    name=f"nats-unsub-{agent_id[:12]}",
                )
                self._pending_sub_tasks.add(task)
                task.add_done_callback(self._pending_sub_tasks.discard)
                task.add_done_callback(self._on_nats_task_done)
            except RuntimeError:
                pass
            # AD-654a: Clean up JetStream dispatch consumer
            durable_name = f"agent-dispatch-{agent_id}"
            try:
                loop = asyncio.get_running_loop()
                loop.create_task(
                    self._nats_bus.delete_consumer("INTENT_DISPATCH", durable_name),
                    name=f"cleanup-dispatch-{agent_id[:12]}",
                )
            except RuntimeError:
                pass

    def _on_nats_task_done(self, task: asyncio.Task) -> None:
        """Log errors from NATS subscribe/unsubscribe tasks."""
        if task.cancelled():
            return
        exc = task.exception()
        if exc:
            logger.warning("NATS sub/unsub task failed: %s", exc)

    @property
    def subscriber_count(self) -> int:
        return len(self._subscribers)

    async def send(self, intent: IntentMessage) -> IntentResult | None:
        """Deliver an intent to a specific agent (targeted dispatch, AD-397).

        AD-637b: Uses NATS request/reply when connected, direct-call fallback otherwise.
        Only one path is used per call — never both.

        AD-637z: BF-221 lifted. Prefix re-subscription (set_subject_prefix)
        ensures NATS subscriptions survive the Phase 7 DID assignment.
        """
        if not intent.target_agent_id:
            raise ValueError("send() requires target_agent_id")

        _send_start = time.monotonic()  # AD-470: timing
        try:
            # NATS path when connected
            if self._nats_bus and self._nats_bus.connected:
                return await self._nats_send(intent)

            # Direct-call fallback when NATS disconnected
            handler = self._subscribers.get(intent.target_agent_id)
            if handler is None:
                return None
            try:
                result = await asyncio.wait_for(handler(intent), timeout=intent.ttl_seconds)
                return result
            except asyncio.TimeoutError:
                return IntentResult(
                    intent_id=intent.id,
                    agent_id=intent.target_agent_id,
                    success=False,
                    error="Agent did not respond in time.",
                    confidence=0.0,
                )
        finally:
            _elapsed_ms = (time.monotonic() - _send_start) * 1000
            self._metrics.record_send(intent.intent, _elapsed_ms)

    async def _nats_send(self, intent: IntentMessage) -> IntentResult | None:
        """Send intent via NATS request/reply to target agent."""
        subject = f"intent.{intent.target_agent_id}"
        try:
            reply = await asyncio.wait_for(
                self._nats_bus.request(
                    subject,
                    self._serialize_intent(intent),
                    timeout=intent.ttl_seconds,
                ),
                timeout=intent.ttl_seconds,
            )
        except asyncio.TimeoutError:
            logger.warning("NATS send timeout: %s → %s", intent.intent, intent.target_agent_id[:12])
            return IntentResult(
                intent_id=intent.id,
                agent_id=intent.target_agent_id or "",
                success=False,
                error="Agent did not respond in time.",
                confidence=0.0,
            )
        if reply is None:
            return None
        data = reply.data if hasattr(reply, 'data') else reply
        if isinstance(data, dict) and data.get("declined"):
            return None
        return self._deserialize_result(data)

    async def broadcast(
        self,
        intent: IntentMessage,
        timeout: float | None = None,
        *,
        federated: bool = True,
    ) -> list[IntentResult]:
        """Broadcast an intent to all subscribers, collect results.

        Each subscriber is called concurrently. Subscribers that return
        None are treated as having declined the intent (self-deselected).
        Waits up to `timeout` seconds (defaults to intent TTL) for results.

        If intent.target_agent_id is set, delegates to send() for targeted dispatch.
        """
        # AD-397: targeted dispatch
        if intent.target_agent_id:
            result = await self.send(intent)
            return [result] if result else []

        # AD-698: pre-intent authorization seam. Default-empty registry
        # means zero overhead. Overlay packages register hooks for RBAC,
        # rate limiting, etc.
        try:
            from probos.extensions.overlay import evaluate_pre_intent_authorization
            allowed, reason = evaluate_pre_intent_authorization(intent)
        except Exception:
            allowed, reason = True, ""
        if not allowed:
            logger.info(
                "AD-698: intent %s denied by pre-auth hook '%s'", intent.intent, reason,
            )
            return []

        timeout = timeout if timeout is not None else intent.ttl_seconds
        _broadcast_start = time.monotonic()  # AD-470: timing

        self.record_broadcast(intent.intent)
        self._signal_manager.track(intent)
        self._pending_results[intent.id] = []

        logger.info(
            "Intent broadcast: %s id=%s urgency=%.1f subscribers=%d",
            intent.intent,
            intent.id[:8],
            intent.urgency,
            len(self._subscribers),
        )

        # Determine which agents to fan out to
        indexed_agents = self._intent_index.get(intent.intent)
        if indexed_agents is not None:
            # Pre-filtered: only invoke agents indexed for this intent
            # Plus any agents not in the index at all (fallback subscribers)
            all_indexed = set()
            for agent_set in self._intent_index.values():
                all_indexed.update(agent_set)
            candidates = {
                aid: handler
                for aid, handler in self._subscribers.items()
                if aid in indexed_agents or aid not in all_indexed
            }
        else:
            # No index entry: fall back to all subscribers
            candidates = dict(self._subscribers)

        # Fan out to selected subscribers concurrently
        tasks = []
        for agent_id, handler in list(candidates.items()):
            tasks.append(
                asyncio.create_task(
                    self._invoke_handler(intent, agent_id, handler),
                    name=f"intent-{intent.id[:8]}-{agent_id[:8]}",
                )
            )

        if tasks:
            # Wait for all handlers, bounded by timeout
            done, pending = await asyncio.wait(tasks, timeout=timeout)
            # Cancel stragglers
            for task in pending:
                task.cancel()

        results = self._pending_results.pop(intent.id, [])
        self._signal_manager.untrack(intent.id)

        # AD-470: Record metrics
        elapsed_ms = (time.monotonic() - _broadcast_start) * 1000
        self._metrics.record_broadcast(intent.intent, len(results), elapsed_ms)

        # Federation: forward to peers if enabled and not an inbound federated intent
        if federated and self._federation_fn:
            try:
                remote_results = await self._federation_fn(intent)
                results.extend(remote_results)
            except Exception as e:
                logger.debug("Federation forwarding failed: %s", e)

        logger.info(
            "Intent resolved: %s id=%s results=%d",
            intent.intent,
            intent.id[:8],
            len(results),
        )
        return results

    async def publish(self, intent: IntentMessage, **kwargs: Any) -> list[IntentResult]:
        """Alias for broadcast() — used by WatchManager dispatch (runtime.py:689)."""
        return await self.broadcast(intent, **kwargs)

    async def dispatch_async(self, intent: IntentMessage) -> None:
        """Fire-and-forget dispatch to a specific agent via JetStream (AD-654a).

        Publishes the intent to the agent's durable JetStream consumer.
        No reply expected — the agent processes asynchronously and posts
        its own response. Falls back to direct async handler invocation
        when NATS/JetStream is unavailable.

        Requires intent.target_agent_id to be set.
        """
        if not intent.target_agent_id:
            raise ValueError("dispatch_async() requires target_agent_id")

        # JetStream path when connected
        if self._nats_bus and self._nats_bus.connected:
            subject = f"intent.dispatch.{intent.target_agent_id}"
            try:
                await self._nats_bus.js_publish(subject, self._serialize_intent(intent))
                logger.debug(
                    "AD-654a: Dispatched %s → %s via JetStream",
                    intent.intent, intent.target_agent_id[:12],
                )
                return
            except Exception as e:
                logger.warning(
                    "AD-654a: JetStream dispatch failed for %s → %s: %s, falling back to direct",
                    intent.intent, intent.target_agent_id[:12], e,
                )
                # Fall through to direct dispatch

        # Direct-call fallback when NATS/JetStream unavailable

        # AD-654b: Try cognitive queue even when NATS is down.
        # This PRECEDES the existing create_task fallback — if the queue
        # accepts the item, return early. If no queue exists (substrate agents)
        # or enqueue is rejected (full + lower priority), fall through to
        # the create_task direct-dispatch path below.
        queue = self._get_agent_queue(intent.target_agent_id)
        if queue:
            priority = Priority.classify(
                intent=intent.intent,
                is_captain=intent.params.get("is_captain", False),
                was_mentioned=intent.params.get("was_mentioned", False),
            )
            # js_msg=None — no JetStream backing for fallback path
            if queue.enqueue(intent, priority):
                return
            # enqueue returned False — fall through to create_task

        # Existing AD-654a fallback: direct handler invocation for agents
        # without cognitive queues (substrate agents) or when queue is full.
        handler = self._subscribers.get(intent.target_agent_id)
        if handler is None:
            logger.debug("AD-654a: No handler for %s, dropping", intent.target_agent_id[:12])
            return

        # Soft cap on pending fallback tasks to prevent unbounded growth
        _MAX_PENDING_TASKS = 200
        if len(self._pending_sub_tasks) >= _MAX_PENDING_TASKS:
            logger.warning(
                "AD-654a: Pending task cap (%d) reached, dropping dispatch for %s",
                _MAX_PENDING_TASKS, intent.target_agent_id[:12],
            )
            return

        async def _run_handler() -> None:
            try:
                await handler(intent)
            except Exception:
                logger.warning(
                    "AD-654a: Direct handler failed for %s",
                    intent.target_agent_id[:12],
                    exc_info=True,
                )

        task = asyncio.get_running_loop().create_task(
            _run_handler(),
            name=f"dispatch-async-{intent.target_agent_id[:12]}",
        )
        self._pending_sub_tasks.add(task)
        task.add_done_callback(self._pending_sub_tasks.discard)

    # ── AD-654b: Cognitive queue management ─────────────────────────

    def set_record_response(self, callback: Callable[[str, str], None]) -> None:
        """AD-654b: Inject response recording callback.

        Replaces the handler.__self__._runtime.ward_room_router reach-through
        that violated Law of Demeter. Called from finalize.py with
        ward_room_router.record_agent_response.
        """
        self._record_response = callback

    def register_queue(self, agent_id: str, queue: Any) -> None:
        """Register an agent's cognitive queue (AD-654b)."""
        self._agent_queues[agent_id] = queue

    def unregister_queue(self, agent_id: str) -> None:
        """Remove an agent's cognitive queue (AD-654b)."""
        self._agent_queues.pop(agent_id, None)

    def _get_agent_queue(self, agent_id: str) -> Any | None:
        """Get the cognitive queue for an agent (AD-654b)."""
        return self._agent_queues.get(agent_id)

    def get_subscriber_map(self) -> dict[str, list[str]]:
        """Return intent_name → [agent_ids] mapping (AD-470).

        Shows which agents are indexed for which intent types.
        Agents not in any index (fallback subscribers) are listed
        under the key "__fallback__".
        """
        result: dict[str, list[str]] = {}
        all_indexed: set[str] = set()

        for intent_name, agent_ids in self._intent_index.items():
            result[intent_name] = sorted(agent_ids)
            all_indexed.update(agent_ids)

        fallback = [
            aid for aid in self._subscribers
            if aid not in all_indexed
        ]
        if fallback:
            result["__fallback__"] = sorted(fallback)

        return result

    def get_metrics(self) -> dict[str, Any]:
        """Return intent bus metrics summary (AD-470)."""
        return self._metrics.get_summary()

    def record_broadcast(self, intent_name: str) -> None:
        """Record a broadcast event with its intent name."""
        self._broadcast_timestamps.append((time.monotonic(), intent_name))

    def demand_metrics(self) -> dict:
        """Return current demand snapshot (system-wide)."""
        now = time.monotonic()
        cutoff = now - self._window_seconds
        self._broadcast_timestamps = [(t, n) for t, n in self._broadcast_timestamps if t > cutoff]
        return {
            "broadcasts_in_window": len(self._broadcast_timestamps),
            "subscriber_count": len(self._subscribers),
        }

    def per_pool_demand(self, pool_intents: dict[str, list[str]]) -> dict[str, int]:
        """Return broadcast counts per pool within the observation window.

        Args:
            pool_intents: mapping of pool_name -> list of intent names that pool handles.

        Returns:
            dict of pool_name -> number of broadcasts targeting that pool's intents.
        """
        now = time.monotonic()
        cutoff = now - self._window_seconds
        self._broadcast_timestamps = [(t, n) for t, n in self._broadcast_timestamps if t > cutoff]

        # Build reverse mapping: intent_name -> pool_name
        intent_to_pool: dict[str, str] = {}
        for pool_name, intents in pool_intents.items():
            for intent_name in intents:
                intent_to_pool[intent_name] = pool_name

        counts: dict[str, int] = {name: 0 for name in pool_intents}
        for _, intent_name in self._broadcast_timestamps:
            pool = intent_to_pool.get(intent_name)
            if pool:
                counts[pool] += 1
        return counts

    async def _invoke_handler(
        self,
        intent: IntentMessage,
        agent_id: str,
        handler: IntentHandler,
    ) -> None:
        """Invoke a single subscriber's handler, catching errors."""
        try:
            t0 = time.monotonic()
            result = await handler(intent)
            elapsed_ms = (time.monotonic() - t0) * 1000
            if elapsed_ms > 100:
                logger.warning(
                    "Slow handler: agent=%s intent=%s elapsed=%.0fms result=%s",
                    agent_id[:16], intent.intent, elapsed_ms,
                    "responded" if result else "declined",
                )
            if result is not None:
                # Agent accepted and responded
                if intent.id in self._pending_results:
                    self._pending_results[intent.id].append(result)
        except Exception as e:
            logger.warning(
                "Handler error for agent %s on intent %s: %s",
                agent_id[:8],
                intent.id[:8],
                e,
            )
            # Record the failure as a result
            if intent.id in self._pending_results:
                self._pending_results[intent.id].append(
                    IntentResult(
                        intent_id=intent.id,
                        agent_id=agent_id,
                        success=False,
                        error=str(e),
                        confidence=0.0,
                    )
                )

    # ------------------------------------------------------------------
    # AD-514: Public API
    # ------------------------------------------------------------------

    def set_federation_handler(self, fn: Callable) -> None:
        """Set the federation forwarding handler for cross-realm intents."""
        self._federation_fn = fn

    def set_nats_bus(self, nats_bus: Any) -> None:
        """Wire NATS transport (called after NATS connects in Phase 1b)."""
        self._nats_bus = nats_bus
        # AD-637z: Register for prefix change notification (logging only —
        # NATSBus handles re-subscription of all tracked subs automatically)
        nats_bus.register_on_prefix_change(self._on_prefix_change)

    async def _on_prefix_change(self, old_prefix: str, new_prefix: str) -> None:
        """Log prefix change — NATSBus has already re-subscribed all agents."""
        logger.info(
            "IntentBus: NATS prefix changed %s → %s, %d agent subs re-subscribed by NATSBus",
            old_prefix[:20], new_prefix[:20], len(self._subscribers),
        )

    # ------------------------------------------------------------------
    # BF-234: Consumer-side dispatch dedup
    # ------------------------------------------------------------------

    # Window must be ≥ JetStream ack_wait to catch duplicates queued behind
    # a slow handler. With max_ack_pending=1, msg #2 waits until msg #1 acks
    # (5–60s for a cognitive chain). 300s matches ack_wait=300 in
    # _js_subscribe_agent_dispatch. Memory: ~84KB worst case at 12 agents ×
    # 10 events/min × 600s eviction.
    _WARD_ROOM_DISPATCH_DEDUP_WINDOW: float = 300.0  # seconds — matches ack_wait

    def _is_duplicate_intent(self, intent_id: str) -> bool:
        """BF-234: Check if intent_id was already seen within the dedup window."""
        if not intent_id:
            return False
        last = self._seen_intents.get(intent_id)
        if last is not None and (time.monotonic() - last) < self._WARD_ROOM_DISPATCH_DEDUP_WINDOW:
            return True
        return False

    def _record_seen_intent(self, intent_id: str) -> None:
        """BF-234: Record that intent_id has been consumed."""
        if intent_id:
            self._seen_intents[intent_id] = time.monotonic()

    def _evict_stale_seen_intents(self, max_age: float = 600.0) -> None:
        """BF-234: Evict seen-intent records older than ``max_age`` seconds."""
        cutoff = time.monotonic() - max_age
        self._seen_intents = {
            k: v for k, v in self._seen_intents.items() if v > cutoff
        }
        self._last_seen_eviction = time.monotonic()

    def _maybe_evict_seen_intents(self, interval: float = 300.0) -> None:
        """BF-234: Periodic eviction — runs at most once per ``interval`` seconds."""
        if time.monotonic() - self._last_seen_eviction >= interval:
            self._evict_stale_seen_intents()

    def get_duplicate_suppressed_count(self) -> int:
        """BF-234: Return total number of transport-layer duplicates suppressed."""
        return self._duplicate_suppressed_count

    def set_emit_event(self, fn: Callable[[str, dict[str, Any]], None]) -> None:
        """BF-234: Inject event emitter for duplicate-suppressed telemetry."""
        self._emit_event_fn = fn

    # ------------------------------------------------------------------
    # AD-637b: Serialization helpers
    # ------------------------------------------------------------------

    @staticmethod
    def _serialize_intent(intent: IntentMessage) -> dict[str, Any]:
        """Serialize IntentMessage for NATS transport.

        All fields must be JSON-serializable. params dict values that are
        not JSON-serializable will raise TypeError — fail fast.

        AD-731 (2026-05-11): BF-265's transport strip removed. Vision
        attachments are now carried as content-addressable refs (~70 bytes
        per image) instead of inline base64. The receiver dereferences from
        the local AttachmentStore inside the LLM client. The bus carries
        only refs; payload size is bounded; the uniform-NATS-transport
        invariant is restored. AD-637z2 closes as part of this AD.
        """
        return {
            "intent": intent.intent,
            "params": intent.params,
            "urgency": intent.urgency,
            "context": intent.context,
            "ttl_seconds": intent.ttl_seconds,
            "id": intent.id,
            "created_at": intent.created_at.isoformat(),
            "target_agent_id": intent.target_agent_id,
        }

    @staticmethod
    def _deserialize_intent(data: dict[str, Any]) -> IntentMessage:
        """Deserialize IntentMessage from NATS transport."""
        created_at = data.get("created_at")
        if isinstance(created_at, str):
            created_at = datetime.fromisoformat(created_at)
        else:
            created_at = datetime.now(timezone.utc)
        return IntentMessage(
            intent=data["intent"],
            params=data.get("params", {}),
            urgency=data.get("urgency", 0.5),
            context=data.get("context", ""),
            ttl_seconds=data.get("ttl_seconds", 60.0),
            id=data.get("id", ""),
            created_at=created_at,
            target_agent_id=data.get("target_agent_id"),
        )

    @staticmethod
    def _serialize_result(result: IntentResult) -> dict[str, Any]:
        """Serialize IntentResult for NATS reply.

        result.result must be JSON-serializable. Non-serializable values
        will raise TypeError — this is intentional (fail fast). Handlers
        using the NATS path must return serializable results.
        """
        return {
            "intent_id": result.intent_id,
            "agent_id": result.agent_id,
            "success": result.success,
            "result": result.result,
            "error": result.error,
            "confidence": result.confidence,
            "timestamp": result.timestamp.isoformat(),
        }

    @staticmethod
    def _deserialize_result(data: dict[str, Any]) -> IntentResult:
        """Deserialize IntentResult from NATS reply."""
        ts = data.get("timestamp")
        if isinstance(ts, str):
            ts = datetime.fromisoformat(ts)
        else:
            ts = datetime.now(timezone.utc)
        return IntentResult(
            intent_id=data.get("intent_id", ""),
            agent_id=data.get("agent_id", ""),
            success=data.get("success", False),
            result=data.get("result"),
            error=data.get("error"),
            confidence=data.get("confidence", 0.0),
            timestamp=ts,
        )
