"""Proactive Cognitive Loop — periodic idle-think for crew agents (Phase 28b)."""

from __future__ import annotations

import asyncio
import json
import hashlib
import logging
import re
import time
from dataclasses import dataclass
from datetime import datetime
from pathlib import Path
from typing import TYPE_CHECKING, Any, Callable

# BF-174: Pattern to strip self-monitoring bracket markers that LLMs parrot back.
# Matches [observed], [observed clinical consensus], [surgical validation], etc.
_BRACKET_MARKER_RE = re.compile(
    r'\s*\[(?:observed|analyzed|verified|confirmed|validated|noted|monitored|detected'
    r'|assessed|reviewed|investigated|evaluated|checked|inspected|measured|recorded'
    r'|tracked|surveyed|documented|reported|logged|flagged|identified|resolved'
    r')[^\]]{0,80}\]\s*',
    re.IGNORECASE,
)


def _strip_bracket_markers(text: str) -> str:
    """BF-174: Remove self-monitoring bracket markers from agent output."""
    return _BRACKET_MARKER_RE.sub(' ', text).strip()

from probos.config import format_trust
from probos.crew_profile import Rank
from probos.crew_utils import is_crew_agent
from probos.cognitive.cognitive_agent import CognitiveAgent
from probos.duty_schedule import DutySchedule
from probos.events import EventType
from probos.duty_schedule import DutyScheduleTracker
from probos.earned_agency import AgencyLevel, agency_from_rank, can_think_proactively
from probos.cognitive.circuit_breaker import CognitiveCircuitBreaker
from probos.cognitive.dm_sanity_gate import apply_dm_sanity
from probos.cognitive.orientation import OrientationContext, derive_watch_section
from probos.types import AnchorFrame, IntentMessage, Priority
from probos.utils import format_duration

if TYPE_CHECKING:
    from probos.cognitive.novelty_gate import NoveltyGate
    from probos.config import DutyScheduleConfig, ProactiveCognitiveConfig, WardRoomConfig
    from probos.knowledge.store import KnowledgeStore
    from probos.runtime import ProbOSRuntime
    from probos.ward_room_pipeline import PostBudget

logger = logging.getLogger(__name__)


# AD-635f: Per-domain recency caps for clinical telemetry context injection.
# Module-level constants (NOT config) — signal-density tuning for LLM context.
_DREAM_RECENT_LIMIT = 5
_CHAIN_TRACE_RECENT_LIMIT = 5
_BREAKER_RECENT_LIMIT = 5


def _parse_hhmm(value: str) -> tuple[int, int]:
    """Parse HH:MM into (hour, minute) for schedule threshold checks."""
    parts = value.split(":", 1)
    hour = int(parts[0])
    minute = int(parts[1]) if len(parts) > 1 else 0
    return hour, minute


def collect_notebook_metrics(runtime: Any, agent_id: str = "") -> dict[str, Any]:
    """AD-553: Collect standardized metrics snapshot for notebook attachment.

    Returns flat dict of metric_name -> value. Degrades gracefully:
    returns empty dict if runtime/VitalsMonitor unavailable.
    """
    metrics: dict[str, Any] = {}
    if runtime is None:
        return metrics

    # VitalsMonitor cached data (no I/O)
    vitals = None
    try:
        for agent in runtime.registry.all():
            if getattr(agent, "agent_type", "") == "vitals_monitor":
                vitals = agent.latest_vitals
                break
    except Exception:
        pass  # Registry unavailable

    if vitals:
        for key in ("trust_mean", "trust_min", "system_health"):
            val = vitals.get(key)
            if val is not None:
                metrics[key] = round(val, 3)

        # Pool health mean
        pool_health = vitals.get("pool_health")
        if pool_health and isinstance(pool_health, dict):
            vals = [v for v in pool_health.values() if isinstance(v, (int, float))]
            if vals:
                metrics["pool_health_mean"] = round(sum(vals) / len(vals), 3)

        # Emergence (AD-557)
        for key in ("emergence_capacity", "coordination_balance"):
            val = vitals.get(key)
            if val is not None:
                metrics[key] = round(val, 3)

        # LLM health
        llm = vitals.get("llm_health")
        if isinstance(llm, dict):
            overall = llm.get("overall")
            if overall:
                metrics["llm_health"] = overall

    # Agent's own trust score
    if agent_id and hasattr(runtime, "trust_network") and runtime.trust_network:
        try:
            score = runtime.trust_network.get_score(agent_id)
            if score is not None:
                metrics["agent_trust"] = round(score, 3)
        except Exception:
            pass

    # Active agent count
    try:
        metrics["active_agents"] = len(runtime.registry.all())
    except Exception:
        pass

    return metrics


def compute_metrics_delta(
    old_metrics: dict[str, Any],
    new_metrics: dict[str, Any],
    *,
    min_numeric_delta: float = 0.01,
) -> dict[str, Any]:
    """AD-553: Compute delta between two metrics snapshots.

    Returns dict of metric_name -> delta for numeric values that changed
    by more than min_numeric_delta, and "old -> new" strings for changed
    string values. Returns empty dict if no meaningful changes.
    """
    delta: dict[str, Any] = {}
    all_keys = set(old_metrics) | set(new_metrics)

    for key in sorted(all_keys):
        old_val = old_metrics.get(key)
        new_val = new_metrics.get(key)

        if old_val is None or new_val is None:
            continue  # Skip if either side is missing

        if isinstance(old_val, (int, float)) and isinstance(new_val, (int, float)):
            diff = new_val - old_val
            if abs(diff) >= min_numeric_delta:
                delta[key] = round(diff, 3)
        elif isinstance(old_val, str) and isinstance(new_val, str):
            if old_val != new_val:
                delta[key] = f"{old_val} \u2192 {new_val}"

    return delta


@dataclass
class ProactiveStatusSnapshot:
    """Runtime-facing proactive status projection for API/UI surfaces."""

    next_inbox_scan: str
    next_calendar_scan: str
    work_hours_active: bool
    quiet_hours_active: bool
    last_scan_count: dict[str, int]


_PROACTIVE_SCAN_INSTRUCTIONS = (
    "You are a policy-governed proactive scan coordinator. "
    "Your role is to evaluate scan eligibility and emit heartbeat-tagged scan intents."
)


class ProactiveScanAgent(CognitiveAgent):
    """Periodic policy-gated scan selector for inbox/calendar/teams."""

    agent_type = "proactive_scan"
    tier = "utility"
    instructions = _PROACTIVE_SCAN_INSTRUCTIONS

    def __init__(self, **kwargs: Any) -> None:
        kwargs.setdefault("pool", "operations_scheduler")
        super().__init__(**kwargs)

    async def perceive(self, intent: dict[str, Any] | None = None) -> IntentMessage:
        """Build a heartbeat-tagged proactive_scan intent with allowed scan types."""
        schedule: DutySchedule | None = getattr(self._runtime, "duty_schedule", None)
        scan_types = ["inbox", "calendar", "teams"]
        if schedule is None:
            return IntentMessage(
                intent="proactive_scan",
                params={
                    "scan_types": [],
                    "tagged_as": "heartbeat",
                    "suppressed_reasons": {kind: "policy_unavailable" for kind in scan_types},
                },
            )

        now = datetime.now()
        allowed: list[str] = []
        suppressed: dict[str, str] = {}
        for scan_type in scan_types:
            if schedule.should_scan(scan_type, now):
                allowed.append(scan_type)
            else:
                suppressed[scan_type] = schedule.reason_code(scan_type, now)

        return IntentMessage(
            intent="proactive_scan",
            params={
                "scan_types": allowed,
                "tagged_as": "heartbeat",
                "suppressed_reasons": suppressed,
            },
        )


class DailyBriefingScheduler:
    """Daily briefing trigger with once-per-day persistence and dismiss support."""

    def __init__(
        self,
        *,
        duty_schedule: DutySchedule,
        runtime: Any | None = None,
        state_path: Path | None = None,
        now_fn: Callable[[], datetime] | None = None,
    ) -> None:
        self._duty_schedule = duty_schedule
        self._runtime = runtime
        self._state_path = state_path or Path("data/briefing_state.json")
        self._now_fn = now_fn or datetime.now

    async def trigger_briefing_if_time(self) -> bool:
        """Trigger a briefing once per day after briefing-time threshold."""
        now = self._now_fn()
        if not self._duty_schedule.work_hours.is_active(now):
            return False

        state = self._load_state()
        day_key = now.date().isoformat()
        if day_key in state.get("dismissed_days", []):
            return False
        if state.get("last_triggered_day") == day_key:
            return False
        if not self._is_briefing_time_reached(now):
            return False

        last_reminder_ts = float(state.get("last_reminder_ts", 0.0))
        reminder_gap = now.timestamp() - last_reminder_ts
        if reminder_gap < float(self._duty_schedule.briefing_reminder_throttle_sec):
            return False

        state["last_triggered_day"] = day_key
        state["last_reminder_ts"] = now.timestamp()
        self._save_state(state)

        rt = self._runtime
        if rt is not None and hasattr(rt, "emit_event"):
            try:
                rt.emit_event(
                    EventType.TASK_SCHEDULED,
                    {
                        "task_kind": "daily_briefing",
                        "scheduled_at": now.timestamp(),
                        "source": "ad-752",
                    },
                )
            except Exception:
                logger.warning(
                    "AD-752: daily briefing emit failed; briefing state persisted and will retry naturally",
                    exc_info=True,
                )
        return True

    def dismiss_for_today(self) -> None:
        """Persist explicit captain dismissal for the current local day."""
        now = self._now_fn()
        state = self._load_state()
        day_key = now.date().isoformat()
        dismissed_days = list(state.get("dismissed_days", []))
        if day_key not in dismissed_days:
            dismissed_days.append(day_key)
        state["dismissed_days"] = dismissed_days[-30:]
        self._save_state(state)

    def _is_briefing_time_reached(self, dt: datetime) -> bool:
        hour, minute = _parse_hhmm(self._duty_schedule.daily_briefing_time)
        return (dt.hour, dt.minute) >= (hour, minute)

    def _load_state(self) -> dict[str, Any]:
        if not self._state_path.exists():
            return {}
        try:
            return json.loads(self._state_path.read_text(encoding="utf-8"))
        except Exception:
            logger.warning(
                "AD-752: briefing state unreadable at %s; resetting state to empty",
                self._state_path,
                exc_info=True,
            )
            return {}

    def _save_state(self, state: dict[str, Any]) -> None:
        self._state_path.parent.mkdir(parents=True, exist_ok=True)
        self._state_path.write_text(
            json.dumps(state, ensure_ascii=True, sort_keys=True, indent=2),
            encoding="utf-8",
        )


def build_proactive_status_snapshot(runtime: Any) -> ProactiveStatusSnapshot:
    """Build status payload for Ward Room proactive visibility endpoint."""
    duty_schedule: DutySchedule | None = getattr(runtime, "duty_schedule", None)
    now = datetime.now()
    default_counts = {"inbox": 0, "calendar": 0, "teams": 0}
    counts = dict(getattr(runtime, "proactive_last_scan_count", default_counts) or default_counts)

    if duty_schedule is None:
        return ProactiveStatusSnapshot(
            next_inbox_scan=now.isoformat(),
            next_calendar_scan=now.isoformat(),
            work_hours_active=False,
            quiet_hours_active=False,
            last_scan_count=counts,
        )

    next_inbox = duty_schedule.next_scan_window("inbox", now)
    next_calendar = duty_schedule.next_scan_window("calendar", now)
    return ProactiveStatusSnapshot(
        next_inbox_scan=next_inbox.isoformat(),
        next_calendar_scan=next_calendar.isoformat(),
        work_hours_active=duty_schedule.work_hours.is_active(now),
        quiet_hours_active=duty_schedule.quiet_hours.is_active(now),
        last_scan_count=counts,
    )


class ProactiveCognitiveLoop:
    """Periodic idle-think cycle for crew agents.

    Every ``interval`` seconds, iterates crew agents sequentially.
    For each agent with sufficient trust (Lieutenant+), gathers recent
    context (episodic memory, bridge alerts, system events) and sends
    a ``proactive_think`` intent. If the agent's LLM produces a meaningful
    response (not ``[NO_RESPONSE]``), creates a Ward Room thread in the
    agent's department channel.

    Follows the InitiativeEngine pattern: asyncio.create_task, fail-open,
    CancelledError propagation.
    """

    # BF-039: Cold-start episode dampening — 3x cooldown for this window
    COLD_START_WINDOW_SECONDS = 600  # 10 minutes

    def __init__(
        self,
        *,
        interval: float = 120.0,
        cooldown: float = 300.0,
        on_event: Callable[[dict], Any] | None = None,
    ) -> None:
        self._interval = interval
        self._cooldown = cooldown
        self._on_event = on_event
        self._last_proactive: dict[str, float] = {}  # agent_id -> monotonic timestamp
        self._agent_cooldowns: dict[str, float] = {}  # agent_id -> override cooldown (seconds)
        self._cooldown_reasons: dict[str, str] = {}  # agent_id -> reason text (AD-505)
        self._knowledge_store: KnowledgeStore | None = None  # AD-415: Set by runtime for persistence
        self._task: asyncio.Task | None = None
        self._runtime: ProbOSRuntime | None = None  # Set via set_runtime()
        self._started_at: float = time.monotonic()  # BF-039: cold-start reference
        self._config: ProactiveCognitiveConfig | None = None   # Set via set_config()
        self._duty_tracker: DutyScheduleTracker | None = None
        self._circuit_breaker = CognitiveCircuitBreaker()
        self._budget_exhausted: dict[str, float] = {}  # AD-617b: agent_id -> exhaustion timestamp
        self._notified_dm_threads: set[str] = set()  # BF-082: dedup guard
        self._dm_send_cooldowns: dict[str, float] = {}  # BF-163: per-agent DM send rate limit
        self._reply_cooldowns: dict[str, float] = {}  # BF-171: per-agent per-channel reply cooldown
        self._last_dm_body: dict[str, str] = {}  # AD-614: self-similarity gate
        self._notified_dm_threads_reset: float = time.monotonic()  # hourly reset
        self._dm_response_counts: dict[str, list[float]] = {}  # BF-257: agent_id -> [timestamps]
        self._dm_pair_counts: dict[str, list[float]] = {}      # BF-257: "a_id:b_id" -> [timestamps]
        self._pending_notebook_reads: dict[str, str] = {}  # AD-504: agent_id -> topic_slug
        self._llm_failure_count: int = 0  # BF-069: consecutive proactive failures
        self._llm_status: str = "operational"  # AD-576: "operational" | "degraded" | "offline"
        self._llm_offline_since: float = 0.0   # AD-576: monotonic timestamp of first failure
        self._orientation_service: Any = None  # AD-567g: Late-bound
        self._novelty_gate: "NoveltyGate | None" = None  # AD-493: Set via set_novelty_gate()
        self._trait_adaptive_enabled: bool = True  # AD-494: trait-adaptive circuit breaker
        self._qualification_config = None  # AD-595e: late-bound
        self._billet_registry = None         # AD-595e: late-bound
        # AD-572d-i: interruptible-wait — replaces bare ``asyncio.sleep`` in
        # ``_think_loop`` so external triggers (DM arrival, Captain message,
        # high-priority intent) can wake the loop early. ``_wakeup`` is set
        # by callers via ``trigger_wakeup()``; the think loop awaits
        # ``wait_for(_wakeup.wait(), interval)`` and clears the event each cycle.
        self._wakeup: asyncio.Event | None = None

    def set_orientation_service(self, svc: Any) -> None:
        """AD-567g / BF-113: Set orientation service (public setter for LoD)."""
        self._orientation_service = svc

    def set_novelty_gate(self, gate: "NoveltyGate") -> None:
        """AD-493: Set novelty gate (public setter for LoD)."""
        self._novelty_gate = gate

    def set_qualification_config(self, config: Any) -> None:
        """AD-595e: Set qualification enforcement config."""
        self._qualification_config = config

    def set_billet_registry(self, registry: Any) -> None:
        """AD-595e: Set billet registry for qualification checks."""
        self._billet_registry = registry

    def _ensure_agent_traits_registered(self, agent: Any) -> None:
        """AD-494: Lazily register personality traits with circuit breaker.

        Called once per agent on first proactive think. Loads seed profile
        and extracts Big Five scores. Log-and-degrade on any failure.
        """
        if not getattr(self, '_trait_adaptive_enabled', True):
            return
        agent_id = agent.id
        if self._circuit_breaker.has_agent_traits(agent_id):
            return
        try:
            from probos.crew_profile import PersonalityTraits, load_seed_profile
            profile = load_seed_profile(agent.agent_type)
            if profile and "personality" in profile:
                traits = PersonalityTraits.from_dict(profile["personality"])
                self._circuit_breaker.set_agent_traits(
                    agent_id,
                    openness=traits.openness,
                    conscientiousness=traits.conscientiousness,
                    extraversion=traits.extraversion,
                    agreeableness=traits.agreeableness,
                    neuroticism=traits.neuroticism,
                )
                logger.debug(
                    "AD-494: Registered trait thresholds for %s (%s)",
                    getattr(agent, 'callsign', agent.agent_type),
                    agent.agent_type,
                )
        except Exception:
            logger.debug("AD-494: Failed to load traits for %s, using defaults", agent_id, exc_info=True)

    async def _check_duty_qualification(self, agent: Any, duty: Any) -> bool:
        """AD-595e: Qualification gate for proactive duty dispatch.

        Returns True if agent is qualified (or gate is disabled/degraded),
        False if blocked. Shadow mode logs but returns True.
        """
        if not self._qualification_config:
            return True
        if not getattr(self._qualification_config, 'enforcement_enabled', False):
            return True
        if not self._billet_registry:
            return True

        try:
            standing = await self._billet_registry.get_qualification_standing(
                agent.agent_type, agent_id=agent.id,
            )
            if standing.get("qualified", True):
                return True

            log_only = getattr(self._qualification_config, 'enforcement_log_only', True)
            payload = {
                "gate": "duty_dispatch",
                "agent_id": agent.id,
                "agent_type": agent.agent_type,
                "callsign": getattr(agent, 'callsign', agent.agent_type),
                "duty_id": getattr(duty, 'duty_id', ''),
                "missing": standing.get("missing", []),
                "log_only": log_only,
            }
            if self._on_event:
                self._on_event({
                    "type": EventType.QUALIFICATION_GATE_BLOCKED.value,
                    "data": payload,
                })

            if log_only:
                logger.info(
                    "AD-595e: Duty gate (shadow): %s missing %s for duty %s",
                    agent.agent_type, standing.get("missing", []),
                    getattr(duty, 'duty_id', ''),
                )
                return True  # Shadow — allow
            else:
                logger.warning(
                    "AD-595e: Duty gate BLOCKED: %s missing %s for duty %s",
                    agent.agent_type, standing.get("missing", []),
                    getattr(duty, 'duty_id', ''),
                )
                return False  # Blocked
        except Exception:
            logger.debug("AD-595e: Duty qualification gate error — allowing", exc_info=True)
            return True

    def set_runtime(self, runtime: ProbOSRuntime) -> None:
        """Wire the runtime reference (provides registry, trust, WR, memory, etc.)."""
        self._runtime = runtime

    def set_config(self, config: ProactiveCognitiveConfig, cb_config: Any = None, trait_config: Any = None) -> None:
        """Store ProactiveCognitiveConfig for trust signal weights (AD-414)."""
        self._config = config
        if cb_config:
            from probos.cognitive.circuit_breaker import CognitiveCircuitBreaker
            self._circuit_breaker = CognitiveCircuitBreaker(config=cb_config)
        # AD-494: Store trait-adaptive enabled flag
        if trait_config is not None:
            self._trait_adaptive_enabled = getattr(trait_config, 'enabled', True)

    @property
    def circuit_breaker(self) -> CognitiveCircuitBreaker:
        """Expose circuit breaker for runtime wiring and API access."""
        return self._circuit_breaker

    async def _is_over_token_budget(self, agent_id: str) -> bool:
        """AD-617b: Check if agent has exceeded hourly token budget.

        Returns True if over budget, False otherwise.
        Fail-closed: if the journal query fails, returns False (allow the think).
        Uses a 60-second cache per agent to avoid hammering the journal DB.
        """
        rt = self._runtime
        if not rt:
            return False

        # Check config
        rate_config = getattr(getattr(rt, 'config', None), 'llm_rate', None)
        if not rate_config:
            return False
        hourly_cap = getattr(rate_config, 'per_agent_hourly_token_cap', 0)
        if hourly_cap <= 0:
            return False  # Disabled

        # Cache check: don't query journal every proactive cycle
        last_exhaustion = self._budget_exhausted.get(agent_id, 0.0)
        if last_exhaustion > 0 and time.monotonic() - last_exhaustion < 60.0:
            return True  # Still in exhaustion window

        # Query journal for hourly usage
        journal = getattr(rt, 'cognitive_journal', None)
        if not journal:
            return False

        since = time.time() - 3600.0  # 1 hour sliding window
        tokens_used = await journal.get_token_usage_since(agent_id, since)

        if tokens_used >= hourly_cap:
            self._budget_exhausted[agent_id] = time.monotonic()
            logger.info(
                "AD-617b: %s over token budget (%d/%d tokens in last hour)",
                agent_id[:8], tokens_used, hourly_cap,
            )
            # Emit event for Counselor awareness (fire-and-forget)
            if self._on_event:
                try:
                    await self._on_event({
                        "type": "token_budget_exhausted",
                        "agent_id": agent_id,
                        "tokens_used": tokens_used,
                        "hourly_cap": hourly_cap,
                        "data": {
                            "agent_id": agent_id,
                            "tokens_used": tokens_used,
                            "hourly_cap": hourly_cap,
                        },
                    })
                except Exception:
                    logger.debug("Budget exhaustion event emission failed", exc_info=True)
            return True

        # Clear exhaustion if recovered (new hour, tokens aged out)
        if agent_id in self._budget_exhausted:
            del self._budget_exhausted[agent_id]
        return False

    @property
    def _default_cooldown(self) -> float:
        return self._cooldown

    def set_cycle_interval(self, seconds: float) -> None:
        """AD-468: public setter for the proactive cycle interval (clamped 10-3600s)."""
        self._interval = max(10.0, min(3600.0, float(seconds)))

    def set_cooldown(self, seconds: float) -> None:
        """AD-468: public setter for the global proactive cooldown default (clamped 60-86400s)."""
        self._cooldown = max(60.0, min(86400.0, float(seconds)))

    def set_duty_schedule(self, config: DutyScheduleConfig | None) -> None:
        """Initialize duty schedule tracker from DutyScheduleConfig."""
        if config and config.enabled and config.schedules:
            self._duty_tracker = DutyScheduleTracker(config.schedules)
            logger.info("Duty schedule loaded: %d agent types configured",
                         len(config.schedules))
        else:
            self._duty_tracker = None

    def get_agent_cooldown(self, agent_id: str) -> float:
        """Get effective cooldown for an agent (override or global default)."""
        return self._agent_cooldowns.get(agent_id, self._cooldown)

    def get_cooldown_reason(self, agent_id: str) -> str:
        """Return the reason for a cooldown override, or empty string."""
        return self._cooldown_reasons.get(agent_id, "")

    def clear_counselor_cooldown(self, agent_id: str) -> None:
        """Remove a Counselor-set cooldown override, restoring the default."""
        if agent_id in self._agent_cooldowns:
            del self._agent_cooldowns[agent_id]
        if agent_id in self._cooldown_reasons:
            del self._cooldown_reasons[agent_id]
        self._persist_cooldowns()

    def set_agent_cooldown(self, agent_id: str, cooldown: float, reason: str = "") -> None:
        """Set per-agent proactive cooldown override. Clamp to [60, 1800]."""
        cooldown = max(60.0, min(1800.0, cooldown))
        self._agent_cooldowns[agent_id] = cooldown
        if reason:
            self._cooldown_reasons[agent_id] = reason
        elif agent_id in self._cooldown_reasons:
            del self._cooldown_reasons[agent_id]
        # AD-415: Write-through to KnowledgeStore
        self._persist_cooldowns()

    def _persist_cooldowns(self) -> None:
        """AD-415: Fire-and-forget persistence of cooldown overrides."""
        if not self._knowledge_store:
            return
        try:
            loop = asyncio.get_running_loop()
            loop.create_task(self._knowledge_store.store_cooldowns(self._agent_cooldowns.copy()))
        except RuntimeError:
            pass  # No event loop — skip persistence (e.g., during shutdown)

    async def restore_cooldowns(self) -> None:
        """AD-415: Restore per-agent cooldowns from KnowledgeStore on boot."""
        if not self._knowledge_store:
            return
        try:
            saved = await self._knowledge_store.load_cooldowns()
            if saved:
                for agent_id, cooldown in saved.items():
                    # Apply same clamping as set_agent_cooldown
                    self._agent_cooldowns[agent_id] = max(60.0, min(1800.0, cooldown))
        except Exception:
            logger.info("Failed to restore proactive cooldowns — agents may be temporarily over-active", exc_info=True)

    async def start(self) -> None:
        """Start the periodic think loop."""
        if self._task is not None:
            return
        self._task = asyncio.create_task(self._think_loop())

    async def stop(self) -> None:
        """Stop the think loop."""
        if self._task:
            self._task.cancel()
            try:
                await self._task
            except asyncio.CancelledError:
                pass
            self._task = None

    async def _think_loop(self) -> None:
        """Main loop: iterate agents every interval seconds.

        AD-572d-i: replaces ``asyncio.sleep`` with ``_interruptible_sleep``
        so callers can wake the loop early via ``trigger_wakeup()``.
        """
        while True:
            try:
                await self._run_cycle()
                await self._interruptible_sleep(self._interval)
            except asyncio.CancelledError:
                raise
            except Exception:
                logger.exception("ProactiveCognitiveLoop cycle failed (fail-open)")
                # BF-211: Sleep inside try/except so the loop survives
                # non-cancellation errors during sleep.
                await self._interruptible_sleep(self._interval)

    async def _interruptible_sleep(self, seconds: float) -> None:
        """AD-572d-i: sleep that returns early when ``trigger_wakeup`` fires.

        Lazily initializes the wakeup event on first call (must happen on
        the running loop, not in __init__). Clears the event after each
        wake so the next cycle starts fresh.
        """
        # AD-572d-i: defensive getattr — some tests construct via __new__
        # bypassing __init__, so the attribute may not exist yet.
        wakeup = getattr(self, "_wakeup", None)
        if wakeup is None:
            wakeup = asyncio.Event()
            self._wakeup = wakeup
        try:
            await asyncio.wait_for(wakeup.wait(), timeout=seconds)
            # Wakeup fired — clear so the next cycle starts blank.
            wakeup.clear()
        except asyncio.TimeoutError:
            # Normal path: full interval elapsed.
            pass

    def trigger_wakeup(self) -> None:
        """AD-572d-i: signal the think loop to wake from its current sleep.

        Safe to call from any thread/coroutine; idempotent within a
        single sleep window.
        """
        wakeup = getattr(self, "_wakeup", None)
        if wakeup is not None:
            wakeup.set()

    async def _run_cycle(self) -> None:
        """One think cycle: iterate all crew agents with AD-636 stagger."""
        rt = self._runtime
        if not rt or not rt.ward_room:
            return

        # AD-636: Count eligible agents for stagger calculation
        eligible_agents: list[Any] = []
        for agent in rt.registry.all():
            if not is_crew_agent(agent, rt.ontology):
                continue
            if not agent.is_alive:
                continue
            # AD-486: Skip agents still in Birth Chamber (pre-graduation).
            # Default-disabled chamber (no runtime.birth_chamber) treats
            # all agents as graduated → zero behavior change.
            chamber = getattr(rt, "birth_chamber", None)
            if chamber is not None and not chamber.is_graduated(agent.id):
                continue
            eligible_agents.append(agent)

        # AD-636: Calculate stagger delay
        stagger_delay = 0.0
        _proactive_config = getattr(getattr(rt, 'config', None), 'proactive_cognitive', None)
        _stagger_enabled = getattr(_proactive_config, 'stagger_enabled', True) if _proactive_config else True
        _min_stagger = getattr(_proactive_config, 'min_stagger_seconds', 5.0) if _proactive_config else 5.0
        if _stagger_enabled and len(eligible_agents) > 1:
            stagger_delay = max(_min_stagger, self._interval / len(eligible_agents))
            # Don't let total stagger time exceed interval
            if stagger_delay * len(eligible_agents) > self._interval:
                stagger_delay = self._interval / len(eligible_agents)

        agents_dispatched = 0
        for agent in eligible_agents:
            # BF-156: Unread DM check BEFORE agency gating.
            # DM delivery is communication reliability, not proactive agency.
            # Ensigns should still receive their DMs.
            await self._check_unread_dms(agent, rt)

            # Agency gating: Ensigns don't think proactively
            trust_score = rt.trust_network.get_score(agent.id)
            rank = Rank.from_trust(trust_score)
            if not can_think_proactively(rank):
                continue

            # Cooldown: skip if agent posted proactively recently
            last = self._last_proactive.get(agent.id, 0.0)
            cooldown = self.get_agent_cooldown(agent.id)
            # BF-039: Cold-start dampening — 3x cooldown for first N minutes
            if time.monotonic() - self._started_at < self.COLD_START_WINDOW_SECONDS:
                cooldown *= 3
            if time.monotonic() - last < cooldown:
                continue

            # AD-494: Ensure agent personality traits are registered with circuit breaker
            self._ensure_agent_traits_registered(agent)

            # AD-488: Circuit breaker gate — skip agents in cognitive cooldown
            if not self._circuit_breaker.should_allow_think(agent.id):
                breaker_status = self._circuit_breaker.get_status(agent.id)
                logger.debug(
                    "AD-488: %s circuit breaker OPEN (trip #%d, cooldown %.0fs remaining)",
                    getattr(agent, 'callsign', agent.agent_type),
                    breaker_status['trip_count'],
                    breaker_status['cooldown_seconds'] - (time.monotonic() - breaker_status['tripped_at']),
                )
                continue

            # AD-617b: Per-agent token budget gate — skip proactive if over budget
            if await self._is_over_token_budget(agent.id):
                logger.debug(
                    "AD-617b: %s proactive think skipped (token budget exhausted)",
                    getattr(agent, 'callsign', agent.agent_type),
                )
                continue

            # AD-442: Check probationary → active transition
            if rt.acm and rt.trust_network:
                try:
                    _onb = getattr(getattr(rt, 'config', None), 'onboarding', None)
                    _act_thresh = float(getattr(_onb, 'activation_trust_threshold', 0.65)) if _onb else 0.65
                    _tscore = rt.trust_network.get_score(agent.id)
                    if isinstance(_tscore, (int, float)) and _tscore >= _act_thresh:
                        activated = await rt.acm.check_activation(agent.id, _tscore, _act_thresh)
                        if activated:
                            logger.info("AD-442: %s activated (trust=%.2f)", getattr(agent, 'callsign', agent.id), _tscore)
                except Exception:
                    logger.debug("ACM activation check failed for agent", exc_info=True)

            try:
                await self._think_for_agent(agent, rank, trust_score)
            except Exception:
                # BF-023: Track the failure so confidence reflects reality.
                # Without this, LLM exceptions leave confidence frozen and
                # agents stay permanently degraded with no recovery path.
                agent.update_confidence(False)
                logger.debug(
                    "Proactive think failed for %s (confidence=%.3f)",
                    agent.agent_type,
                    agent.confidence,
                    exc_info=True,
                )

            # AD-636: Stagger delay between agents to distribute LLM load
            agents_dispatched += 1
            if stagger_delay > 0 and agents_dispatched < len(eligible_agents):
                await asyncio.sleep(stagger_delay)

    async def _check_unread_dms(self, agent: Any, rt: Any) -> None:
        """Check for and route unread DMs for an agent."""
        if not rt.ward_room or not rt.ward_room_router:
            return

        # Hourly reset of dedup set to allow re-notification of stale DMs
        if time.monotonic() - self._notified_dm_threads_reset > 3600:
            self._notified_dm_threads.clear()
            self._notified_dm_threads_reset = time.monotonic()

        try:
            # BF-164: pass exchange limit so query excludes capped threads
            dm_limit = getattr(rt.config.ward_room, 'dm_exchange_limit', 6)
            unread_dms = await rt.ward_room.get_unread_dms(
                agent.id, limit=2, exchange_limit=dm_limit,
            )
            if not unread_dms:
                return

            routed = 0
            for dm in unread_dms:
                tid = dm["thread_id"]
                if tid in self._notified_dm_threads:
                    continue

                # BF-257: DM response budget check -- skip Captain DMs.
                # v1: callsign-based check (case-insensitive). The Captain's
                # callsign is canonical at "Captain" per AD-499 ShipNamingPolicy
                # and BF-244 ontology callsign sync. If a future AD introduces a
                # canonical captain DID or `is_captain(rt, author_id)` helper,
                # this check should switch to identity-based.
                author_callsign = (dm.get("author_callsign") or "").strip().lower()
                if author_callsign != "captain":
                    wr_config = getattr(
                        getattr(rt, 'config', None), 'ward_room', None,
                    )
                    if wr_config:
                        reason = self._dm_response_budget_exceeded(
                            agent.id, dm["author_id"], wr_config,
                        )
                        if reason:
                            logger.info(
                                "BF-257: %s DM response to @%s throttled (%s)",
                                getattr(agent, 'callsign', agent.agent_type),
                                dm.get("author_callsign", dm["author_id"]),
                                reason,
                            )
                            # Don't add to _notified_dm_threads -- allow retry
                            # after window expires.
                            continue

                self._notified_dm_threads.add(tid)

                # BF-257: Record this DM response in budget trackers.
                now = time.monotonic()
                self._dm_response_counts.setdefault(agent.id, []).append(now)
                # Canonical pair key -- A->B and B->A share the same counter.
                pair_key = ":".join(sorted([agent.id, dm["author_id"]]))
                self._dm_pair_counts.setdefault(pair_key, []).append(now)

                # Route through existing notification pipeline
                event_data = {
                    "thread_id": tid,
                    "channel_id": dm["channel_id"],
                    "author_id": dm["author_id"],
                    "author_callsign": dm["author_callsign"],
                    "title": dm["title"],
                    "body": dm["body"],
                }
                # AD-575c: flag DMs whose forwarded body mentions this agent
                # (e.g. "Captain said about you: @<callsign>"). Read-only
                # check; preserves upstream content.
                agent_callsign = (getattr(agent, "callsign", "") or "").strip()
                if agent_callsign:
                    body_text = dm.get("body") or ""
                    needle = f"@{agent_callsign}".lower()
                    if needle in body_text.lower():
                        event_data["self_referenced"] = True
                await rt.ward_room_router.route_event(
                    "ward_room_thread_created", event_data,
                )
                routed += 1
            if routed:
                logger.info(
                    "BF-082: %s has %d unread DMs, notified",
                    getattr(agent, 'callsign', agent.agent_type),
                    routed,
                )
        except Exception as exc:
            logger.warning(
                "BF-082: Unread DM check failed for %s: %s",
                agent.agent_type, exc,
            )

    def _dm_response_budget_exceeded(
        self, agent_id: str, partner_id: str, config: "WardRoomConfig",
    ) -> str | None:
        """BF-257: Check if agent has exceeded DM response budget.

        Returns a reason string if budget is exceeded, None if allowed.
        Uses a sliding window -- expired timestamps are pruned on each call.
        Synchronous: pure in-memory check (no I/O).
        """
        now = time.monotonic()
        window = config.dm_response_window_seconds
        cutoff = now - window

        # Layer 1: Per-agent global DM response budget
        budget = config.dm_response_budget
        agent_times = self._dm_response_counts.get(agent_id, [])
        agent_times = [t for t in agent_times if t > cutoff]
        self._dm_response_counts[agent_id] = agent_times
        if len(agent_times) >= budget:
            return f"agent_budget ({len(agent_times)}/{budget} in {window:.0f}s)"

        # Layer 2: Per-pair exchange budget (bidirectional key)
        pair_budget = config.dm_pair_exchange_budget
        pair_key = ":".join(sorted([agent_id, partner_id]))
        pair_times = self._dm_pair_counts.get(pair_key, [])
        pair_times = [t for t in pair_times if t > cutoff]
        self._dm_pair_counts[pair_key] = pair_times
        if len(pair_times) >= pair_budget:
            return f"pair_budget ({len(pair_times)}/{pair_budget} in {window:.0f}s)"

        return None

    async def _think_for_agent(self, agent: Any, rank: Rank, trust_score: float) -> None:
        """Gather context, send proactive_think intent, post result if meaningful.

        AD-419: If a duty is due, send a duty-specific prompt. If no duty is due,
        send a free-form think prompt that requires justification.
        """
        rt = self._runtime
        context_parts = await self._gather_context(agent, trust_score)

        # AD-419: Check duty schedule
        duty = None
        if self._duty_tracker:
            due_duties = self._duty_tracker.get_due_duties(agent.agent_type)
            if due_duties:
                duty = due_duties[0]  # Highest priority
            else:
                # BF-021 refined: No duty due — allow free-form thinks but at
                # reduced frequency (3x cooldown). Agents stay alive between
                # duty cycles instead of going completely dark.
                # BF-025: Guard with last > 0 so first-ever think always passes.
                # time.monotonic() on fresh CI runners can be < idle_cooldown.
                last = self._last_proactive.get(agent.id, 0.0)
                idle_cooldown = self.get_agent_cooldown(agent.id) * 3
                if last > 0 and time.monotonic() - last < idle_cooldown:
                    return

        # AD-595e: Qualification gate for duty dispatch
        if duty and not await self._check_duty_qualification(agent, duty):
            return

        # AD-502: Inject post count for temporal awareness
        try:
            rt = self._runtime
            if rt and hasattr(rt, 'ward_room') and rt.ward_room:
                one_hour_ago = time.time() - 3600
                cooldown_ts = rt._ward_room_cooldowns.get(agent.id, 0)
                # Count as 1 if posted within the last hour, 0 otherwise
                # This is a rough proxy — exact count would require WR query
                agent._recent_post_count = 1 if cooldown_ts > one_hour_ago else 0
            else:
                agent._recent_post_count = 0
        except Exception:
            logger.debug("Recent post count tracking failed", exc_info=True)

        intent = IntentMessage(
            intent="proactive_think",
            params={
                "context_parts": context_parts,
                "trust_score": format_trust(trust_score),
                "agency_level": agency_from_rank(rank).value,
                "rank": rank.value,  # AD-437: for action space awareness
                "agent_type": agent.agent_type,
                "duty": {
                    "duty_id": duty.duty_id,
                    "description": duty.description,
                } if duty else None,
            },
            target_agent_id=agent.id,
        )

        # AD-576b: tight in-cycle retry on transient LLM errors before
        # incrementing the failure counter. Two attempts at [0.5, 1.5]s
        # backoff; counter increments only after both fail.
        _BACKOFFS_SECONDS = (0.5, 1.5)
        _LLM_ERROR_KEYWORDS = (
            "llm", "timeout", "connection", "unreachable",
            "rate limit", "api error", "httpx", "openai",
        )
        result = await agent.handle_intent(intent)
        for _backoff in _BACKOFFS_SECONDS:
            if result and result.success and result.result:
                break
            _is_transient = (
                not result
                or (result and hasattr(result, 'error') and result.error and
                    any(kw in str(result.error).lower() for kw in _LLM_ERROR_KEYWORDS))
            )
            if not _is_transient:
                break
            await asyncio.sleep(_backoff)
            result = await agent.handle_intent(intent)

        if not result or not result.success or not result.result:
            # BF-228: Only count actual LLM errors toward LLM failure status,
            # not empty responses or chain-level issues.
            is_llm_error = (
                not result
                or (result and hasattr(result, 'error') and result.error and
                    any(kw in str(result.error).lower() for kw in _LLM_ERROR_KEYWORDS))
            )
            if is_llm_error:
                self._llm_failure_count += 1
                # AD-576: Update LLM status state machine
                await self._update_llm_status(failure=True)
            # BF-069: Log failure details for visibility
            error_detail = ""
            if result and hasattr(result, 'error') and result.error:
                error_detail = str(result.error)
            elif result and not result.success:
                error_detail = "agent returned unsuccessful result"
            if error_detail:
                logger.warning(
                    "BF-069: Proactive think failed for %s: %s (consecutive failures: %d)",
                    agent.agent_type, error_detail, self._llm_failure_count,
                )
            return

        response_text = str(result.result).strip()
        # BF-199: Extract text from leaked chain JSON
        from probos.utils.text_sanitize import sanitize_ward_room_text
        response_text = sanitize_ward_room_text(response_text)
        if not response_text or "[NO_RESPONSE]" in response_text:
            # Record duty execution even if agent had nothing to report
            if duty and self._duty_tracker:
                self._duty_tracker.record_execution(agent.agent_type, duty.duty_id)
            # AD-414: Optional trust signal for disciplined silence
            if self._config:
                no_response_weight = self._config.trust_no_response_weight
                if no_response_weight > 0:
                    self._runtime.trust_network.record_outcome(
                        agent.id,
                        success=True,
                        weight=no_response_weight,
                        intent_type="proactive_no_response",
                    )
            # AD-430a: Store no-response as episodic memory (prevents redundant re-analysis)
            if hasattr(rt, 'episodic_memory') and rt.episodic_memory:
                try:
                    from probos.types import AnchorFrame, Episode, MemorySource
                    callsign = ""
                    if hasattr(rt, 'callsign_registry'):
                        callsign = rt.callsign_registry.get_callsign(agent.agent_type)

                    # AD-567a: Resolve department for anchor
                    _dept = ""
                    try:
                        _ont = getattr(rt, 'ontology', None)
                        if _ont:
                            _dept = _ont.get_agent_department(agent.agent_type) or ""
                        if not _dept:
                            from probos.cognitive.standing_orders import get_department as _get_dept
                            _dept = _get_dept(agent.agent_type) or ""
                    except Exception:
                        pass

                    episode = Episode(
                        user_input=f"[Proactive thought — no response] {callsign or agent.agent_type}: reviewed context, nothing to report",
                        timestamp=time.time(),
                        agent_ids=[getattr(agent, 'sovereign_id', None) or agent.id],
                        outcomes=[{
                            "intent": "proactive_think",
                            "success": True,
                            "response": "[NO_RESPONSE]",
                            "duty_id": duty.duty_id if duty else None,
                            "agent_type": agent.agent_type,
                        }],
                        reflection=f"{callsign or agent.agent_type} reviewed context but had nothing to report.",
                        source=MemorySource.DIRECT,
                        anchors=AnchorFrame(
                            channel="duty_report",
                            duty_cycle_id=duty.duty_id if duty else "",
                            department=_dept,
                            trigger_type="duty_cycle" if duty else "proactive_think",
                            watch_section=derive_watch_section(),
                            event_log_window=float(len(rt.event_log.recent(seconds=60))) if hasattr(rt, 'event_log') and hasattr(rt.event_log, 'recent') else 0.0,
                        ),
                    )
                    from probos.cognitive.episodic import EpisodicMemory
                    if EpisodicMemory.should_store(episode):
                        await rt.episodic_memory.store(episode)
                except Exception:
                    logger.debug("Failed to store no-response episode for %s", agent.agent_type, exc_info=True)
            # AD-488: Record no-response as event (rapid no-responses can also indicate loops)
            self._circuit_breaker.record_event(agent.id, "no_response", "")
            return

        # BF-032: Skip if too similar to agent's recent posts
        if await self._is_similar_to_recent_posts(agent, response_text):
            logger.debug(
                "BF-032: Suppressed similar proactive post from %s",
                agent.agent_type,
            )
            # Still record duty execution if applicable
            if duty and self._duty_tracker:
                self._duty_tracker.record_execution(agent.agent_type, duty.duty_id)
            return

        # AD-493: Semantic novelty gate — suppress rehashed observations
        if getattr(self, '_novelty_gate', None):
            try:
                verdict = self._novelty_gate.check(agent.id, response_text)
                if not verdict.is_novel:
                    logger.info(
                        "AD-493: Suppressed rehashed observation from %s (sim=%.3f, matched='%s')",
                        agent.agent_type, verdict.similarity, verdict.matched_preview[:60],
                    )
                    if duty and self._duty_tracker:
                        self._duty_tracker.record_execution(agent.agent_type, duty.duty_id)
                    return
            except Exception:
                logger.debug("AD-493: Novelty gate check failed, allowing post", exc_info=True)

        # BF-172: Suppress raw JSON intent payloads leaking from LLM mode-confusion
        stripped = response_text.lstrip()
        if stripped.startswith("{") and '"intents"' in stripped[:200]:
            logger.warning(
                "BF-172: Suppressed raw intent JSON from %s (%d chars)",
                agent.agent_type, len(response_text),
            )
            if duty and self._duty_tracker:
                self._duty_tracker.record_execution(agent.agent_type, duty.duty_id)
            return

        # AD-437: Extract and process structured actions from proactive response
        cleaned_text, actions_taken = await self._extract_and_execute_actions(
            agent, response_text
        )
        if cleaned_text != response_text:
            response_text = cleaned_text

        # BF-215: Guard against empty text after tag stripping (BF-203/BF-174)
        if not response_text.strip():
            logger.debug(
                "BF-215: Suppressed empty post from %s after action extraction",
                agent.agent_type,
            )
            if duty and self._duty_tracker:
                self._duty_tracker.record_execution(agent.agent_type, duty.duty_id)
            return

        # AD-573: Record proactive observation to working memory
        try:
            wm = getattr(agent, 'working_memory', None)
            if wm and response_text and "[NO_RESPONSE]" not in response_text:
                wm.record_observation(
                    response_text[:200],
                    source="proactive",
                )
        except Exception:
            logger.debug("AD-573: Working memory observation record failed", exc_info=True)

        # Post to Ward Room — find agent's department channel
        await self._post_to_ward_room(agent, response_text)
        self._last_proactive[agent.id] = time.monotonic()
        # BF-228: Decay failure count instead of hard reset — prevents flap cycles.
        # One success reduces count by 1; requires sustained success to reach operational.
        if self._llm_failure_count > 0:
            self._llm_failure_count -= 1

        # AD-576: Update LLM status on recovery
        if self._llm_status != "operational":
            await self._update_llm_status(failure=False)

        # AD-488: Record cognitive event and check for rumination
        self._circuit_breaker.record_event(
            agent.id,
            "proactive_think",
            response_text[:500] if response_text else "",
            infrastructure_degraded=(self._llm_status != "operational"),  # AD-576
        )
        # AD-488: Record Ward Room post as cognitive event
        self._circuit_breaker.record_event(
            agent.id,
            "ward_room_post",
            response_text[:500] if response_text else "",
            infrastructure_degraded=(self._llm_status != "operational"),  # AD-576
        )
        if self._circuit_breaker.check_and_trip(agent.id):
            # Breaker tripped — fire bridge alert for Counselor awareness
            if self._on_event:
                callsign = getattr(agent, 'callsign', agent.agent_type)
                self._on_event({
                    "type": EventType.BRIDGE_ALERT.value,
                    "source": "circuit_breaker",
                    "severity": "warning",
                    "title": f"Circuit Breaker: {callsign}",
                    "detail": (
                        f"{callsign}'s cognitive circuit breaker has activated "
                        f"(trip #{self._circuit_breaker.get_status(agent.id)['trip_count']}). "
                        "Repetitive thought patterns detected. "
                        "Forced cooldown applied — not punishment, health protection."
                    ),
                })
            # AD-503/AD-495: Emit circuit_breaker_trip event for Counselor subscription
            if self._on_event:
                breaker_status = self._circuit_breaker.get_status(agent.id)
                self._on_event({
                    "type": EventType.CIRCUIT_BREAKER_TRIP.value,
                    "data": {
                        "agent_id": agent.id,
                        "agent_type": agent.agent_type,
                        "callsign": getattr(agent, "callsign", ""),
                        "trip_count": breaker_status.get("trip_count", 1),
                        "cooldown_seconds": breaker_status.get("cooldown_seconds", 900.0),
                        "trip_reason": breaker_status.get("trip_reason", "unknown"),
                        "zone": breaker_status.get("zone", "red"),  # AD-506a
                    },
                })
        else:
            # AD-506a: Check zone and emit concern if amber (not tripped)
            zone = self._circuit_breaker.get_zone(agent.id)
            if zone == "amber" and self._on_event:
                status = self._circuit_breaker.get_status(agent.id)
                self._on_event({
                    "type": EventType.SELF_MONITORING_CONCERN.value,
                    "data": {
                        "agent_id": agent.id,
                        "agent_callsign": getattr(agent, "callsign", ""),
                        "zone": "amber",
                        "similarity_ratio": status.get("similarity_ratio", 0.0),
                        "velocity_ratio": status.get("velocity_ratio", 0.0),
                        "infrastructure_correlated": self._llm_status != "operational",  # AD-576
                    },
                })

        # AD-506b: Emit zone recovery event
        transition = self._circuit_breaker.get_last_zone_transition(agent.id)
        if transition and self._on_event:
            old_zone, new_zone = transition
            # Recovery = zone improved (lower severity)
            zone_order = {"green": 0, "amber": 1, "red": 2, "critical": 3}
            if zone_order.get(new_zone, 0) < zone_order.get(old_zone, 0):
                self._on_event({
                    "type": EventType.ZONE_RECOVERY.value,
                    "data": {
                        "agent_id": agent.id,
                        "agent_callsign": getattr(agent, "callsign", ""),
                        "old_zone": old_zone,
                        "new_zone": new_zone,
                    },
                })

        # AD-414: Emit attenuated trust signal for successful proactive think
        if self._config:
            trust_weight = self._config.trust_reward_weight
            if duty:
                trust_weight += self._config.trust_duty_bonus  # Duty completion bonus
            if trust_weight > 0:
                new_score = self._runtime.trust_network.record_outcome(
                    agent.id,
                    success=True,
                    weight=trust_weight,
                    intent_type="proactive_think",
                )

        # Record duty execution after successful post
        if duty and self._duty_tracker:
            self._duty_tracker.record_execution(agent.agent_type, duty.duty_id)

        # AD-430a: Store proactive thought as episodic memory
        # BF-039: Skip if posted to Ward Room — WR creates its own episode,
        # storing here too would be a duplicate.
        wr_available = hasattr(rt, 'ward_room') and rt.ward_room
        if not wr_available and hasattr(rt, 'episodic_memory') and rt.episodic_memory:
            try:
                from probos.types import AnchorFrame, Episode
                callsign = ""
                if hasattr(rt, 'callsign_registry'):
                    callsign = rt.callsign_registry.get_callsign(agent.agent_type)
                thought_summary = response_text[:200]

                # AD-567a: Resolve department for anchor
                _dept = ""
                try:
                    _ont = getattr(rt, 'ontology', None)
                    if _ont:
                        _dept = _ont.get_agent_department(agent.agent_type) or ""
                    if not _dept:
                        from probos.cognitive.standing_orders import get_department as _get_dept
                        _dept = _get_dept(agent.agent_type) or ""
                except Exception:
                    pass

                # AD-577: Use earliest WR source timestamp if available
                _wr_activity = context.get("ward_room_activity", [])
                _earliest_source_ts = min(
                    (a.get("created_at", 0.0) for a in _wr_activity if a.get("created_at", 0.0) > 0),
                    default=0.0,
                )

                # AD-663: Provenance - identify observed WR content
                _wr_origin = ""
                _wr_version = ""
                if _wr_activity:
                    _first_post_id = _wr_activity[0].get("post_id", "") if _wr_activity else ""
                    if _first_post_id:
                        _wr_origin = f"wr-post:{_first_post_id}"
                    _post_ids = sorted(
                        a.get("post_id", "") for a in _wr_activity if a.get("post_id")
                    )
                    if _post_ids:
                        _wr_version = hashlib.sha256(
                            "|".join(_post_ids).encode("utf-8")
                        ).hexdigest()[:16]

                episode = Episode(
                    user_input=f"[Proactive thought] {callsign or agent.agent_type}: {thought_summary}",
                    timestamp=time.time(),
                    agent_ids=[getattr(agent, 'sovereign_id', None) or agent.id],
                    outcomes=[{
                        "intent": "proactive_think",
                        "success": True,
                        "response": response_text[:500],
                        "duty_id": duty.duty_id if duty else None,
                        "agent_type": agent.agent_type,
                    }],
                    reflection=f"{callsign or agent.agent_type} observed: {thought_summary}",
                    source=MemorySource.DIRECT,
                    anchors=AnchorFrame(
                        channel="duty_report",
                        duty_cycle_id=duty.duty_id if duty else "",
                        department=_dept,
                        trigger_type="duty_cycle" if duty else "proactive_think",
                        watch_section=derive_watch_section(),
                        event_log_window=float(len(rt.event_log.recent(seconds=60))) if hasattr(rt, 'event_log') and hasattr(rt.event_log, 'recent') else 0.0,
                        source_timestamp=_earliest_source_ts,  # AD-577
                        # AD-663: Provenance
                        source_origin_id=_wr_origin,
                        artifact_version=_wr_version,
                    ),
                )
                from probos.cognitive.episodic import EpisodicMemory
                if EpisodicMemory.should_store(episode):
                    await rt.episodic_memory.store(episode)
            except Exception:
                logger.debug("Failed to store proactive thought episode for %s", agent.agent_type, exc_info=True)

        if self._on_event:
            self._on_event({
                "type": EventType.PROACTIVE_THOUGHT.value,
                "data": {
                    "agent_id": agent.id,
                    "agent_type": agent.agent_type,
                    "response_length": len(response_text),
                    "duty_id": duty.duty_id if duty else None,
                },
            })

        logger.info(
            "Proactive thought from %s (%s): %d chars%s",
            agent.agent_type, rank.value, len(response_text),
            f" [duty: {duty.duty_id}]" if duty else " [free-form]",
        )

    async def _gather_context(self, agent: Any, trust_score: float) -> dict:
        """Gather recent context for the agent's proactive review."""
        rt = self._runtime
        context: dict[str, Any] = {}

        # AD-502: Temporal context for agent prompt
        if hasattr(rt, '_start_time_wall'):
            context["system_start_time"] = rt._start_time_wall
        if hasattr(rt, '_lifecycle_state'):
            context["lifecycle_state"] = rt._lifecycle_state
        if hasattr(rt, '_stasis_duration'):
            context["stasis_duration"] = rt._stasis_duration

        # BF-034: Cold-start context note for agents
        # AD-567g: Orientation subsumes this note when available
        _has_orientation = bool(getattr(agent, '_orientation_rendered', None))
        if not _has_orientation and hasattr(rt, 'is_cold_start') and rt.is_cold_start:
            context["system_note"] = (
                "SYSTEM NOTE: This is a fresh start after a system reset. "
                "All trust scores are at baseline (0.5). This is normal initialization, "
                "not a demotion. Build trust through demonstrated competence. "
                "You have no prior episodic memories — do not reference or invent past experiences."
            )

        # AD-572b: Captain engagement signals into proactive context
        engagement_provider = getattr(rt, "captain_engagement_provider", None)
        if engagement_provider is not None and hasattr(engagement_provider, "snapshot"):
            try:
                context["captain_engagement"] = engagement_provider.snapshot()
            except Exception:
                logger.debug("AD-572b: captain engagement snapshot failed", exc_info=True)

            # AD-572c: ward room activity summary (per-channel aggregation)
            if hasattr(engagement_provider, "wardroom_activity_summary"):
                try:
                    summary = await engagement_provider.wardroom_activity_summary()
                    if isinstance(context.get("captain_engagement"), dict):
                        context["captain_engagement"]["wardroom_activity_summary"] = summary
                except Exception:
                    logger.debug(
                        "AD-572c: wardroom_activity_summary failed", exc_info=True,
                    )

            # AD-572e: Task awareness in Captain DM context
            if hasattr(engagement_provider, "task_awareness"):
                try:
                    task_summary = await engagement_provider.task_awareness(agent.id)
                    if isinstance(context.get("captain_engagement"), dict):
                        context["captain_engagement"]["task_awareness"] = task_summary
                except Exception:
                    logger.debug(
                        "AD-572e: task_awareness injection failed", exc_info=True,
                    )

        # AD-567g: Proactive orientation supplement (diminishing)
        if getattr(self, '_orientation_service', None) and self._config:
            try:
                _ocfg = getattr(rt, 'config', None)
                if _ocfg and getattr(_ocfg, 'orientation', None) and _ocfg.orientation.proactive_supplement:
                    _birth = getattr(agent, '_birth_timestamp', None)
                    _age = (time.time() - _birth) if _birth else float('inf')
                    if _age < _ocfg.orientation.orientation_window_seconds:
                        _ep_count = 0
                        if hasattr(rt, 'episodic_memory') and rt.episodic_memory:
                            try:
                                _sid = getattr(agent, 'sovereign_id', None) or agent.id
                                _eps = await rt.episodic_memory.recall("", agent_id=_sid, k=1)
                                _ep_count = len(_eps) if _eps else 0
                            except Exception:
                                pass
                        _ctx = self._orientation_service.build_orientation(
                            agent,
                            lifecycle_state=getattr(rt, '_lifecycle_state', 'restart'),
                            episodic_memory_count=_ep_count,
                            trust_score=trust_score,
                        )
                        _ctx = OrientationContext(
                            **{
                                **{f.name: getattr(_ctx, f.name) for f in _ctx.__dataclass_fields__.values()},
                                'agent_age_seconds': _age,
                            }
                        )
                        _supp = self._orientation_service.render_proactive_orientation(_ctx)
                        if _supp:
                            context["orientation_supplement"] = _supp
            except Exception:
                logger.debug("AD-567g: proactive supplement failed", exc_info=True)

        # 1. Recent episodic memories (sovereign — only this agent's experiences)
        if hasattr(rt, 'episodic_memory') and rt.episodic_memory:
            try:
                _agent_mem_id = getattr(agent, 'sovereign_id', None) or agent.id  # AD-441

                # AD-567b: Use salience-weighted recall with dynamic query
                em = rt.episodic_memory
                episodes = []

                # AD-620: Resolve recall tier from rank + billet clearance
                from probos.earned_agency import effective_recall_tier, resolve_billet_clearance, resolve_active_grants, RecallTier
                from probos.cognitive.episodic import resolve_recall_tier_params
                _rank = getattr(agent, 'rank', None)
                _billet_clearance = resolve_billet_clearance(
                    getattr(agent, 'agent_type', ''),
                    getattr(rt, 'ontology', None),
                )
                # AD-622: Include active grants in tier resolution
                _active_grants = resolve_active_grants(
                    getattr(agent, 'sovereign_id', None) or agent.id,
                    getattr(rt, 'clearance_grant_store', None),
                )
                _recall_tier = effective_recall_tier(_rank, _billet_clearance, _active_grants)
                mem_cfg = None
                if hasattr(rt, 'config') and hasattr(rt.config, 'memory'):
                    mem_cfg = rt.config.memory
                _tier_cfg = getattr(mem_cfg, 'recall_tiers', None) if mem_cfg else None
                _tier_params = resolve_recall_tier_params(_recall_tier.value, _tier_cfg)

                # AD-568a: Classify retrieval strategy
                from probos.cognitive.source_governance import (
                    classify_retrieval_strategy, RetrievalStrategy,
                    compute_adaptive_budget, compute_source_framing,
                )
                _episode_count = 0
                if hasattr(em, 'count_for_agent'):
                    try:
                        _episode_count = await em.count_for_agent(_agent_mem_id)
                    except Exception:
                        _episode_count = 1
                # AD-568d: Thread confabulation rate from Counselor profile
                _confab_rate = 0.0
                try:
                    if rt and hasattr(rt, 'registry'):
                        _counselor_agents = rt.registry.get_by_pool("counselor")
                        if _counselor_agents:
                            _counselor = _counselor_agents[0]
                            if hasattr(_counselor, 'get_profile'):
                                _profile = _counselor.get_profile(agent.id)
                                if _profile:
                                    _confab_rate = getattr(_profile, 'confabulation_rate', 0.0)
                except Exception:
                    pass  # Non-critical — default 0.0

                _retrieval_strategy = classify_retrieval_strategy(
                    "proactive_think",
                    episodic_count=_episode_count,
                    recent_confabulation_rate=_confab_rate,  # AD-568d
                )

                scored_results = []
                _anchor_episodes = None  # AD-570c
                if _retrieval_strategy == RetrievalStrategy.NONE:
                    logger.debug("AD-568a: Skipping episodic recall for proactive (strategy=NONE)")
                    episodes = []
                else:
                    # AD-568a DEEP: Expand params (unlikely for proactive_think, but handled)
                    if _retrieval_strategy == RetrievalStrategy.DEEP:
                        _tier_params = dict(_tier_params)
                        _tier_params["k"] = int(_tier_params.get("k", 5) * 1.5)
                        _tier_params["context_budget"] = int(_tier_params.get("context_budget", 4000) * 1.5)
                        _tier_params["anchor_confidence_gate"] = max(
                            0.0, _tier_params.get("anchor_confidence_gate", 0.3) - 0.1
                        )

                    if hasattr(em, 'recall_weighted') and _tier_params.get("use_salience_weights", True):
                        # Derive query from agent context instead of hardcoded "recent activity"
                        _duty_type = ""
                        query = f"{agent.agent_type} {_duty_type} recent duty observations".strip()

                        # AD-570c: Try anchor-indexed recall for proactive queries
                        _anchor_episodes = None
                        try:
                            from probos.cognitive.source_governance import parse_anchor_query
                            known_callsigns = []
                            if hasattr(rt, 'callsign_registry'):
                                try:
                                    _all = rt.callsign_registry.all_callsigns()
                                    known_callsigns = list(_all.values()) if isinstance(_all, dict) else list(_all)
                                except Exception:
                                    pass
                            _anchor_q = parse_anchor_query(query, known_callsigns=known_callsigns)
                            if _anchor_q.has_anchor_signal and hasattr(em, 'recall_by_anchor'):
                                _anchor_results = await em.recall_by_anchor(
                                    department=_anchor_q.department,
                                    trigger_agent=_anchor_q.trigger_agent,
                                    participants=_anchor_q.participants if _anchor_q.participants else None,
                                    time_range=_anchor_q.time_range,
                                    semantic_query=_anchor_q.semantic_query,
                                    agent_id=_agent_mem_id,
                                    limit=10,
                                )
                                if _anchor_results and isinstance(_anchor_results, list) and len(_anchor_results) > 0:
                                    _anchor_episodes = _anchor_results
                                    logger.debug("AD-570c: Proactive anchor recall returned %d episodes", len(_anchor_results))
                        except Exception:
                            logger.debug("AD-570c: Proactive anchor recall failed", exc_info=True)

                        trust_net = getattr(rt, 'trust_network', None)
                        heb_router = getattr(rt, 'hebbian_router', None)

                        scored_results = await em.recall_weighted(
                            _agent_mem_id, query,
                            trust_network=trust_net,
                            hebbian_router=heb_router,
                            intent_type="proactive_think",
                            k=_tier_params.get("k", 5),
                            context_budget=_tier_params.get("context_budget", 4000),
                            weights=getattr(mem_cfg, 'recall_weights', None) if mem_cfg else None,
                            anchor_confidence_gate=_tier_params.get("anchor_confidence_gate", 0.3),
                            composite_score_floor=_tier_params.get("composite_score_floor", 0.0),
                            max_recall_episodes=_tier_params.get("max_recall_episodes", 0),
                            recall_quality_floor=_tier_params.get("recall_quality_floor", 0.0),
                        )
                        episodes = [rs.episode for rs in scored_results]
                    else:
                        # BASIC tier: vector similarity only
                        episodes = await em.recall_for_agent(
                            _agent_mem_id, "recent activity", k=_tier_params.get("k", 3)
                        )

                    # AD-568b: Adaptive budget scaling
                    if scored_results and _retrieval_strategy != RetrievalStrategy.NONE:
                        _budget_adj = compute_adaptive_budget(
                            _tier_params.get("context_budget", 4000),
                            recall_scores=scored_results,
                            episode_count=_episode_count,
                            strategy=_retrieval_strategy,
                        )
                        if _budget_adj.scale_factor != 1.0:
                            logger.debug(
                                "AD-568b: Proactive budget adjusted %d→%d (%s)",
                                _budget_adj.original_budget, _budget_adj.adjusted_budget,
                                _budget_adj.reason,
                            )
                            _adjusted_episodes = []
                            _budget_used = 0
                            for rs in scored_results:
                                _ep_len = len(rs.episode.user_input) if hasattr(rs.episode, 'user_input') else 0
                                if _budget_used + _ep_len > _budget_adj.adjusted_budget and _adjusted_episodes:
                                    break
                                _adjusted_episodes.append(rs)
                                _budget_used += _ep_len
                            scored_results = _adjusted_episodes
                            episodes = [rs.episode for rs in scored_results]

                    # Fallback to old recall path
                    if not episodes:
                        episodes = await em.recall_for_agent(
                            _agent_mem_id, "recent activity", k=_tier_params.get("k", 5)
                        )
                    if not episodes and hasattr(em, 'recent_for_agent'):
                        episodes = await em.recent_for_agent(_agent_mem_id, k=_tier_params.get("k", 5))

                    # AD-570c: Merge anchor recall into proactive context
                    if _anchor_episodes:
                        _seen = {getattr(ep, 'id', id(ep)) for ep in _anchor_episodes}
                        for ep in episodes:
                            if getattr(ep, 'id', id(ep)) not in _seen:
                                _anchor_episodes.append(ep)
                                _seen.add(getattr(ep, 'id', id(ep)))
                        episodes = _anchor_episodes

                # AD-568c: Source framing for proactive path
                _framing = None
                if scored_results:
                    _scores = [getattr(rs, 'composite_score', 0.0) for rs in scored_results]
                    _confs = [getattr(rs, 'anchor_confidence', 0.0) for rs in scored_results]
                    _framing = compute_source_framing(
                        mean_anchor_confidence=sum(_confs) / len(_confs) if _confs else 0.0,
                        recall_count=len(scored_results),
                        mean_recall_score=sum(_scores) / len(_scores) if _scores else 0.0,
                        strategy=_retrieval_strategy,
                    )
                elif _retrieval_strategy == RetrievalStrategy.NONE:
                    _framing = compute_source_framing(strategy=RetrievalStrategy.NONE)
                if _framing:
                    context["_source_framing"] = _framing

                if episodes:
                    # AD-502: Include relative timestamps on recalled memories
                    include_ts = True
                    if hasattr(rt, 'config') and hasattr(rt.config, 'temporal'):
                        include_ts = rt.config.temporal.include_episode_timestamps

                    memory_list = []
                    for ep in episodes:
                        mem = {
                            "input": (ep.user_input[:500] + " [trimmed]") if ep.user_input and len(ep.user_input) > 500 else (ep.user_input or ""),
                            "reflection": (ep.reflection[:500] + " [trimmed]") if ep.reflection and len(ep.reflection) > 500 else (ep.reflection or ""),
                            "source": getattr(ep, 'source', 'direct'),
                            "verified": False,  # AD-567b: proactive path parity
                        }
                        if include_ts and ep.timestamp > 0:
                            mem["age"] = format_duration(time.time() - ep.timestamp)
                        # AD-567b: Anchor context for formatting
                        anchors = getattr(ep, 'anchors', None)
                        if isinstance(anchors, AnchorFrame):
                            mem["anchor_channel"] = anchors.channel or ""
                            mem["anchor_department"] = anchors.department or ""
                            mem["anchor_participants"] = ", ".join(anchors.participants) if anchors.participants else ""
                            mem["anchor_trigger"] = anchors.trigger_type or ""
                        memory_list.append(mem)

                    context["recent_memories"] = memory_list
            except Exception:
                logger.debug("Episodic recall failed for %s", agent.id, exc_info=True)

        # AD-657: Surface diagnostic exemplars for any consolidated dream pattern that matches recent activity
        try:
            store = getattr(rt, 'procedure_store', None)
            em_for_exemplars = getattr(rt, 'episodic_memory', None)
            if store and em_for_exemplars:
                query = f"{agent.agent_type} recent duty observations".strip()
                matches = await store.find_matching(
                    query, n_results=1, exclude_negative=True,
                )
                if matches and matches[0].get("score", 0.0) >= 0.5:
                    proc_id = matches[0]["id"]
                    procedure = await store.get(proc_id)
                    if procedure and procedure.trace_exemplars:
                        exemplar_eps = await em_for_exemplars.get_by_ids(
                            procedure.trace_exemplars[:3],
                        )
                        if exemplar_eps:
                            now = time.time()
                            exemplars_payload = []
                            for ep in exemplar_eps:
                                ui = ep.user_input or ""
                                rf = ep.reflection or ""
                                exemplars_payload.append({
                                    "input": (ui[:300] + " [trimmed]") if len(ui) > 300 else ui,
                                    "reflection": (rf[:300] + " [trimmed]") if len(rf) > 300 else rf,
                                    "importance": ep.importance,
                                    "age": format_duration(now - ep.timestamp) if ep.timestamp > 0 else "",
                                })
                            context["recalled_procedure_exemplars"] = {
                                "procedure_name": procedure.name,
                                "procedure_id": procedure.id,
                                "exemplars": exemplars_payload,
                            }
        except Exception:
            logger.debug(
                "AD-657: trace-exemplar recall failed for %s", agent.id, exc_info=True,
            )

        # AD-462d: Social Memory — check for open memory queries to respond to
        if hasattr(rt, '_social_memory_service') and rt._social_memory_service:
            try:
                since = time.time() - 3600  # Last hour
                responses = await rt._social_memory_service.check_and_respond_to_queries(
                    agent_id=getattr(agent, 'sovereign_id', None) or agent.id,
                    agent_callsign=getattr(agent, 'callsign', '') or '',
                    since=since,
                    max_queries=2,
                )
                if responses:
                    context["memory_query_responses"] = responses
            except Exception:
                logger.debug("AD-462d: Social memory check failed", exc_info=True)

        # 2. Recent bridge alerts
        if hasattr(rt, 'bridge_alerts') and rt.bridge_alerts:
            try:
                alerts = rt.bridge_alerts.get_recent_alerts(limit=5)
                if alerts:
                    context["recent_alerts"] = [
                        {
                            "severity": a.severity.value,
                            "title": a.title,
                            "source": a.source,
                        }
                        for a in alerts
                    ]
            except Exception:
                logger.debug("Bridge alerts fetch failed", exc_info=True)

        # AD-440: Active chain-of-command orders for this agent
        order_mgr = getattr(rt, "order_manager", None)
        if order_mgr is not None:
            try:
                pending = order_mgr.list_active_for_agent(agent.id)
                if pending:
                    lines = ["ACTIVE ORDERS (act on these in priority order):"]
                    for o in pending:
                        lines.append(f"  - [{o.id}] from {o.from_post_id}: {o.directive}")
                    context["active_orders"] = "\n".join(lines)
            except Exception:
                logger.debug("AD-440: order context injection failed", exc_info=True)

        # 3. Recent system events
        if hasattr(rt, 'event_log') and rt.event_log:
            try:
                events = await rt.event_log.query(limit=10)
                if events:
                    context["recent_events"] = [
                        {
                            "category": e.get("category", ""),
                            "event": e.get("event", ""),
                            "agent_type": e.get("agent_type", ""),
                        }
                        for e in events[:10]
                    ]
            except Exception:
                logger.debug("Event log query failed", exc_info=True)

        # 4. Recent Ward Room activity in agent's department (AD-413)
        if hasattr(rt, 'ward_room') and rt.ward_room:
            try:
                from probos.cognitive.standing_orders import get_department
                # AD-429e: Prefer ontology, fall back to legacy dict
                ont = getattr(rt, 'ontology', None)
                dept = (ont.get_agent_department(agent.agent_type) if ont else None) or get_department(agent.agent_type)

                # BF-032: Build self-ID set for filtering own posts
                callsign = ""
                if hasattr(rt, 'callsign_registry'):
                    callsign = rt.callsign_registry.get_callsign(agent.agent_type)
                self_ids = {agent.id, agent.agent_type}
                if callsign:
                    self_ids.add(callsign)

                # BF-198: Skip threads agent already responded to via router
                wr_router = getattr(rt, 'ward_room_router', None)

                if dept:
                    channels = await rt.ward_room.list_channels()
                    dept_channel = None
                    all_hands_ch = None
                    for ch in channels:
                        if ch.channel_type == "department" and ch.department == dept:
                            dept_channel = ch
                        elif ch.name == "All Hands":
                            all_hands_ch = ch

                    # Look back one cooldown window (what happened since last think)
                    cooldown = self._agent_cooldowns.get(agent.id, self._default_cooldown)
                    since = time.time() - cooldown

                    if dept_channel:
                        activity = await rt.ward_room.get_recent_activity(
                            dept_channel.id, since=since, limit=5
                        )
                        if activity:
                            context["ward_room_activity"] = [
                                {
                                    "type": a["type"],
                                    "author": a["author"],
                                    # BF-224: Include both title AND body for threads.
                                    # Previously used title-only via a.get("title", a.get("body", "")),
                                    # so agents could never read thread body content.
                                    "body": (f"[{(a.get('post_id', a.get('id', '')) or '')[:8]}] " if a.get("post_id", a.get("id")) else "")
                                        + (
                                            (a["title"] + "\n" + a["body"])[:500]
                                            if a.get("title") and a.get("body")
                                            else (a.get("title") or a.get("body", ""))[:500]
                                        ),
                                    "net_score": a.get("net_score", 0),       # AD-426
                                    "post_id": a.get("post_id", a.get("id", "")),  # AD-426
                                    "thread_id": a.get("thread_id", ""),  # AD-437
                                    "created_at": a.get("created_at", 0.0),  # AD-577
                                }
                                for a in activity
                                if (a.get("author_id", "") or a.get("author", "")) not in self_ids  # BF-032
                                and not (wr_router and wr_router.has_agent_responded(agent.id, a.get("thread_id", "")))  # BF-198
                            ]
                            # AD-425: Mark channel as seen after consuming activity
                            try:
                                await rt.ward_room.update_last_seen(agent.id, dept_channel.id)
                            except Exception:
                                logger.debug("update_last_seen failed", exc_info=True)

                    # AD-425: Also include recent All Hands activity (ship-wide)
                    if all_hands_ch and (not dept_channel or all_hands_ch.id != dept_channel.id):
                        all_hands_activity = await rt.ward_room.get_recent_activity(
                            all_hands_ch.id, since=since, limit=3
                        )
                        # Filter: only DISCUSS threads (INFORM already consumed, ACTION is targeted)
                        all_hands_filtered = [
                            a for a in all_hands_activity
                            if a.get("thread_mode") != "inform"
                        ]
                        if all_hands_filtered:
                            if "ward_room_activity" not in context:
                                context["ward_room_activity"] = []
                            context["ward_room_activity"].extend([
                                {
                                    "type": item["type"],
                                    "author": item.get("author", "unknown"),
                                    # BF-224: Include both title AND body for threads
                                    "body": (f"[{(item.get('post_id', item.get('id', '')) or '')[:8]}] " if item.get("post_id", item.get("id")) else "")
                                        + (
                                            (item["title"] + "\n" + item["body"])[:500]
                                            if item.get("title") and item.get("body")
                                            else (item.get("title") or item.get("body", ""))[:500]
                                        ),
                                    "channel": "All Hands",
                                    "net_score": item.get("net_score", 0),       # AD-426
                                    "post_id": item.get("post_id", item.get("id", "")),  # AD-426
                                    "thread_id": item.get("thread_id", ""),  # AD-437
                                    "created_at": item.get("created_at", 0.0),  # AD-577
                                }
                                for item in all_hands_filtered[:3]
                                if (item.get("author_id", "") or item.get("author", "")) not in self_ids  # BF-032
                                and not (wr_router and wr_router.has_agent_responded(agent.id, item.get("thread_id", "")))  # BF-198
                            ])
                        # BF-226: Always mark All Hands as seen after checking,
                        # even when no new non-inform activity exists. Otherwise
                        # last_seen stays at crew creation time and unread count
                        # grows unboundedly.
                        try:
                            await rt.ward_room.update_last_seen(agent.id, all_hands_ch.id)
                        except Exception:
                            logger.debug("update_last_seen failed", exc_info=True)

                    # AD-526a: Recreation channel activity
                    rec_ch = next((c for c in channels if c.name == "Recreation"), None)
                    if rec_ch:
                        rec_items = await rt.ward_room.get_recent_activity(
                            rec_ch.id, limit=2, since=since
                        )
                        rec_filtered = [
                            item for item in rec_items
                            if (item.get("author_id", "") or item.get("author", "")) not in self_ids
                            and not (wr_router and wr_router.has_agent_responded(agent.id, item.get("thread_id", "")))  # BF-198
                        ]
                        if rec_filtered:
                            if "ward_room_activity" not in context:
                                context["ward_room_activity"] = []
                            context["ward_room_activity"].extend([
                                {
                                    "type": item["type"],
                                    "author": item.get("author", "unknown"),
                                    # BF-224: Include both title AND body for threads
                                    "body": (f"[{(item.get('post_id', item.get('id', '')) or '')[:8]}] " if item.get("post_id", item.get("id")) else "")
                                        + (
                                            (item["title"] + "\n" + item["body"])[:500]
                                            if item.get("title") and item.get("body")
                                            else (item.get("title") or item.get("body", ""))[:500]
                                        ),
                                    "channel": "Recreation",
                                    "net_score": item.get("net_score", 0),
                                    "post_id": item.get("post_id", item.get("id", "")),
                                    "thread_id": item.get("thread_id", ""),
                                    "created_at": item.get("created_at", 0.0),  # AD-577
                                }
                                for item in rec_filtered[:2]
                            ])
                        # BF-226: Always mark Recreation as seen after checking.
                        try:
                            await rt.ward_room.update_last_seen(agent.id, rec_ch.id)
                        except Exception:
                            logger.debug("update_last_seen failed", exc_info=True)
            except Exception:
                logger.debug("Ward Room context fetch failed for %s", agent.id, exc_info=True)

        # AD-577: Sort ward_room_activity by source timestamp and assign sequence indices
        if context.get("ward_room_activity"):
            context["ward_room_activity"].sort(key=lambda a: a.get("created_at", 0.0))
            for idx, item in enumerate(context["ward_room_activity"], start=1):
                item["sequence_index"] = idx

        # BF-110: Inject active game state so agent can see the board and know it's their turn
        rec_svc = getattr(rt, 'recreation_service', None)
        if rec_svc:
            try:
                callsign = ""
                if hasattr(rt, 'callsign_registry'):
                    callsign = rt.callsign_registry.get_callsign(agent.agent_type)
                if callsign:
                    game = rec_svc.get_game_by_player(callsign)  # AD-572: DRY
                    if game:
                        state = game.get("state", {})
                        players = [game.get("challenger", ""), game.get("opponent", "")]
                        board = rec_svc.render_board(game["game_id"])
                        valid_moves = rec_svc.get_valid_moves(game["game_id"])
                        is_my_turn = state.get("current_player") == callsign
                        context["active_game"] = {
                            "game_id": game["game_id"],
                            "game_type": game.get("game_type", ""),
                            "opponent": next((p for p in players if p != callsign), ""),
                            "is_my_turn": is_my_turn,
                            "board": board,
                            "valid_moves": valid_moves,
                            "moves_count": game.get("moves_count", 0),
                        }
            except Exception:
                logger.debug("BF-110: Game context injection failed for %s", agent.id, exc_info=True)

        # AD-573: Sync active game to working memory
        if "active_game" in context:
            try:
                wm = getattr(agent, 'working_memory', None)
                if wm:
                    ag = context["active_game"]
                    from probos.cognitive.agent_working_memory import ActiveEngagement
                    game_id = ag["game_id"]
                    if not wm.get_engagement(game_id):
                        wm.add_engagement(ActiveEngagement(
                            engagement_type="game",
                            engagement_id=game_id,
                            summary=f"Playing {ag.get('game_type', 'game')} against {ag.get('opponent', '?')}",
                            state={
                                "game_type": ag.get("game_type", ""),
                                "opponent": ag.get("opponent", ""),
                                "is_my_turn": ag.get("is_my_turn", False),
                                "render": ag.get("board", ""),
                            },
                        ))
                    else:
                        wm.update_engagement(game_id, state={
                            "is_my_turn": ag.get("is_my_turn", False),
                            "render": ag.get("board", ""),
                        })
            except Exception:
                logger.debug("AD-573: Working memory game sync failed for %s", agent.id, exc_info=True)

        # 5. Ontology context (AD-429a) — formal identity grounding
        if hasattr(rt, 'ontology') and rt.ontology:
            try:
                crew_ctx = rt.ontology.get_crew_context(agent.agent_type)
                if crew_ctx:
                    context["ontology"] = crew_ctx
            except Exception:
                logger.debug("Ontology context fetch failed for %s", agent.id, exc_info=True)

        # 6. Skill profile context (AD-429b) + AD-428b v1 development goals
        if hasattr(rt, 'skill_service') and rt.skill_service:
            try:
                profile = await rt.skill_service.get_profile(agent.id)
                if profile:
                    skill_summary = []
                    for record in profile.all_skills:
                        skill_summary.append(f"{record.skill_id}: level {record.proficiency.value} ({record.proficiency.name})")
                    if skill_summary:
                        context["skill_profile"] = skill_summary
            except Exception:
                logger.debug("Skill profile fetch failed for %s", agent.id, exc_info=True)
            # AD-428b v1: development goals (separate try so a goal-fetch
            # failure doesn't suppress skill_profile, and vice versa).
            try:
                goals = await rt.skill_service.get_development_goals(agent.id)
                if goals:
                    context["development_goals"] = goals
            except Exception:
                logger.debug(
                    "AD-428b: development_goals fetch failed for %s",
                    agent.id, exc_info=True,
                )

        # AD-630: Subordinate communication stats for Chiefs
        if hasattr(rt, 'ontology') and rt.ontology:
            try:
                subordinate_types = rt.ontology.get_subordinate_agent_types(
                    agent.agent_type
                )
                if subordinate_types and hasattr(rt, 'ward_room_service') and rt.ward_room_service:
                    sub_stats: dict = {}
                    since = getattr(rt, '_start_time_wall', None)
                    for sub_type in subordinate_types:
                        sub_agent = rt.agent_pool.get(sub_type) if hasattr(rt, 'agent_pool') else None
                        if sub_agent:
                            sub_id = getattr(sub_agent, 'sovereign_id', None) or sub_agent.id
                            stats = await rt.ward_room_service.get_agent_comm_stats(
                                sub_id, since=since
                            )
                            if stats.get("posts_total", 0) >= 3:
                                callsign = getattr(sub_agent, 'callsign', sub_type)
                                sub_stats[callsign] = stats
                    if sub_stats:
                        context["subordinate_stats"] = sub_stats
            except Exception:
                logger.debug("AD-630: Subordinate stats fetch failed for %s", agent.id, exc_info=True)

        # AD-635f: Clinical telemetry summary for clinical agents (Chapel, Echo)
        try:
            from probos.cognitive.clinical_telemetry import CLINICAL_ROLES
            _clinical_service = getattr(rt, "clinical_telemetry", None)
            if (
                _clinical_service is not None
                and getattr(agent, "agent_type", "") in CLINICAL_ROLES
            ):
                _clin: dict[str, Any] = {}

                # Dreams (most recent N)
                try:
                    _dreams = await _clinical_service.query_dream_history(
                        requester_agent_id=agent.id,
                        limit=_DREAM_RECENT_LIMIT,
                    )
                    if _dreams:
                        _latest = _dreams[0] if isinstance(_dreams, list) else None
                        _clin["dreams"] = {
                            "count": len(_dreams),
                            "latest_ts": (_latest or {}).get("ts"),
                        }
                except Exception:
                    logger.debug(
                        "AD-635f: dream summary fetch failed for %s", agent.id,
                        exc_info=True,
                    )

                # Chain traces (self-targeted)
                try:
                    _traces = await _clinical_service.query_agent_chain_traces(
                        requester_agent_id=agent.id,
                        target_agent_id=agent.id,
                        limit=_CHAIN_TRACE_RECENT_LIMIT,
                    )
                    if _traces:
                        _latest = _traces[0] if isinstance(_traces, list) else None
                        _clin["chain_traces"] = {
                            "count": len(_traces),
                            "latest_ts": (_latest or {}).get("ts"),
                            "latest_outcome": (_latest or {}).get("outcome"),
                        }
                except Exception:
                    logger.debug(
                        "AD-635f: chain trace summary fetch failed for %s", agent.id,
                        exc_info=True,
                    )

                # Circuit breakers (fleet-wide)
                try:
                    _breakers = await _clinical_service.query_circuit_breaker_history(
                        requester_agent_id=agent.id,
                        target_agent_id=None,
                        limit=_BREAKER_RECENT_LIMIT,
                    )
                    if _breakers:
                        _recent: list[dict[str, Any]] = []
                        for row in _breakers[:_BREAKER_RECENT_LIMIT]:
                            if not isinstance(row, dict):
                                continue
                            _recent.append({
                                "agent": row.get("agent_id") or row.get("target_agent_id"),
                                "from": row.get("from_zone") or row.get("from"),
                                "to": row.get("to_zone") or row.get("to"),
                                "ts": row.get("ts"),
                            })
                        _clin["breakers"] = {
                            "count": len(_breakers),
                            "recent_transitions": _recent,
                        }
                except Exception:
                    logger.debug(
                        "AD-635f: breaker summary fetch failed for %s", agent.id,
                        exc_info=True,
                    )

                if _clin:
                    context["clinical_telemetry"] = _clin
        except Exception:
            logger.debug(
                "AD-635f: clinical telemetry injection failed for %s", agent.id,
                exc_info=True,
            )

        # AD-504: Self-monitoring context
        try:
            callsign = ""
            if hasattr(rt, 'callsign_registry'):
                callsign = rt.callsign_registry.get_callsign(agent.agent_type)
            self_monitoring = await self._build_self_monitoring_context(agent, callsign, rt)
            if self_monitoring:
                context["self_monitoring"] = self_monitoring
        except Exception:
            logger.debug("Self-monitoring context failed for %s", agent.id, exc_info=True)

        # AD-588: Introspective telemetry for proactive path
        try:
            _telemetry_svc = getattr(rt, '_introspective_telemetry', None)
            if _telemetry_svc:
                _agent_id = getattr(agent, 'sovereign_id', None) or agent.id
                _snapshot = await _telemetry_svc.get_full_snapshot(_agent_id)
                _rendered = _telemetry_svc.render_telemetry_context(_snapshot)
                if _rendered:
                    context["introspective_telemetry"] = _rendered
        except Exception:
            logger.debug("AD-588: telemetry context assembly failed", exc_info=True)

        # AD-576: Infrastructure awareness context
        if self._llm_status != "operational":
            context["infrastructure_status"] = {
                "llm_status": self._llm_status,
                "consecutive_failures": self._llm_failure_count,
                "message": (
                    "The ship's communications array (LLM backend) is currently"
                    f" {self._llm_status}. If you receive an empty or degraded"
                    " response, this is an infrastructure issue — not a reflection"
                    " of your cognitive state. Do not self-diagnose based on empty"
                    " cycles during this period."
                ),
            }

        # AD-471: Night Orders context for conn-holder
        if hasattr(rt, 'conn_manager') and rt.conn_manager and rt.conn_manager.is_active:
            conn_state = rt.conn_manager.state
            if conn_state.holder_agent_id == agent.id:
                # This agent holds the conn — inject Night Orders
                night_ctx: dict[str, Any] = {
                    "role": "You currently hold the conn (temporary command authority).",
                    "conn_scope": {
                        "can_approve_builds": conn_state.can_approve_builds,
                        "can_change_alert_yellow": conn_state.can_change_alert_yellow,
                        "can_issue_orders": conn_state.can_issue_orders,
                    },
                }
                if hasattr(rt, '_night_orders_mgr') and rt._night_orders_mgr and rt._night_orders_mgr.active:
                    orders = rt._night_orders_mgr.orders
                    night_ctx["night_orders"] = {
                        "template": orders.template or "custom",
                        "instructions": orders.instructions,
                        "alert_boundary": orders.alert_boundary,
                        "escalation_triggers": orders.escalation_triggers,
                        "remaining_hours": round(max(0, orders.expires_at - time.time()) / 3600, 1),
                    }
                context["conn_authority"] = night_ctx

        # 7. Cognitive skill catalog (AD-596b) — available skills for this agent
        if hasattr(rt, 'cognitive_skill_catalog') and rt.cognitive_skill_catalog:
            try:
                _dept = None
                if hasattr(rt, 'ontology') and rt.ontology:
                    _dept = rt.ontology.get_agent_department(agent.agent_type)
                _rank_str = None
                if hasattr(rt, 'trust_network') and rt.trust_network:
                    try:
                        from probos.crew_profile import Rank
                        _trust = rt.trust_network.get_score(agent.id)
                        _rank_str = Rank.from_trust(_trust).value
                    except Exception:
                        pass
                skill_descs = rt.cognitive_skill_catalog.get_descriptions(
                    department=_dept,
                    agent_rank=_rank_str,
                )
                if skill_descs:
                    context["cognitive_skills"] = [
                        {"name": name, "description": desc}
                        for name, desc in skill_descs
                    ]
            except Exception:
                logger.debug("AD-596b: Cognitive skill context failed for %s", agent.id, exc_info=True)

        return context

    async def _build_self_monitoring_context(
        self,
        agent: Any,
        callsign: str,
        rt: Any,
    ) -> dict[str, Any]:
        """AD-504: Build self-monitoring context for agent self-regulation.

        Returns dict with keys: recent_posts, self_similarity, notebook_index,
        notebook_content, memory_state. All values are optional/nullable.
        Earned Agency tier controls verbosity.
        """
        result: dict[str, Any] = {}
        trust_score = rt.trust_network.get_score(agent.id) if hasattr(rt, 'trust_network') and rt.trust_network else 0.5
        rank = Rank.from_trust(trust_score)
        agency = agency_from_rank(rank)

        # Tier-based feature gating
        TIER_CONFIG = {
            AgencyLevel.REACTIVE: {"posts": 0, "notebooks": False, "similarity": False},
            AgencyLevel.SUGGESTIVE: {"posts": 3, "notebooks": False, "similarity": True},
            AgencyLevel.AUTONOMOUS: {"posts": 5, "notebooks": True, "similarity": True},
            AgencyLevel.UNRESTRICTED: {"posts": 5, "notebooks": True, "similarity": True},
        }
        tier = TIER_CONFIG.get(agency, TIER_CONFIG[AgencyLevel.SUGGESTIVE])

        # --- AD-506a: Cognitive zone awareness (before tier gating) ---
        if hasattr(self, '_circuit_breaker'):
            zone = self._circuit_breaker.get_zone(agent.id)
            # BF-161: Always include zone (including green) so working memory
            # stays current and telemetry doesn't show UNKNOWN.
            result["cognitive_zone"] = zone
            if zone == "amber":
                result["zone_note"] = (
                    "Your recent posts show increasing similarity to each other. "
                    "This is a natural signal to pause and consider: do you have "
                    "genuinely new information to contribute, or are you circling "
                    "the same ground? If unsure, try [NO_RESPONSE] or write to "
                    "your notebook instead."
                )
            elif zone == "red":
                result["zone_note"] = (
                    "Your cognitive circuit breaker has activated. This is health "
                    "protection, not punishment. The Counselor has been notified. "
                    "Focus on a different aspect of operations or respond with "
                    "[NO_RESPONSE] until you have a genuinely fresh perspective."
                )
            elif zone == "critical":
                result["zone_note"] = (
                    "Critical cognitive state — repeated pattern loops detected. "
                    "The Captain has been notified. Extended mandatory cooldown is "
                    "in effect. When you return, deliberately choose a completely "
                    "different topic. Your previous train of thought needs rest."
                )

        if tier["posts"] == 0:
            return result

        # --- (1) Recent output window ---
        try:
            if hasattr(rt, 'ward_room') and rt.ward_room:
                since = time.time() - 3600  # Last hour
                limit = tier["posts"]
                posts = await rt.ward_room.get_posts_by_author(callsign, limit=limit, since=since)
                if posts:
                    result["recent_posts"] = [
                        {
                            "body": p["body"][:150],
                            "age": format_duration(time.time() - p["created_at"]),
                        }
                        for p in posts
                    ]
        except Exception:
            logger.debug("Self-monitoring: failed to get recent posts for %s", callsign, exc_info=True)

        # --- (2) Self-similarity score ---
        if tier["similarity"]:
            try:
                posts_for_sim = result.get("recent_posts", [])
                if len(posts_for_sim) >= 2:
                    from probos.cognitive.similarity import jaccard_similarity, text_to_words
                    word_sets = [text_to_words(p["body"]) for p in posts_for_sim]
                    total_sim = 0.0
                    pair_count = 0
                    for j in range(len(word_sets)):
                        for k in range(j + 1, len(word_sets)):
                            total_sim += jaccard_similarity(word_sets[j], word_sets[k])
                            pair_count += 1
                    if pair_count > 0:
                        result["self_similarity"] = round(total_sim / pair_count, 2)
            except Exception:
                logger.debug("Self-monitoring: similarity calc failed for %s", callsign, exc_info=True)

        # --- (4) Dynamic cooldown ("take a breath") ---
        sim = result.get("self_similarity", 0.0)
        if sim >= 0.5:
            current_cooldown = self.get_agent_cooldown(agent.id)
            new_cooldown = min(current_cooldown * 1.5, 1800)
            if new_cooldown > current_cooldown:
                self.set_agent_cooldown(agent.id, new_cooldown)
                result["cooldown_increased"] = True

        # AD-505: Include cooldown reason if set
        reason = self.get_cooldown_reason(agent.id)
        if reason:
            result["cooldown_reason"] = reason

        # AD-573: Sync cognitive state to working memory
        try:
            wm = getattr(agent, 'working_memory', None)
            if wm:
                updates: dict = {}
                if "cognitive_zone" in result:
                    updates["zone"] = result["cognitive_zone"]
                if "cooldown_reason" in result:
                    updates["cooldown_reason"] = result["cooldown_reason"]
                if updates:
                    wm.update_cognitive_state(**updates)
        except Exception:
            logger.debug("AD-573: Working memory cognitive state sync failed", exc_info=True)

        # --- (7) Memory state awareness ---
        try:
            if hasattr(rt, 'episodic_memory') and rt.episodic_memory:
                episode_count = await rt.episodic_memory.count_for_agent(
                    getattr(agent, 'sovereign_id', agent.id)
                )
                lifecycle = getattr(rt, '_lifecycle_state', 'first_boot')
                uptime = time.time() - getattr(rt, '_start_time_wall', time.time())
                result["memory_state"] = {
                    "episode_count": episode_count,
                    "lifecycle": lifecycle,
                    "uptime_hours": round(uptime / 3600, 1),
                }
        except Exception:
            logger.debug("Self-monitoring: memory state failed for %s", callsign, exc_info=True)

        # --- (8) Notebook continuity ---
        # AD-567f: Cascade context delivered via Bridge Alerts + Counselor, not self-monitoring.
        # Verification context available as a tool agents invoke, not default injection.
        if tier["notebooks"]:
            try:
                if hasattr(rt, '_records_store') and rt._records_store:
                    entries = await rt._records_store.list_entries(
                        f"notebooks/{callsign}",
                        author=callsign,
                    )
                    if entries:
                        # AD-550: Enhanced index with content previews + recency
                        sorted_entries = sorted(
                            entries,
                            key=lambda e: e.get("frontmatter", {}).get("updated", ""),
                            reverse=True,
                        )
                        now_ts = time.time()

                        # Entry/topic counts
                        result["notebook_summary"] = {
                            "total_entries": len(entries),
                            "total_topics": len(set(
                                e.get("frontmatter", {}).get("topic", e["path"].split("/")[-1].replace(".md", ""))
                                for e in entries
                            )),
                        }

                        top5 = sorted_entries[:5]
                        enriched_index = []
                        for e in top5:
                            fm = e.get("frontmatter", {})
                            topic = fm.get("topic", e["path"].split("/")[-1].replace(".md", ""))
                            updated_str = fm.get("updated", "")

                            # Human-readable recency
                            recency = ""
                            if updated_str:
                                try:
                                    entry_ts = datetime.fromisoformat(updated_str).timestamp()
                                    delta_s = now_ts - entry_ts
                                    if delta_s < 3600:
                                        recency = f"{int(delta_s / 60)}m ago"
                                    elif delta_s < 86400:
                                        recency = f"{int(delta_s / 3600)}h ago"
                                    else:
                                        recency = f"{int(delta_s / 86400)}d ago"
                                except (ValueError, TypeError):
                                    pass

                            # Content preview (first 150 chars)
                            preview = ""
                            try:
                                entry_data = await rt._records_store.read_entry(
                                    e["path"],
                                    reader_id=callsign,
                                    reader_department=getattr(agent, 'department', ''),
                                )
                                if entry_data and entry_data.get("content"):
                                    preview = entry_data["content"][:150].replace("\n", " ").strip()
                            except Exception:
                                pass

                            enriched_index.append({
                                "topic": topic,
                                "updated": updated_str,
                                "recency": recency,
                                "preview": preview,
                                "revision": fm.get("revision", 1),  # AD-552
                            })

                        result["notebook_index"] = enriched_index

                        # AD-552: Repetition warnings for high-revision topics
                        _rep_thresh = 3
                        _rep_window_h = 48.0
                        if hasattr(rt, 'config') and hasattr(rt.config, 'records'):
                            _rep_thresh = getattr(rt.config.records, 'notebook_repetition_threshold_count', 3)
                            _rep_window_h = getattr(rt.config.records, 'notebook_repetition_window_hours', 48.0)
                        warnings = []
                        for ei in enriched_index:
                            rev = ei.get("revision", 1)
                            if rev >= _rep_thresh and ei.get("updated"):
                                try:
                                    _ei_ts = datetime.fromisoformat(ei["updated"]).timestamp()
                                    if (now_ts - _ei_ts) < (_rep_window_h * 3600):
                                        warnings.append(
                                            f"You've written about {ei['topic']} {rev} times recently. "
                                            f"Review your existing entry before writing again."
                                        )
                                except (ValueError, TypeError):
                                    pass
                        if warnings:
                            result["notebook_repetition_warnings"] = warnings

                    # Semantic pull: search notebooks for content relevant to current think
                    dept = getattr(agent, 'department', '')
                    if dept and entries:
                        search_results = await rt._records_store.search(
                            dept, scope=f"notebooks/{callsign}",
                        )
                        if search_results:
                            top = search_results[0]
                            result["notebook_content"] = {
                                "topic": top.get("path", "").split("/")[-1].replace(".md", ""),
                                "snippet": top.get("snippet", "")[:300],
                            }
            except Exception:
                logger.debug("Self-monitoring: notebook query failed for %s", callsign, exc_info=True)

            # --- Pending notebook reads from previous cycle ---
            pending_key = agent.id
            if pending_key in self._pending_notebook_reads:
                topic_slug = self._pending_notebook_reads.pop(pending_key)
                try:
                    if hasattr(rt, '_records_store') and rt._records_store:
                        entry = await rt._records_store.read_entry(
                            f"notebooks/{callsign}/{topic_slug}.md",
                            reader_id=callsign,
                            reader_department=getattr(agent, 'department', ''),
                        )
                        if entry:
                            result["notebook_content"] = {
                                "topic": topic_slug,
                                "snippet": entry.get("content", "")[:500],
                            }
                except Exception:
                    logger.debug("Self-monitoring: notebook read failed for %s/%s", callsign, topic_slug, exc_info=True)

        return result

    def _get_comm_gate_overrides(self, agent):
        """AD-625: Look up communication proficiency gate overrides."""
        _profiles = getattr(self._runtime, '_comm_profiles', {})
        _profile = _profiles.get(agent.id)
        if _profile is None:
            return None
        for rec in _profile.all_skills:
            if rec.skill_id == "communication":
                from probos.cognitive.comm_proficiency import get_gate_overrides
                return get_gate_overrides(rec.proficiency)
        return None

    # ------------------------------------------------------------------
    # AD-654a: Public wrappers for WardRoomPostPipeline cross-module calls
    # ------------------------------------------------------------------

    async def extract_and_execute_actions(
        self, agent: Any, text: str,
        *,
        post_budget: PostBudget | None = None,
    ) -> tuple[str, list[dict]]:
        """Public wrapper for _extract_and_execute_actions (AD-654a)."""
        return await self._extract_and_execute_actions(agent, text, post_budget=post_budget)

    async def is_similar_to_recent_posts(
        self, agent: Any, text: str, threshold: float = 0.5,
    ) -> bool:
        """Public wrapper for _is_similar_to_recent_posts (AD-654a)."""
        return await self._is_similar_to_recent_posts(agent, text, threshold)

    async def _is_similar_to_recent_posts(self, agent: Any, text: str, threshold: float = 0.5) -> bool:
        """BF-032: Check if proposed post is too similar to agent's recent Ward Room posts.

        Uses Jaccard similarity on word sets. Returns True if any recent post
        exceeds the similarity threshold.
        """
        rt = self._runtime
        if not rt or not hasattr(rt, 'ward_room') or not rt.ward_room:
            return False

        try:
            from probos.cognitive.standing_orders import get_department
            # AD-429e: Prefer ontology, fall back to legacy dict
            ont = getattr(rt, 'ontology', None)
            dept = (ont.get_agent_department(agent.agent_type) if ont else None) or get_department(agent.agent_type)
            channels = await rt.ward_room.list_channels()
            agent_posts: list[str] = []

            for ch in channels:
                try:
                    activity = await rt.ward_room.get_recent_activity(
                        ch.id, limit=10, since=None,
                    )
                    for a in activity:
                        author = a.get("author_id", "") or a.get("author", "")
                        if author == agent.id:
                            body = a.get("body", "")
                            if body:
                                agent_posts.append(body)
                except Exception:
                    logger.debug("Skipping channel", exc_info=True)
                    continue

            if not agent_posts:
                return False

            # Jaccard similarity on word sets
            new_words = set(text.lower().split())
            for post in agent_posts[:10]:  # BF-062: Check last 10 posts (was 3)
                old_words = set(post.lower().split())
                if not new_words or not old_words:
                    continue
                intersection = new_words & old_words
                union = new_words | old_words
                similarity = len(intersection) / len(union) if union else 0.0
                if similarity >= threshold:
                    return True

                # BF-197b: Containment check — if the shorter message's words
                # are mostly contained in the longer one, it's a near-duplicate
                # even when Jaccard is low (superset expansion dilutes Jaccard).
                shorter, longer = (new_words, old_words) if len(new_words) <= len(old_words) else (old_words, new_words)
                if shorter:
                    containment = len(shorter & longer) / len(shorter)
                    if containment >= 0.85:
                        return True

                # BF-062: Bigram similarity catches paraphrased near-duplicates
                new_bigrams = set(zip(text.lower().split(), text.lower().split()[1:]))
                old_bigrams = set(zip(post.lower().split(), post.lower().split()[1:]))
                if new_bigrams and old_bigrams:
                    bi_intersection = new_bigrams & old_bigrams
                    bi_union = new_bigrams | old_bigrams
                    bi_similarity = len(bi_intersection) / len(bi_union) if bi_union else 0.0
                    if bi_similarity >= threshold:
                        return True

            return False
        except Exception:
            logger.debug("Similarity check failed for %s", agent.id, exc_info=True)
            return False

    async def _post_to_ward_room(self, agent: Any, text: str) -> None:
        """Create a Ward Room thread with the agent's proactive observation."""
        rt = self._runtime

        # Find agent's department channel
        from probos.cognitive.standing_orders import get_department
        # AD-429e: Prefer ontology, fall back to legacy dict
        ont = getattr(rt, 'ontology', None)
        dept = (ont.get_agent_department(agent.agent_type) if ont else None) or get_department(agent.agent_type)

        channels = await rt.ward_room.list_channels()
        target_channel = None

        if dept:
            # Prefer department channel
            for ch in channels:
                if ch.channel_type == "department" and ch.department == dept:
                    target_channel = ch
                    break

        if not target_channel:
            # Fallback to All Hands (ship-wide)
            for ch in channels:
                if ch.channel_type == "ship":
                    target_channel = ch
                    break

        if not target_channel:
            logger.debug("No target channel found for proactive post from %s", agent.agent_type)
            return

        # Get callsign
        callsign = ""
        if hasattr(rt, 'callsign_registry'):
            callsign = rt.callsign_registry.get_callsign(agent.agent_type)

        # Truncate to first sentence/line for title, use full text as body
        text = _strip_bracket_markers(text)  # BF-174
        title_text = text.split('\n')[0][:100]
        if len(title_text) < len(text.split('\n')[0]):
            title_text += "..."

        _obs_thread = await rt.ward_room.create_thread(
            channel_id=target_channel.id,
            author_id=agent.id,
            title=f"[Observation] {title_text}",
            body=text,
            author_callsign=callsign or agent.agent_type,
        )
        # BF-198: Record own thread so router won't double-respond on event fan-out
        if rt.ward_room_router and _obs_thread is not None:
            _obs_tid = getattr(_obs_thread, 'id', '') or ''
            if _obs_tid:
                rt.ward_room_router.record_agent_response(agent.id, _obs_tid)

        # AD-493: Record observation fingerprint after successful posting
        if getattr(self, '_novelty_gate', None):
            try:
                self._novelty_gate.record(agent.id, text)
            except Exception:
                logger.debug("AD-493: Fingerprint recording failed", exc_info=True)

    async def _extract_and_execute_actions(
        self, agent: Any, text: str,
        *,
        post_budget: PostBudget | None = None,
    ) -> tuple[str, list[dict]]:
        """AD-437: Extract structured actions from proactive response and execute them.

        Currently supports:
        - [ENDORSE post_id UP/DOWN] — endorse a Ward Room post
        - [REPLY thread_id] ... [/REPLY] — reply to an existing thread

        Actions are gated by Earned Agency tier:
        - Ensign: no actions (can't think proactively anyway)
        - Lieutenant: endorse only
        - Commander+: endorse + reply

        Returns (cleaned_text, actions_executed).
        """
        rt = self._runtime
        if not rt or not rt.ward_room:
            return text, []

        # Determine agent's action permissions
        trust_score = rt.trust_network.get_score(agent.id)
        rank = Rank.from_trust(trust_score)
        actions_executed: list[dict] = []

        # AD-724-5: lift BF-120 markdown strip + log-only quality checks
        # (length floor, repetition, orphaned tags) into the shared DM sanity
        # gate. The gate itself never blocks; warnings log only. When the
        # runtime exposes no gate, apply_dm_sanity returns a no-op result
        # that preserves the input — Tier-2 log-and-degrade.
        _sanity = apply_dm_sanity(rt, agent.id, text)
        text = _sanity.cleaned_text

        # --- Endorsements (Lieutenant+) ---
        if rank.value != Rank.ENSIGN.value:
            cleaned, endorsements = rt.ward_room_router.extract_endorsements(text) if rt.ward_room_router else (text, [])
            if endorsements:
                await rt.ward_room_router.process_endorsements(endorsements, agent_id=agent.id)
                actions_executed.extend(
                    {"type": "endorse", "target": e["post_id"], "direction": e["direction"]}
                    for e in endorsements
                )
                text = cleaned

                # AD-428: Record exercise of Communication PCC
                if hasattr(rt, 'skill_service') and rt.skill_service:
                    try:
                        await rt.skill_service.record_exercise(agent.id, "communication")
                    except Exception:
                        logger.debug("Skill exercise recording failed", exc_info=True)

        # --- Replies (Lieutenant+) --- BF-061
        if rank.value != Rank.ENSIGN.value:
            text, reply_actions = await self._extract_and_execute_replies(
                agent, text, post_budget=post_budget,
            )
            actions_executed.extend(reply_actions)

        # --- Direct Messages --- AD-453/AD-485
        # Read configurable minimum rank (default: ensign = everyone can DM)
        dm_min_rank_str = "ensign"
        if hasattr(rt, 'config') and hasattr(rt.config, 'communications'):
            dm_min_rank_str = rt.config.communications.dm_min_rank
        dm_min_rank = Rank[dm_min_rank_str.upper()] if dm_min_rank_str.upper() in Rank.__members__ else Rank.ENSIGN
        _RANK_ORDER_DM = [Rank.ENSIGN, Rank.LIEUTENANT, Rank.COMMANDER, Rank.SENIOR]
        if _RANK_ORDER_DM.index(rank) >= _RANK_ORDER_DM.index(dm_min_rank):
            text, dm_actions = await self.extract_and_execute_dms(agent, text)
            actions_executed.extend(dm_actions)

        # --- Notebook writes (AD-434) ---
        notebook_pattern = r'\[NOTEBOOK\s+([\w-]+)\](.*?)\[/NOTEBOOK\]'
        notebook_matches = re.findall(notebook_pattern, text, re.DOTALL)
        for topic_slug, notebook_content in notebook_matches:
            notebook_content = notebook_content.strip()
            if not notebook_content or not self._runtime._records_store:
                continue
            try:
                callsign = getattr(agent, 'callsign', '') or agent.agent_type  # BF-218
                department = ""
                if self._runtime.ontology:
                    dept = self._runtime.ontology.get_agent_department(agent.agent_type)
                    if dept:
                        department = dept.department_id if hasattr(dept, 'department_id') else str(dept)

                # AD-550: Read-before-write dedup gate
                dedup_enabled = True
                dedup_threshold = 0.8
                dedup_staleness = 72.0
                dedup_max_scan = 20
                if hasattr(self._runtime, 'config') and hasattr(self._runtime.config, 'records'):
                    rc = self._runtime.config.records
                    dedup_enabled = getattr(rc, 'notebook_dedup_enabled', True)
                    dedup_threshold = getattr(rc, 'notebook_similarity_threshold', 0.8)
                    dedup_staleness = getattr(rc, 'notebook_staleness_hours', 72.0)
                    dedup_max_scan = getattr(rc, 'notebook_max_scan_entries', 20)

                if dedup_enabled:
                    try:
                        dedup_result = await self._runtime._records_store.check_notebook_similarity(
                            callsign=callsign,
                            topic_slug=topic_slug,
                            new_content=notebook_content,
                            similarity_threshold=dedup_threshold,
                            staleness_hours=dedup_staleness,
                            max_scan_entries=dedup_max_scan,
                        )
                    except Exception:
                        logger.debug("AD-550: Dedup check failed for %s/%s, writing anyway", callsign, topic_slug, exc_info=True)
                        dedup_result = {"action": "write", "reason": "dedup_check_failed", "existing_path": None, "existing_content": None, "similarity": 0.0}

                    if dedup_result["action"] == "suppress":
                        logger.info(
                            "AD-550: Notebook write suppressed for %s/%s: %s (similarity=%.2f)",
                            callsign, topic_slug, dedup_result["reason"], dedup_result["similarity"],
                        )
                        # AD-555: Record dedup suppression for quality metrics
                        _quality_engine = getattr(self._runtime, '_notebook_quality_engine', None)
                        if _quality_engine:
                            _quality_engine.record_event("dedup_suppression")
                        actions_executed.append({
                            "type": "notebook_suppressed",
                            "topic": topic_slug,
                            "callsign": callsign,
                            "reason": dedup_result["reason"],
                        })
                        continue  # Skip this notebook block

                    # AD-552: Cumulative frequency check for self-repetition
                    rep_enabled = True
                    rep_window = 48.0
                    rep_threshold = 3
                    rep_novelty = 0.2
                    rep_suppress_count = 5
                    if hasattr(self._runtime, 'config') and hasattr(self._runtime.config, 'records'):
                        rc2 = self._runtime.config.records
                        rep_enabled = getattr(rc2, 'notebook_repetition_enabled', True)
                        rep_window = getattr(rc2, 'notebook_repetition_window_hours', 48.0)
                        rep_threshold = getattr(rc2, 'notebook_repetition_threshold_count', 3)
                        rep_novelty = getattr(rc2, 'notebook_repetition_novelty_threshold', 0.2)
                        rep_suppress_count = getattr(rc2, 'notebook_repetition_suppression_count', 5)

                    if rep_enabled:
                        try:
                            _rev = dedup_result.get("revision", 0)
                            _created_iso = dedup_result.get("created_iso")
                            if _rev >= rep_threshold and _created_iso:
                                _created_ts = datetime.fromisoformat(_created_iso).timestamp()
                                _hours_active = (time.time() - _created_ts) / 3600.0
                                if _hours_active < rep_window:
                                    _novelty = 1.0 - dedup_result.get("similarity", 0.0)
                                    _suppressed = False

                                    # Suppress: high revision + low novelty
                                    if _rev >= rep_suppress_count and _novelty < rep_novelty:
                                        _suppressed = True
                                        logger.info(
                                            "AD-552: Suppressing write — %s has written %s %d times in %.1fh with <%.0f%% novel content",
                                            callsign, topic_slug, _rev, _hours_active, rep_novelty * 100,
                                        )
                                        dedup_result["action"] = "suppress"

                                    # Emit event for detection (any revision over threshold in window)
                                    if _novelty < rep_novelty or (_rev >= rep_suppress_count and _novelty < 0.3):
                                        logger.info(
                                            "AD-552: Self-repetition detected for %s on %s (revision=%d, hours_active=%.1f, novelty=%.2f)",
                                            callsign, topic_slug, _rev, _hours_active, _novelty,
                                        )
                                        if hasattr(self._runtime, 'emit_event'):
                                            from probos.events import NotebookSelfRepetitionEvent
                                            evt = NotebookSelfRepetitionEvent(
                                                agent_id=agent.id,
                                                agent_callsign=callsign,
                                                topic_slug=topic_slug,
                                                revision=_rev,
                                                hours_active=round(_hours_active, 1),
                                                novelty=round(_novelty, 2),
                                                suppressed=_suppressed,
                                            )
                                            try:
                                                self._runtime.emit_event(evt)
                                            except Exception:
                                                logger.debug("AD-552: Event emission failed", exc_info=True)
                                        # AD-555: Record repetition alert for quality metrics
                                        _quality_engine = getattr(self._runtime, '_notebook_quality_engine', None)
                                        if _quality_engine:
                                            _quality_engine.record_event("repetition_alert", callsign=callsign)

                                    if _suppressed:
                                        actions_executed.append({
                                            "type": "notebook_suppressed",
                                            "topic": topic_slug,
                                            "callsign": callsign,
                                            "reason": "self_repetition_suppressed",
                                        })
                                        continue  # Skip this notebook block
                        except Exception:
                            logger.debug("AD-552: Frequency check failed for %s/%s", callsign, topic_slug, exc_info=True)

                # AD-553: Collect metrics snapshot and compute delta
                _nb_metrics: dict[str, Any] = {}
                _metric_capture_enabled = True
                if hasattr(self._runtime, 'config') and hasattr(self._runtime.config, 'records'):
                    _metric_capture_enabled = getattr(
                        self._runtime.config.records, 'notebook_metrics_enabled', True
                    )

                if _metric_capture_enabled:
                    try:
                        _nb_metrics = collect_notebook_metrics(self._runtime, agent.id)
                        # Baseline delta: compare with previous metrics if updating
                        if dedup_result.get("action") == "update":
                            _old_metrics = dedup_result.get("existing_metrics", {})
                            if _old_metrics and _nb_metrics:
                                _nb_metrics_delta = compute_metrics_delta(_old_metrics, _nb_metrics)
                                if _nb_metrics_delta:
                                    _nb_metrics["metrics_delta"] = _nb_metrics_delta
                    except Exception:
                        logger.debug("AD-553: Metric collection failed for %s/%s", callsign, topic_slug, exc_info=True)

                await self._runtime._records_store.write_notebook(
                    callsign=callsign,
                    topic_slug=topic_slug,
                    content=notebook_content,
                    department=department,
                    tags=[topic_slug],
                    metrics=_nb_metrics if _nb_metrics else None,  # AD-553
                )
                actions_executed.append({
                    "type": "notebook_write",
                    "topic": topic_slug,
                    "callsign": callsign,
                })
                # AD-555: Record successful write for quality metrics
                _quality_engine = getattr(self._runtime, '_notebook_quality_engine', None)
                if _quality_engine:
                    _quality_engine.record_event("dedup_write")
                logger.info("Notebook entry written: %s/%s", callsign, topic_slug)

                # AD-554: Real-time cross-agent convergence/divergence detection
                _conv_enabled = True
                if hasattr(self._runtime, 'config') and hasattr(self._runtime.config, 'records'):
                    _conv_enabled = getattr(
                        self._runtime.config.records, 'realtime_convergence_enabled', True
                    )

                if _conv_enabled and department:
                    try:
                        _rc554 = getattr(self._runtime.config, 'records', None) if hasattr(self._runtime, 'config') else None
                        conv_result = await self._runtime._records_store.check_cross_agent_convergence(
                            anchor_callsign=callsign,
                            anchor_department=department,
                            anchor_topic_slug=topic_slug,
                            anchor_content=notebook_content,
                            convergence_threshold=getattr(_rc554, 'realtime_convergence_threshold', 0.5) if _rc554 else 0.5,
                            divergence_threshold=getattr(_rc554, 'realtime_divergence_threshold', 0.3) if _rc554 else 0.3,
                            staleness_hours=getattr(_rc554, 'realtime_convergence_staleness_hours', 72.0) if _rc554 else 72.0,
                            max_scan_per_agent=getattr(_rc554, 'realtime_max_scan_per_agent', 5) if _rc554 else 5,
                            min_convergence_agents=getattr(_rc554, 'realtime_min_convergence_agents', 2) if _rc554 else 2,
                            min_convergence_departments=getattr(_rc554, 'realtime_min_convergence_departments', 2) if _rc554 else 2,
                        )

                        if conv_result.get("convergence_detected"):
                            # 1. Auto-generate convergence report
                            report_path = await self._write_convergence_report(
                                conv_result, callsign, topic_slug,
                            )
                            conv_result["report_path"] = report_path or ""

                            # 2. Emit typed event
                            if hasattr(self._runtime, 'emit_event'):
                                from probos.events import ConvergenceDetectedEvent
                                evt = ConvergenceDetectedEvent(
                                    agents=conv_result["convergence_agents"],
                                    departments=conv_result["convergence_departments"],
                                    topic=conv_result.get("convergence_topic", topic_slug),
                                    coherence=conv_result.get("convergence_coherence", 0.0),
                                    source="realtime",
                                    report_path=conv_result.get("report_path", ""),
                                )
                                try:
                                    self._runtime.emit_event(evt)
                                except Exception:
                                    logger.debug("AD-554: Convergence event emission failed", exc_info=True)

                            # 3. Bridge Alert
                            await self._emit_convergence_bridge_alert(conv_result)

                            # AD-583: Wrong convergence escalation
                            if not conv_result.get("convergence_is_independent", True):
                                # Emit WrongConvergenceDetectedEvent
                                if hasattr(self._runtime, 'emit_event'):
                                    from probos.events import WrongConvergenceDetectedEvent
                                    wc_evt = WrongConvergenceDetectedEvent(
                                        agents=conv_result["convergence_agents"],
                                        departments=conv_result["convergence_departments"],
                                        topic=conv_result.get("convergence_topic", topic_slug),
                                        coherence=conv_result.get("convergence_coherence", 0.0),
                                        independence_score=conv_result.get("convergence_independence_score", 0.0),
                                        source="realtime",
                                    )
                                    try:
                                        self._runtime.emit_event(wc_evt)
                                    except Exception:
                                        logger.debug("AD-583: Wrong convergence event emission failed", exc_info=True)
                                # Bridge Alert (ALERT severity)
                                await self._emit_wrong_convergence_bridge_alert(conv_result)
                                logger.warning(
                                    "AD-583: Wrong convergence detected — %d agents, independence=%.2f",
                                    len(conv_result["convergence_agents"]),
                                    conv_result.get("convergence_independence_score", 0.0),
                                )

                            # AD-555: Record convergence for quality metrics
                            _quality_engine = getattr(self._runtime, '_notebook_quality_engine', None)
                            if _quality_engine:
                                _quality_engine.record_event(
                                    "convergence",
                                    agents=conv_result.get("convergence_agents", []),
                                )

                            logger.info(
                                "AD-554: Real-time convergence detected! %d agents from %d depts on %s",
                                len(conv_result["convergence_agents"]),
                                len(conv_result["convergence_departments"]),
                                conv_result.get("convergence_topic", topic_slug),
                            )

                        if conv_result.get("divergence_detected"):
                            # 1. Emit typed event
                            if hasattr(self._runtime, 'emit_event'):
                                from probos.events import DivergenceDetectedEvent
                                evt = DivergenceDetectedEvent(
                                    agents=conv_result["divergence_agents"],
                                    departments=conv_result["divergence_departments"],
                                    topic=conv_result.get("divergence_topic", topic_slug),
                                    similarity=conv_result.get("divergence_similarity", 0.0),
                                )
                                try:
                                    self._runtime.emit_event(evt)
                                except Exception:
                                    logger.debug("AD-554: Divergence event emission failed", exc_info=True)

                            # 2. Bridge Alert
                            await self._emit_divergence_bridge_alert(conv_result)

                            # AD-555: Record divergence for quality metrics
                            _quality_engine = getattr(self._runtime, '_notebook_quality_engine', None)
                            if _quality_engine:
                                _quality_engine.record_event("divergence")

                            logger.info(
                                "AD-554: Divergence detected! %s on %s (similarity=%.2f)",
                                ", ".join(conv_result["divergence_agents"]),
                                conv_result.get("divergence_topic", topic_slug),
                                conv_result.get("divergence_similarity", 0.0),
                            )

                    except Exception:
                        logger.debug("AD-554: Cross-agent scan failed for %s/%s", callsign, topic_slug, exc_info=True)
            except Exception as e:
                logger.warning("Notebook write failed for %s: %s", topic_slug, e)

        # BF-060: Strip ALL notebook blocks from text using regex (not string replace)
        # text.replace() failed because .strip() on captured content caused whitespace mismatch
        text = re.sub(notebook_pattern, '', text, flags=re.DOTALL).strip()

        # AD-504: [READ_NOTEBOOK topic-slug] — queue for next cycle injection
        read_nb_pattern = r'\[READ_NOTEBOOK\s+([\w-]+)\]'
        for match in re.finditer(read_nb_pattern, text):
            topic_slug = match.group(1)
            self._pending_notebook_reads[agent.id] = topic_slug
            logger.debug("Queued notebook read for %s: %s",
                        getattr(agent, 'callsign', agent.agent_type), topic_slug)
        text = re.sub(read_nb_pattern, '', text).strip()

        # --- AD-573c: Agent-writable scratchpad NOTE action tag ---
        # Mirrors notebook_pattern shape: [NOTE tag]body[/NOTE].
        # Pipes the body through working_memory.add_scratchpad() (Combo A
        # AD-573b helper) and emits WORKING_MEMORY_NOTE_RECORDED.
        note_pattern = r'\[NOTE\s+([\w-]+)\](.*?)\[/NOTE\]'
        wm = getattr(self._runtime, "working_memory", None)
        for tag, note_body in re.findall(note_pattern, text, re.DOTALL):
            note_text = (note_body or "").strip()
            if not note_text or wm is None:
                continue
            try:
                wm.add_scratchpad(note_text)
            except Exception:
                logger.warning(
                    "AD-573c: add_scratchpad failed for %s; note dropped",
                    getattr(agent, "callsign", agent.agent_type),
                    exc_info=True,
                )
                continue
            try:
                self._runtime.emit_event(
                    EventType.WORKING_MEMORY_NOTE_RECORDED,
                    {
                        "agent_id": agent.id,
                        "tag": tag,
                        "text_len": len(note_text),
                    },
                )
            except Exception:
                logger.debug(
                    "AD-573c: WORKING_MEMORY_NOTE_RECORDED emit failed",
                    exc_info=True,
                )
        text = re.sub(note_pattern, '', text, flags=re.DOTALL).strip()

        # --- Proposals (AD-412) — all ranks ---
        proposal_pattern = r'\[PROPOSAL\]\s*\n(.*?)\n\s*\[/PROPOSAL\]'
        proposal_match = re.search(proposal_pattern, text, re.DOTALL)
        if proposal_match:
            proposal_block = proposal_match.group(1)

            # Parse structured fields
            title = ""
            rationale = ""
            affected: list[str] = []
            priority = "medium"

            in_rationale = False
            rationale_lines: list[str] = []
            for line in proposal_block.split('\n'):
                stripped = line.strip()
                if stripped.lower().startswith("title:"):
                    title = stripped[6:].strip()
                    in_rationale = False
                elif stripped.lower().startswith("affected systems:"):
                    raw = stripped[17:].strip()
                    affected = [s.strip() for s in raw.split(",") if s.strip()]
                    in_rationale = False
                elif stripped.lower().startswith("priority:"):
                    p = stripped[9:].strip().lower()
                    if p in ("low", "medium", "high"):
                        priority = p
                    in_rationale = False
                elif stripped.lower().startswith("rationale:"):
                    rest = stripped[10:].strip()
                    if rest:
                        rationale_lines.append(rest)
                    in_rationale = True
                elif in_rationale:
                    rationale_lines.append(stripped)
            rationale = "\n".join(rationale_lines).strip()

            if title and rationale and rt.ward_room_router:
                try:
                    from probos.types import IntentMessage
                    intent = IntentMessage(
                        intent="propose_improvement",
                        params={
                            "title": title,
                            "rationale": rationale,
                            "affected_systems": affected,
                            "priority_suggestion": priority,
                        },
                        context=f"Proactive proposal from {getattr(agent, 'agent_type', 'unknown')}",
                    )
                    await rt.ward_room_router.handle_propose_improvement(intent, agent)
                    actions_executed.append({
                        "type": "proposal",
                        "title": title,
                        "priority": priority,
                    })
                    logger.debug(
                        "AD-412: Proposal extracted from %s: %s",
                        getattr(agent, 'agent_type', 'unknown'), title,
                    )
                except Exception:
                    logger.debug(
                        "AD-412: Failed to post proposal from %s",
                        getattr(agent, 'agent_type', 'unknown'), exc_info=True,
                    )
        elif re.search(r'\[PROPOSAL\]', text):
            # [PROPOSAL] present but didn't match strict pattern — log for debugging
            logger.debug(
                "AD-412: Malformed [PROPOSAL] block from %s — requires newlines between tags",
                getattr(agent, 'agent_type', 'unknown'),
            )

        # Strip proposal block from text regardless of submission success.
        # Uses the same strict pattern — if the regex didn't match above,
        # BF-203 catch-all below handles the orphaned [PROPOSAL] tag.
        # [/PROPOSAL] is explicitly stripped here since BF-203's regex does NOT
        # match [/UPPERCASE] closing tags.
        text = re.sub(proposal_pattern, '', text, flags=re.DOTALL).strip()
        text = re.sub(r'\[/PROPOSAL\]', '', text).strip()

        # AD-526a: Recreation actions — rank-gated via config
        rec_min_rank_str = "ensign"
        if hasattr(rt, 'config') and hasattr(rt.config, 'communications'):
            rec_min_rank_str = rt.config.communications.recreation_min_rank
        rec_min_rank = Rank[rec_min_rank_str.upper()] if rec_min_rank_str.upper() in Rank.__members__ else Rank.ENSIGN
        _RANK_ORDER_REC = [Rank.ENSIGN, Rank.LIEUTENANT, Rank.COMMANDER, Rank.SENIOR]
        if _RANK_ORDER_REC.index(rank) >= _RANK_ORDER_REC.index(rec_min_rank):
            # BF-122: Resolve callsign for this agent (was undefined — relied on
            # leaked variable from notebook loop, causing NameError when agent
            # produced [CHALLENGE] without [NOTEBOOK])
            callsign = ""
            if hasattr(rt, 'callsign_registry'):
                callsign = rt.callsign_registry.get_callsign(agent.agent_type)

            # AD-526a: [CHALLENGE @callsign game_type] — challenge another crew member to a game
            challenge_pattern = r'\[CHALLENGE\s+@(\w+)\s+(\w+)\]'
            for match in re.finditer(challenge_pattern, text):
                target_callsign = match.group(1)
                game_type = match.group(2)
                try:
                    rec_svc = getattr(rt, 'recreation_service', None)
                    if rec_svc:
                        # Resolve target callsign to agent
                        target_agent = None
                        if hasattr(rt, 'callsign_registry'):
                            target_agent = rt.callsign_registry.resolve(target_callsign)
                        if not target_agent:
                            logger.debug("AD-526a: Target callsign %s not found", target_callsign)
                            continue
                        # Create game in Recreation channel thread
                        rec_ch = None
                        if rt.ward_room:
                            channels = await rt.ward_room.list_channels()
                            rec_ch = next((c for c in channels if c.name == "Recreation"), None)
                        thread_id = ""
                        if rec_ch and rt.ward_room:
                            thread = await rt.ward_room.create_thread(
                                channel_id=rec_ch.id,
                                author_id=agent.id,
                                title=f"[Challenge] {callsign} challenges {target_callsign} to {game_type}!",
                                body=f"{callsign} has challenged {target_callsign} to a game of {game_type}! Reply to accept.",
                                author_callsign=callsign,
                            )
                            thread_id = thread.id if thread else ""
                        game_info = await rec_svc.create_game(
                            game_type=game_type,
                            challenger=callsign,
                            opponent=target_callsign,
                            thread_id=thread_id,
                        )
                        actions_executed.append({
                            "action": "challenge",
                            "target": target_callsign,
                            "game_type": game_type,
                            "game_id": game_info["game_id"],
                        })
                        logger.info("AD-526a: %s challenged %s to %s (game %s)",
                                    callsign, target_callsign, game_type, game_info["game_id"])

                        # AD-573: Register game engagement in working memory
                        try:
                            wm = getattr(agent, 'working_memory', None)
                            if wm:
                                from probos.cognitive.agent_working_memory import ActiveEngagement
                                wm.add_engagement(ActiveEngagement(
                                    engagement_type="game",
                                    engagement_id=game_info["game_id"],
                                    summary=f"Playing {game_type} against {target_callsign}",
                                    state={
                                        "game_type": game_type,
                                        "opponent": target_callsign,
                                    },
                                ))
                        except Exception:
                            logger.debug("AD-573: Working memory game engagement record failed", exc_info=True)
                except Exception as e:
                    logger.warning("AD-526a: CHALLENGE failed for %s: %s", callsign, e)
            text = re.sub(challenge_pattern, '', text).strip()

            # AD-526a: [MOVE position] — make a move in an active game
            move_pattern = r'\[MOVE\s+(\S+)\]'
            for match in re.finditer(move_pattern, text):
                position = match.group(1)
                try:
                    rec_svc = getattr(rt, 'recreation_service', None)
                    if rec_svc:
                        # AD-572: DRY — use get_game_by_player()
                        player_game = rec_svc.get_game_by_player(callsign)
                        if player_game and player_game.get("state", {}).get("current_player") == callsign:
                            game_info = await rec_svc.make_move(
                                game_id=player_game["game_id"],
                                player=callsign,
                                move=position,
                            )
                            actions_executed.append({
                                "action": "game_move",
                                "game_id": player_game["game_id"],
                                "move": position,
                            })
                            # Post board update to Recreation channel
                            if rt.ward_room and player_game.get("thread_id"):
                                board = rec_svc.render_board(player_game["game_id"]) if not game_info.get("result") else ""
                                result = game_info.get("result")
                                if result:
                                    status = result.get("status", "")
                                    winner = result.get("winner", "")
                                    body = f"Game over! {'Winner: ' + winner if winner else 'Draw!'}"
                                else:
                                    _next = game_info['state']['current_player']
                                    # BF-212: @mention next player so they receive the notification
                                    body = f"```\n{board}\n```\nYour move, @{_next}"
                                # BF-237: Enforce single-post budget
                                if post_budget is not None and post_budget.spent:
                                    logger.warning(
                                        "BF-237: Suppressing additional [MOVE] board post from %s — post budget spent",
                                        agent.agent_type,
                                    )
                                else:
                                    try:
                                        await rt.ward_room.create_post(
                                            thread_id=player_game["thread_id"],
                                            author_id=agent.id,
                                            body=body,
                                            author_callsign=callsign,
                                        )
                                        # BF-237: Mark budget as spent
                                        if post_budget is not None:
                                            post_budget.spent = True
                                    except Exception:
                                        logger.debug("AD-526a: Board update post failed", exc_info=True)

                            # BF-125: Game-over WM cleanup handled by GAME_COMPLETED subscriber.
                            # In-progress state sync handled by proactive loop (line 1139-1165).
                        else:
                            logger.debug("AD-526a: No active game for %s", callsign)
                except Exception as e:
                    logger.warning("AD-526a: MOVE failed for %s: %s", callsign, e)
            text = re.sub(move_pattern, '', text).strip()

        # --- ASSIGN (AD-654d) — Lieutenant+ only ---
        assign_min_rank = Rank.LIEUTENANT
        _RANK_ORDER_ASSIGN = [Rank.ENSIGN, Rank.LIEUTENANT, Rank.COMMANDER, Rank.SENIOR]
        can_assign = _RANK_ORDER_ASSIGN.index(rank) >= _RANK_ORDER_ASSIGN.index(assign_min_rank)

        assign_pattern = r'\[ASSIGN\s+@([\w-]+)\]\s*(.*?)\s*\[/ASSIGN\]'
        if can_assign:
            for match in re.finditer(assign_pattern, text, re.DOTALL):
                target_callsign = match.group(1)
                task_description = match.group(2).strip()
                if rt.dispatcher and rt.callsign_registry:
                    resolved = rt.callsign_registry.resolve(target_callsign)
                    target_agent_id = resolved.get("agent_id") if resolved else None
                    if target_agent_id and target_agent_id != agent.id:
                        from probos.activation import task_event_for_agent
                        event = task_event_for_agent(
                            agent_id=target_agent_id,
                            source_type="agent",
                            source_id=agent.id,
                            event_type="task_assigned",
                            priority=Priority.NORMAL,
                            payload={
                                "from_agent_id": agent.id,
                                "from_callsign": getattr(agent, 'callsign', ''),
                                "task_description": task_description,
                            },
                        )
                        await rt.dispatcher.dispatch(event)
                        actions_executed.append({"type": "assign", "target": target_callsign, "task": task_description})
                        logger.info("AD-654d: %s assigned task to @%s", agent.id[:12], target_callsign)
        text = re.sub(assign_pattern, '', text, flags=re.DOTALL).strip()

        # --- HANDOFF (AD-654d) — any rank ---
        handoff_pattern = r'\[HANDOFF\s+@([\w-]+)\]\s*(.*?)\s*\[/HANDOFF\]'
        for match in re.finditer(handoff_pattern, text, re.DOTALL):
            target_callsign = match.group(1)
            handoff_context = match.group(2).strip()
            if rt.dispatcher and rt.callsign_registry:
                resolved = rt.callsign_registry.resolve(target_callsign)
                target_agent_id = resolved.get("agent_id") if resolved else None
                if target_agent_id and target_agent_id != agent.id:
                    from probos.activation import task_event_for_agent
                    event = task_event_for_agent(
                        agent_id=target_agent_id,
                        source_type="agent",
                        source_id=agent.id,
                        event_type="task_handoff",
                        priority=Priority.CRITICAL,
                        payload={
                            "from_agent_id": agent.id,
                            "from_callsign": getattr(agent, 'callsign', ''),
                            "handoff_context": handoff_context,
                        },
                    )
                    await rt.dispatcher.dispatch(event)
                    actions_executed.append({"type": "handoff", "target": target_callsign, "context": handoff_context})
                    logger.info("AD-654d: %s handed off to @%s", agent.id[:12], target_callsign)
        text = re.sub(handoff_pattern, '', text, flags=re.DOTALL).strip()

        # BF-203: Strip unrecognized bracket command tags that the LLM hallucinated.
        # Known tags (REPLY, DM, ENDORSE, NOTEBOOK, READ_NOTEBOOK, CHALLENGE, MOVE,
        # PROPOSAL with closing [/PROPOSAL]) are extracted above. BF-203's regex does
        # NOT match [/UPPERCASE] closing-tag patterns, so any tag with a closing form
        # must be fully stripped by its handler above — including the closing tag.
        # Any remaining [UPPERCASE_COMMAND ...] patterns are hallucinations.
        text = re.sub(r'\[(?:[A-Z][A-Z_]+)(?:\s[^\]]{0,120})?\]', '', text).strip()

        return text, actions_executed

    async def _write_convergence_report(
        self, conv_result: dict, anchor_callsign: str, topic_slug: str,
    ) -> str | None:
        """AD-554: Auto-generate a convergence report in Ship's Records."""
        try:
            from uuid import uuid4 as _uuid4
            ts_slug = datetime.utcnow().strftime("%Y%m%d-%H%M%S")
            uid = str(_uuid4())[:8]
            report_path = f"reports/convergence/convergence-{ts_slug}-{uid}.md"

            agents = conv_result.get("convergence_agents", [])
            departments = conv_result.get("convergence_departments", [])
            coherence = conv_result.get("convergence_coherence", 0.0)
            topic = conv_result.get("convergence_topic", topic_slug)

            # Build perspectives from match data
            perspectives = ""
            matches = conv_result.get("convergence_matches", [])
            for m in matches:
                cs = m.get("callsign", "unknown")
                dept = m.get("department", "unknown")
                path = m.get("path", "")
                # Read content for the perspective
                snippet = ""
                if path and hasattr(self._runtime, '_records_store'):
                    try:
                        entry = await self._runtime._records_store.read_entry(
                            path, reader_id="system", reader_department="",
                        )
                        if entry:
                            snippet = entry.get("content", "")[:300].strip()
                    except Exception:
                        pass
                perspectives += f"\n### {cs} ({dept})\n\n{snippet}\n"

            report_content = (
                f"## Real-Time Convergence Report\n\n"
                f"**Detected:** {datetime.utcnow().isoformat()}Z\n\n"
                f"**Source:** Real-time notebook monitor\n\n"
                f"**Topic:** {topic}\n\n"
                f"**Agents:** {', '.join(sorted(agents))}\n\n"
                f"**Departments:** {', '.join(sorted(departments))}\n\n"
                f"**Coherence:** {coherence:.3f}\n\n"
                f"## Contributing Perspectives\n{perspectives}\n"
            )

            await self._runtime._records_store.write_entry(
                author="system",
                path=report_path,
                content=report_content,
                message=f"AD-554: Real-time convergence report ({topic})",
                classification="ship",
                tags=["convergence", "ad-554", "realtime"],
            )
            return report_path
        except Exception:
            logger.debug("AD-554: Failed to write convergence report", exc_info=True)
            return None

    async def _emit_convergence_bridge_alert(self, conv_result: dict) -> None:
        """AD-554: Create and deliver a convergence BridgeAlert."""
        rt = self._runtime
        if not rt or not hasattr(rt, 'bridge_alerts') or not rt.bridge_alerts:
            return
        if not hasattr(rt, 'ward_room_router') or not rt.ward_room_router:
            return

        alerts = rt.bridge_alerts.check_realtime_convergence(conv_result)
        for alert in alerts:
            try:
                await rt.ward_room_router.deliver_bridge_alert(alert)
            except Exception:
                logger.debug("AD-554: Bridge alert delivery failed", exc_info=True)

    async def _emit_wrong_convergence_bridge_alert(self, conv_result: dict) -> None:
        """AD-583: Create and deliver a wrong convergence BridgeAlert."""
        rt = self._runtime
        if not rt or not hasattr(rt, 'bridge_alerts') or not rt.bridge_alerts:
            return
        if not hasattr(rt, 'ward_room_router') or not rt.ward_room_router:
            return

        alerts = rt.bridge_alerts.check_wrong_convergence(conv_result)
        for alert in alerts:
            try:
                await rt.ward_room_router.deliver_bridge_alert(alert)
            except Exception:
                logger.debug("AD-583: Wrong convergence alert delivery failed", exc_info=True)

    async def _emit_divergence_bridge_alert(self, conv_result: dict) -> None:
        """AD-554: Create and deliver a divergence BridgeAlert."""
        rt = self._runtime
        if not rt or not hasattr(rt, 'bridge_alerts') or not rt.bridge_alerts:
            return
        if not hasattr(rt, 'ward_room_router') or not rt.ward_room_router:
            return

        alerts = rt.bridge_alerts.check_divergence(conv_result)
        for alert in alerts:
            try:
                await rt.ward_room_router.deliver_bridge_alert(alert)
            except Exception:
                logger.debug("AD-554: Divergence bridge alert delivery failed", exc_info=True)

    async def _resolve_thread_id(self, thread_id: str) -> str | None:
        """BF-061: Resolve a full or partial thread ID to an actual thread ID."""
        rt = self._runtime
        # Try exact match first
        thread = await rt.ward_room.get_thread(thread_id)
        if thread:
            return thread_id
        # Try prefix match on recent threads across channels
        if len(thread_id) < 36:
            channels = await rt.ward_room.list_channels()
            for ch in channels:
                try:
                    activity = await rt.ward_room.get_recent_activity(ch.id, limit=20, since=None)
                    for item in activity:
                        tid = item.get("thread_id", "") or item.get("id", "")
                        if tid and tid.startswith(thread_id):
                            return tid
                except Exception:
                    logger.debug("Skipping channel", exc_info=True)
                    continue
        return None

    async def _extract_and_execute_replies(
        self, agent: Any, text: str,
        *,
        post_budget: PostBudget | None = None,
    ) -> tuple[str, list[dict]]:
        """AD-437: Extract [REPLY thread_id]...[/REPLY] blocks and post as replies.

        Allows Lieutenant+ agents to reply to existing threads instead of
        always creating new threads for every observation.
        """
        import re
        rt = self._runtime
        actions: list[dict] = []

        # BF-061: More flexible pattern — optional thread: prefix, no mandatory newline
        pattern = re.compile(
            r'\[REPLY\s+(?:thread:?\s*)?(\S+)\]\s*(.*?)\s*\[/REPLY\]',
            re.DOTALL | re.IGNORECASE,
        )

        for match in pattern.finditer(text):
            thread_id = match.group(1)
            reply_body = match.group(2).strip()
            if not reply_body:
                continue

            try:
                # BF-061: Resolve thread ID (may be partial or prefixed)
                resolved_id = await self._resolve_thread_id(thread_id)
                if not resolved_id:
                    logger.debug("AD-437: Reply target thread %s not found", thread_id)
                    continue
                thread_id = resolved_id
                thread = await rt.ward_room.get_thread(thread_id)

                # Check thread isn't locked
                thread_data = thread.get("thread", thread)
                if thread_data.get("locked"):
                    logger.debug("AD-437: Reply target thread %s is locked", thread_id)
                    continue

                # BF-105: Self-similarity guard for replies (mirrors BF-032 for new threads)
                if await self._is_similar_to_recent_posts(agent, reply_body):
                    logger.debug(
                        "BF-105: Suppressed similar reply from %s to thread %s",
                        agent.agent_type, thread_id[:8],
                    )
                    continue

                # BF-171 + AD-625: Per-agent per-channel reply cooldown (proficiency-modulated)
                channel_id = thread_data.get("channel_id", "")
                if channel_id:
                    _reply_cd = 120  # Default (Competent tier)
                    _overrides = self._get_comm_gate_overrides(agent)
                    if _overrides is not None:
                        _reply_cd = _overrides.reply_cooldown_seconds
                    cooldown_key = f"{agent.id}:{channel_id}"
                    last_reply = self._reply_cooldowns.get(cooldown_key, 0.0)
                    if last_reply > 0 and (time.monotonic() - last_reply) < _reply_cd:
                        logger.debug(
                            "BF-171: Reply cooldown for %s in channel %s (%.0fs remaining)",
                            agent.agent_type, channel_id[:8],
                            _reply_cd - (time.monotonic() - last_reply),
                        )
                        continue

                # BF-201: Thread post cap — proactive path
                wr_router = getattr(rt, 'ward_room_router', None)
                if wr_router and thread_id:
                    try:
                        _td = await rt.ward_room.get_thread(thread_id)
                        if _td:
                            _max = getattr(rt.config.ward_room, 'max_thread_posts', 50)
                            if len(_td.get("posts", [])) >= _max:
                                logger.debug("BF-201: Thread %s at post cap, skipping proactive reply", thread_id[:8])
                                continue
                    except Exception:
                        pass  # Safe default: allow reply

                # Get callsign
                callsign = ""
                if hasattr(rt, 'callsign_registry'):
                    callsign = rt.callsign_registry.get_callsign(agent.agent_type)

                # BF-121: Extract CHALLENGE/MOVE from reply body before posting.
                # Agents wrap structured commands inside [REPLY] blocks when
                # responding to threads. Without this, commands are posted as
                # text but never parsed — the REPLY block is stripped from
                # `text` before the command regexes run.
                reply_body, reply_cmd_actions = await self._extract_commands_from_reply(
                    agent, reply_body, callsign,
                )
                # AD-724-5: run reply body through the shared sanity gate so
                # WR replies get the same orphaned-tag / repetition /
                # length-floor visibility as DM one-shots.
                _reply_sanity = apply_dm_sanity(rt, agent.id, reply_body)
                reply_body = _reply_sanity.cleaned_text
                reply_body = _strip_bracket_markers(reply_body)  # BF-174
                if not reply_body:
                    continue

                # BF-237: Enforce single-post budget per pipeline invocation.
                # Check BEFORE create_post so the second [REPLY] match in
                # a multi-REPLY LLM response is suppressed, not posted.
                if post_budget is not None and post_budget.spent:
                    logger.warning(
                        "BF-237: Suppressing additional [REPLY] from %s — post budget spent",
                        agent.agent_type,
                    )
                    continue

                await rt.ward_room.create_post(
                    thread_id=thread_id,
                    author_id=agent.id,
                    body=reply_body,
                    author_callsign=callsign or agent.agent_type,
                )
                # BF-237: Mark budget as spent after successful post
                if post_budget is not None:
                    post_budget.spent = True
                # BF-198: Record reply so router won't double-respond
                if rt.ward_room_router:
                    rt.ward_room_router.record_agent_response(agent.id, thread_id)
                # AD-638: Track post for boot camp graduation
                if hasattr(rt, 'boot_camp') and rt.boot_camp and rt.boot_camp.is_enrolled(agent.id):
                    try:
                        _ch_type = ""
                        if channel_id:
                            try:
                                _ch_obj = await rt.ward_room.get_channel(channel_id)
                                _ch_type = getattr(_ch_obj, 'channel_type', '')
                            except Exception:
                                pass
                        await rt.boot_camp.on_agent_post(agent.id, _ch_type)
                    except Exception:
                        logger.debug("AD-638: Boot camp post tracking failed", exc_info=True)
                # BF-171: Record reply timestamp for channel cooldown
                if channel_id:
                    self._reply_cooldowns[f"{agent.id}:{channel_id}"] = time.monotonic()
                actions.append({
                    "type": "reply",
                    "thread_id": thread_id,
                    "length": len(reply_body),
                })
                actions.extend(reply_cmd_actions)
                logger.debug(
                    "AD-437: %s replied to thread %s (%d chars)",
                    agent.agent_type, thread_id, len(reply_body),
                )
            except Exception:
                logger.debug(
                    "AD-437: Reply to thread %s failed for %s",
                    thread_id, agent.agent_type, exc_info=True,
                )

        # Strip all [REPLY]...[/REPLY] blocks from text
        cleaned = pattern.sub('', text).strip()
        return cleaned, actions

    async def _extract_commands_from_reply(
        self, agent: Any, reply_body: str, callsign: str,
    ) -> tuple[str, list[dict]]:
        """BF-121: Extract CHALLENGE/MOVE structured commands from reply body.

        When agents wrap their output in [REPLY] blocks, structured commands
        inside the reply are posted as text but never parsed — the reply
        extraction strips them from the main text before command regexes run.
        This extracts and executes commands from the reply body, returning
        the cleaned body text and any actions taken.
        """
        rt = self._runtime
        actions: list[dict] = []

        # BF-120: Strip markdown from reply body too
        reply_body = re.sub(r'[`*]{1,3}\[', '[', reply_body)
        reply_body = re.sub(r'\][`*]{1,3}', ']', reply_body)

        # --- CHALLENGE ---
        challenge_pattern = r'\[CHALLENGE\s+@(\w+)\s+(\w+)\]'
        for match in re.finditer(challenge_pattern, reply_body):
            target_callsign = match.group(1)
            game_type = match.group(2)
            try:
                rec_svc = getattr(rt, 'recreation_service', None)
                if not rec_svc:
                    continue
                target_agent = None
                if hasattr(rt, 'callsign_registry'):
                    target_agent = rt.callsign_registry.resolve(target_callsign)
                if not target_agent:
                    logger.debug("BF-121: Target callsign %s not found in reply", target_callsign)
                    continue
                # Create Recreation channel thread
                rec_ch = None
                if rt.ward_room:
                    channels = await rt.ward_room.list_channels()
                    rec_ch = next((c for c in channels if c.name == "Recreation"), None)
                thread_id = ""
                if rec_ch and rt.ward_room:
                    thread = await rt.ward_room.create_thread(
                        channel_id=rec_ch.id,
                        author_id=agent.id,
                        title=f"[Challenge] {callsign} challenges {target_callsign} to {game_type}!",
                        body=f"{callsign} has challenged {target_callsign} to a game of {game_type}! Reply to accept.",
                        author_callsign=callsign,
                    )
                    thread_id = thread.id if thread else ""
                game_info = await rec_svc.create_game(
                    game_type=game_type,
                    challenger=callsign,
                    opponent=target_callsign,
                    thread_id=thread_id,
                )
                actions.append({
                    "action": "challenge",
                    "target": target_callsign,
                    "game_type": game_type,
                    "game_id": game_info["game_id"],
                })
                logger.info("BF-121: %s challenged %s to %s via reply (game %s)",
                            callsign, target_callsign, game_type, game_info["game_id"])

                # Register game engagement in working memory
                try:
                    wm = getattr(agent, 'working_memory', None)
                    if wm:
                        from probos.cognitive.agent_working_memory import ActiveEngagement
                        wm.add_engagement(ActiveEngagement(
                            engagement_type="game",
                            engagement_id=game_info["game_id"],
                            summary=f"Playing {game_type} against {target_callsign}",
                            state={
                                "game_type": game_type,
                                "opponent": target_callsign,
                            },
                        ))
                except Exception:
                    logger.debug("BF-121: Working memory game engagement failed", exc_info=True)
            except Exception as e:
                logger.warning("BF-121: CHALLENGE in reply failed for %s: %s", callsign, e)
        reply_body = re.sub(challenge_pattern, '', reply_body).strip()

        # --- MOVE ---
        move_pattern = r'\[MOVE\s+(\S+)\]'
        for match in re.finditer(move_pattern, reply_body):
            position = match.group(1)
            try:
                rec_svc = getattr(rt, 'recreation_service', None)
                if rec_svc:
                    player_game = rec_svc.get_game_by_player(callsign)
                    if player_game and player_game.get("state", {}).get("current_player") == callsign:
                        game_info = await rec_svc.make_move(
                            game_id=player_game["game_id"],
                            player=callsign,
                            move=position,
                        )
                        actions.append({
                            "action": "move",
                            "position": position,
                            "game_id": player_game["game_id"],
                        })
                        logger.info("BF-121: %s played move %s via reply (game %s)",
                                    callsign, position, player_game["game_id"])
            except Exception as e:
                logger.warning("BF-121: MOVE in reply failed for %s: %s", callsign, e)
        reply_body = re.sub(move_pattern, '', reply_body).strip()

        return reply_body, actions

    async def extract_and_execute_dms(
        self, agent: Any, text: str,
    ) -> tuple[str, list[dict]]:
        """AD-453: Extract [DM @callsign]...[/DM] blocks and send as DMs."""
        import re
        rt = self._runtime
        actions: list[dict] = []

        # AD-612: Two-tier DM extraction — tolerant of format variations.
        # Tier 1+2 (unified): Closed DMs with any whitespace (multiline or single-line)
        pattern = re.compile(
            r'\[DM\s+@?(\S+)\]'        # Opening tag, capture callsign
            r'\s*'                      # Optional whitespace (including newlines)
            r'((?:(?!\[DM\s).)*?)'      # Body (non-greedy, can't cross [DM boundary)
            r'\[/DM\]'                  # Closing tag
            ,
            re.DOTALL | re.IGNORECASE,
        )

        # Tier 3: Unclosed DMs — match [DM @callsign] text to next [DM or end
        unclosed_pattern = re.compile(
            r'\[DM\s+@?(\S+)\]'        # Opening tag, capture callsign
            r'\s*'                      # Optional whitespace
            r'(.+?)'                    # Body (non-greedy, at least 1 char)
            r'(?=\[DM\s|\Z)'           # Lookahead: next [DM tag or end of string
            ,
            re.DOTALL | re.IGNORECASE,
        )

        # Collect all DM matches (closed first, then unclosed from remaining text)
        all_dm_matches: list[tuple[str, str]] = []  # (callsign, body)
        for match in pattern.finditer(text):
            all_dm_matches.append((match.group(1), match.group(2).strip()))
        remaining_after_closed = pattern.sub('', text)
        for match in unclosed_pattern.finditer(remaining_after_closed):
            all_dm_matches.append((match.group(1), match.group(2).strip()))

        for target_callsign, dm_body in all_dm_matches:
            if not dm_body:
                continue

            # BF-163: Per-agent per-target DM send cooldown (60s).
            # Prevents DM flood loops where Agent A DMs Agent B, B responds,
            # A gets notified, A DMs again — overwhelming the LLM proxy.
            dm_pair_key = f"{agent.id}:{target_callsign.lower()}"
            now = time.monotonic()
            last_dm_send = self._dm_send_cooldowns.get(dm_pair_key, 0.0)
            if now - last_dm_send < 60.0:
                logger.debug(
                    "BF-163: %s DM to @%s throttled (%.0fs remaining)",
                    getattr(agent, 'callsign', agent.agent_type),
                    target_callsign,
                    60.0 - (now - last_dm_send),
                )
                continue
            self._dm_send_cooldowns[dm_pair_key] = now

            # AD-614: Self-similarity gate — suppress near-duplicate DMs.
            from probos.cognitive.similarity import jaccard_similarity
            last_body = self._last_dm_body.get(dm_pair_key, "")
            if last_body:
                sim = jaccard_similarity(
                    set(dm_body.lower().split()),
                    set(last_body.lower().split()),
                )
                if sim >= 0.6:
                    logger.debug(
                        "AD-614: %s DM to @%s suppressed (similarity %.2f)",
                        getattr(agent, 'callsign', agent.agent_type),
                        target_callsign,
                        sim,
                    )
                    continue
            self._last_dm_body[dm_pair_key] = dm_body

            # AD-485: Special case — DM to Captain
            if target_callsign.lower() == "captain":
                try:
                    sender_callsign = ""
                    if hasattr(rt, 'callsign_registry'):
                        sender_callsign = rt.callsign_registry.get_callsign(agent.agent_type)

                    captain_channel_name = f"dm-captain-{agent.id[:8]}"
                    dm_channel = None
                    channels = await rt.ward_room.list_channels()
                    for ch in channels:
                        if ch.name == captain_channel_name and ch.channel_type == "dm":
                            dm_channel = ch
                            break
                    if not dm_channel:
                        dm_channel = await rt.ward_room.create_channel(
                            name=captain_channel_name,
                            description=f"DM: {sender_callsign or agent.agent_type} → Captain",
                            channel_type="dm",
                            created_by=agent.id,
                        )

                    await rt.ward_room.create_thread(
                        channel_id=dm_channel.id,
                        author_id=agent.id,
                        title=f"[DM to Captain from @{sender_callsign or agent.agent_type}]",
                        body=dm_body,
                        author_callsign=sender_callsign or agent.agent_type,
                    )

                    actions.append({"type": "dm", "target_callsign": "captain", "target_agent_id": "captain"})
                    logger.info("AD-485: %s sent DM to Captain", sender_callsign or agent.agent_type)
                except Exception as e:
                    logger.warning("AD-485: DM to Captain failed: %s", e)
                continue

            # Resolve callsign to agent_type
            target_agent_type = None
            if hasattr(rt, 'callsign_registry'):
                resolved = rt.callsign_registry.resolve(target_callsign)
                if resolved:
                    target_agent_type = resolved.get("agent_type")
            if not target_agent_type:
                logger.debug("AD-453: DM target @%s not found in registry", target_callsign)
                continue

            # Don't DM yourself
            if target_agent_type == agent.agent_type or target_agent_type == agent.id:
                continue

            # Resolve target's full agent ID
            target_full_id = None
            for a in rt.registry.all():
                if a.agent_type == target_agent_type:
                    target_full_id = a.id
                    break
            if not target_full_id:
                continue

            try:
                sender_callsign = ""
                if hasattr(rt, 'callsign_registry'):
                    sender_callsign = rt.callsign_registry.get_callsign(agent.agent_type)

                dm_channel = await rt.ward_room.get_or_create_dm_channel(
                    agent.id, target_full_id,
                    callsign_a=sender_callsign or agent.agent_type,
                    callsign_b=target_callsign,
                )

                await rt.ward_room.create_thread(
                    channel_id=dm_channel.id,
                    author_id=agent.id,
                    title=f"[DM to @{target_callsign}]",
                    body=dm_body,
                    author_callsign=sender_callsign or agent.agent_type,
                )

                actions.append({
                    "type": "dm",
                    "target_callsign": target_callsign,
                    "target_agent_id": target_full_id,
                })
                logger.info("AD-453: %s sent DM to @%s", sender_callsign or agent.agent_type, target_callsign)

                # Record Hebbian social connection
                if hasattr(rt, 'hebbian_router') and rt.hebbian_router:
                    from probos.mesh.routing import REL_SOCIAL
                    rt.hebbian_router.record_interaction(
                        source=agent.id, target=target_full_id,
                        success=True, rel_type=REL_SOCIAL,
                    )
                    rt.emit_event(EventType.HEBBIAN_UPDATE, {
                        "source": agent.id, "target": target_full_id,
                        "weight": format_trust(rt.hebbian_router.get_weight(agent.id, target_full_id)),
                        "rel_type": "social",
                    })
            except Exception as e:
                logger.warning("AD-453: DM to @%s failed: %s", target_callsign, e)

        # AD-612: Strip all matched DM blocks from public text
        cleaned = pattern.sub('', text).strip()
        cleaned = unclosed_pattern.sub('', cleaned).strip()
        return cleaned, actions

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

    def set_knowledge_store(self, store) -> None:
        """Inject knowledge store for cooldown persistence."""
        self._knowledge_store = store

    def get_cooldowns(self) -> dict:
        """Return a copy of per-agent cooldown data for persistence."""
        return dict(self._agent_cooldowns)

    @property
    def llm_failure_count(self) -> int:
        """BF-069: Number of consecutive proactive loop failures."""
        return self._llm_failure_count

    async def _update_llm_status(self, failure: bool) -> None:
        """AD-576: Update LLM status state machine and emit events on transitions.

        State transitions (BF-228: hysteresis — require 3+ failures before degraded):
            operational -> degraded (failure_count >= 3, matches _UNREACHABLE_THRESHOLD)
            degraded -> offline (failure_count >= 6, sustained degradation)
            any non-operational -> operational (failure_count reaches 0 via decay)
        """
        old_status = self._llm_status

        if failure:
            if self._llm_failure_count >= 6:
                new_status = "offline"
            elif self._llm_failure_count >= 3:
                new_status = "degraded"
            else:
                new_status = old_status  # BF-228: stay in current state below threshold
            if self._llm_offline_since == 0.0 and new_status != "operational":
                self._llm_offline_since = time.monotonic()
        else:
            new_status = "operational" if self._llm_failure_count == 0 else old_status
            # Don't reset _llm_offline_since yet — bridge alert needs it for downtime calc

        if old_status == new_status:
            return

        self._llm_status = new_status

        # Emit typed event on transition
        if self._on_event:
            from probos.events import LlmHealthChangedEvent
            downtime = (time.monotonic() - self._llm_offline_since) if self._llm_offline_since else 0.0
            # BF-240: Get max consecutive_successes across tiers from the LLM client
            cs = 0
            if hasattr(self, "_llm_client") and self._llm_client:
                try:
                    health = self._llm_client.get_health_status()
                    cs = max(
                        t.get("consecutive_successes", 0)
                        for t in health.get("tiers", {}).values()
                    ) if health.get("tiers") else 0
                except Exception:
                    pass  # Log-and-degrade: event still emits with cs=0
            try:
                event = LlmHealthChangedEvent(
                    old_status=old_status,
                    new_status=new_status,
                    consecutive_failures=self._llm_failure_count,
                    consecutive_successes=cs,  # BF-240
                    downtime_seconds=downtime if new_status == "operational" else 0.0,
                )
                self._on_event(event.to_dict())
            except Exception:
                logger.debug("AD-576: LLM health event emission failed", exc_info=True)

        # Emit Bridge Alerts on transition
        await self._emit_llm_status_bridge_alert(old_status, new_status)

        # Now safe to reset offline_since after alert delivery
        if not failure:
            self._llm_offline_since = 0.0

    async def _emit_llm_status_bridge_alert(self, old_status: str, new_status: str) -> None:
        """AD-576: Emit Bridge Alert on LLM status transitions."""
        rt = self._runtime
        if not rt or not hasattr(rt, 'bridge_alerts') or not rt.bridge_alerts:
            return
        if not hasattr(rt, 'ward_room_router') or not rt.ward_room_router:
            return

        from probos.bridge_alerts import AlertSeverity, BridgeAlert

        downtime_str = ""
        if new_status == "operational" and self._llm_offline_since:
            elapsed = time.monotonic() - self._llm_offline_since
            minutes = int(elapsed // 60)
            seconds = int(elapsed % 60)
            downtime_str = f" ({minutes}m {seconds}s total downtime)" if minutes else f" ({seconds}s total downtime)"

        alert_map = {
            "offline": BridgeAlert(
                id=f"llm_status_{int(time.time())}",
                severity=AlertSeverity.ALERT,
                source="proactive_loop",
                alert_type="llm_offline",
                title="Communications Array Offline",
                detail=(
                    "The LLM backend is unreachable. All crew cognitive functions are"
                    " suspended. Proactive cycles will continue but produce no output"
                    " until the array is restored. This is an infrastructure issue —"
                    " not a crew performance concern."
                ),
                department=None,
                dedup_key="llm_offline",
            ),
            "degraded": BridgeAlert(
                id=f"llm_status_{int(time.time())}",
                severity=AlertSeverity.ADVISORY,
                source="proactive_loop",
                alert_type="llm_degraded",
                title="Communications Array Degraded",
                detail=(
                    "The LLM backend is experiencing intermittent connectivity."
                    f" {self._llm_failure_count} consecutive failures."
                    " Crew cognition may be temporarily impaired."
                    " This is an infrastructure issue — not a crew performance concern."
                ),
                department=None,
                dedup_key="llm_degraded",
            ),
            "operational": BridgeAlert(
                id=f"llm_status_{int(time.time())}",
                severity=AlertSeverity.INFO,
                source="proactive_loop",
                alert_type="llm_restored",
                title="Communications Array Restored",
                detail=(
                    "LLM backend connectivity has been re-established."
                    " All crew cognitive functions resuming normal operations."
                    + downtime_str
                ),
                department=None,
                dedup_key="llm_restored",
            ),
        }

        alert = alert_map.get(new_status)
        if not alert:
            return

        try:
            await rt.ward_room_router.deliver_bridge_alert(alert)
        except Exception:
            logger.debug("AD-576: LLM status bridge alert delivery failed", exc_info=True)
