"""User ego context builder — assembles user-focused operational briefing.

The user ego sees the world through the user's eyes: their conversations,
their interests, their pending work, and what Genesis can do for them.
Genesis infrastructure is intentionally excluded — system issues reach
the user ego ONLY via genesis ego escalations.

This is an automated prompt generator — it builds the prompt the user
WOULD give if they were sitting here. The quality of this prompt
determines the quality of the ego's autocomplete.
"""

from __future__ import annotations

import json
import logging
from datetime import UTC, datetime
from pathlib import Path
from typing import Any

import aiosqlite

from genesis.ego.types import NEUTRAL_STATUS

logger = logging.getLogger(__name__)

# Observation categories that represent user-world signals.
# Used by GenesisEgoContextBuilder to EXCLUDE these from its system-focused view.
_USER_WORLD_CATEGORIES = frozenset({
    "email_recon", "inbox", "finding", "interest", "interests",
    "contribution", "user_model_delta",
})

# Keywords indicating a session is about Genesis internals (not user-world).
# Used to annotate conversation transcripts so the user ego knows which
# sessions fall under the Genesis ego's jurisdiction.
_GENESIS_INTERNAL_KEYWORDS = frozenset({
    "surplus", "dream cycle", "dream_cycle", "genesis runtime",
    "routing config", "circuit breaker", "guardian", "sentinel", "qdrant",
    "awareness loop", "health check", "dead letter", "model_routing",
    "worktree", "genesis-development", "dashboard fix", "ego cycle",
    "model eval", "surplus_task", "provider fallback", "watchdog",
    "systemd", "genesis server", "eval batch", "j9 eval",
    "runtime init", "embedding chain", "embedding fallback",
})


def _is_genesis_internal(topic: str) -> bool:
    """Check if a session topic is about Genesis internals."""
    topic_lower = topic.lower()
    return any(kw in topic_lower for kw in _GENESIS_INTERNAL_KEYWORDS)


class UserEgoContextBuilder:
    """Builds the user-focused operational briefing for the user ego.

    Current-state only — no historical lookups. The ego uses
    memory_recall on demand for historical context.

    Data sources (ordered by signal value):
    1. User model — who the user is, what they care about
    2. User activity pulse — user-facing awareness signals
    3. Recent conversations — what the user is working on
    4. User-world observations — email, inbox, findings
    5. Genesis ego escalations — things Genesis can't resolve alone
    6. Capabilities — what Genesis CAN do but ISN'T doing
    7. Open threads — pending follow-ups

    NOTE: System health is intentionally EXCLUDED. The user ego has no
    jurisdiction over Genesis infrastructure. System issues reach it ONLY
    via genesis ego escalations (source 5 above).
    """

    def __init__(
        self,
        *,
        db: aiosqlite.Connection,
        health_data: Any | None = None,  # Retained for API compat; unused
        capabilities: dict[str, str] | None = None,
    ) -> None:
        self._db = db
        self._capabilities = capabilities or {}

    async def build(self) -> str:
        """Assemble the full user ego context."""
        sections: list[str] = []
        sections.append("# USER_EGO_CONTEXT — What Does the User Need?\n")
        sections.append(
            "*You are Genesis's user-facing intelligence. "
            "This is your user's world. What would they ask you "
            "to do if they were here right now?*\n"
        )

        sections.append(await self._user_model_section())
        sections.append(self._ego_notepad_section())
        sections.append(await self._user_goals_section())
        sections.append(await self._user_directives_section())
        sections.append(await self._world_snapshot_section())
        sections.append(await self._user_activity_pulse_section())
        sections.append(await self._recent_conversations_section())
        sections.append(await self._backlog_summary_section())
        sections.append(await self._genesis_escalations_section())
        sections.append(await self._capabilities_section())
        # System status removed — user ego has no jurisdiction over Genesis
        # health.  System issues reach it ONLY via genesis ego escalations.
        sections.append(await self._follow_ups_section())
        sections.append(await self._proposal_history_section())
        sections.append(await self._proposal_board_section())
        sections.append(await self._execution_outcomes_section())
        sections.append(await self._goal_progress_section())
        sections.append(await self._capability_performance_section())
        sections.append(await self._autonomy_readiness_section())
        sections.append(await self._recurring_patterns_section())
        sections.append(self._output_contract_section())

        return "\n".join(sections)

    # -- Section builders --

    async def _user_model_section(self) -> str:
        """Who the user is — from user_model_cache."""
        lines = ["## User Profile\n"]

        try:
            cursor = await self._db.execute(
                "SELECT model_json, version, synthesized_at, evidence_count "
                "FROM user_model_cache WHERE id = 'current'"
            )
            row = await cursor.fetchone()
        except Exception:
            logger.error("Failed to query user_model_cache", exc_info=True)
            lines.append("*User model not available.*\n")
            return "\n".join(lines)

        if not row:
            lines.append("*No user model synthesized yet.*\n")
            return "\n".join(lines)

        model_json, version, synthesized_at, evidence_count = row
        lines.append(
            f"*v{version}, {evidence_count} evidence points, "
            f"last synthesized {synthesized_at[:10]}*\n"
        )

        # 3d: Model freshness warning — flag when model is stale
        # relative to recent conversation activity.
        try:
            synth_dt = datetime.fromisoformat(synthesized_at)
            if synth_dt.tzinfo is None:
                synth_dt = synth_dt.replace(tzinfo=UTC)
            age_days = (datetime.now(UTC) - synth_dt).days
            cursor = await self._db.execute(
                "SELECT COUNT(*) FROM cc_sessions "
                "WHERE source_tag = 'foreground' "
                "AND started_at > ?",
                (synthesized_at,),
            )
            freshness_row = await cursor.fetchone()
            sessions_since = freshness_row[0] if freshness_row else 0
            if age_days >= 3 and sessions_since >= 3:
                lines.append(
                    f"**Warning: Model may be stale**: {age_days}d old, "
                    f"{sessions_since} conversations since last synthesis\n"
                )
        except Exception:
            pass  # Non-critical enrichment

        try:
            model = json.loads(model_json) if model_json else {}
        except (json.JSONDecodeError, TypeError):
            model = {}

        if not model:
            lines.append("*Model data empty.*\n")
            return "\n".join(lines)

        # Extract the most actionable fields for ego decision-making
        priority_keys = [
            "active_projects", "current_focus", "priorities",
            "goals", "professional_role", "expertise_areas",
            "communication_preferences", "autonomy_preferences",
            "decision_making_style", "interests", "active_investigations",
            "binding_constraints", "bottleneck_awareness",
        ]

        for key in priority_keys:
            if key in model:
                val = model[key]
                if isinstance(val, str) and len(val) > 300:
                    val = val[:300] + "..."
                elif isinstance(val, (list, dict)):
                    val = json.dumps(val, default=str)
                    if len(val) > 300:
                        val = val[:300] + "..."
                lines.append(f"- **{key}**: {val}")

        # Count remaining fields the ego can explore via MCP
        shown = len([k for k in priority_keys if k in model])
        remaining = len(model) - shown
        if remaining > 0:
            lines.append(
                f"\n*{remaining} more fields available via memory_recall.*"
            )

        lines.append("")
        return "\n".join(lines)

    # -- Ego notepad (persistent qualitative observations) --

    _NOTEPAD_PATH = Path(__file__).resolve().parent.parent / "identity" / "EGO_NOTEPAD.md"
    _NOTEPAD_MARKER = "# Ego Notepad"

    def _ego_notepad_section(self) -> str:
        """Inject the ego's persistent notepad into context."""
        try:
            if not self._NOTEPAD_PATH.exists():
                return ""
            content = self._NOTEPAD_PATH.read_text().strip()
            if not content or self._NOTEPAD_MARKER not in content:
                return ""
            return f"## Your Knowledge Notepad\n\n{content}\n"
        except Exception:
            logger.warning("Failed to read ego notepad", exc_info=True)
            return ""

    async def _user_goals_section(self) -> str:
        """Active user goals — the bedrock of the world model."""
        lines = ["## User Goals\n"]

        try:
            from genesis.db.crud import user_goals
            goals = await user_goals.list_active(self._db, limit=10)
        except Exception:
            logger.debug("Failed to query user goals", exc_info=True)
            lines.append("*Goal tracking not yet populated.*\n")
            return "\n".join(lines)

        if not goals:
            lines.append(
                "*No goals tracked yet. Goals will be detected from "
                "conversations and stored automatically.*\n"
            )
            return "\n".join(lines)

        from datetime import UTC, datetime

        from genesis.db.crud import ego as ego_crud

        now = datetime.now(UTC)

        for g in goals:
            goal_id = g.get("id", "?")
            priority = g.get("priority", "medium")
            title = g.get("title", "?")[:120]
            category = g.get("category", "")

            # Staleness: days since updated_at
            updated_at = g.get("updated_at") or g.get("created_at") or ""
            stale_str = ""
            if updated_at:
                try:
                    updated = datetime.fromisoformat(updated_at)
                    days_since = (now - updated).days
                    if days_since >= 7:
                        stale_str = f" STALE ({days_since}d)"
                except (ValueError, TypeError):
                    pass

            # Latest progress note
            progress_str = ""
            progress_notes_raw = g.get("progress_notes", "[]")
            try:
                import json
                notes = json.loads(progress_notes_raw) if isinstance(progress_notes_raw, str) else progress_notes_raw
                if notes and isinstance(notes, list):
                    latest = notes[-1]
                    note_text = latest.get("note", str(latest)) if isinstance(latest, dict) else str(latest)
                    progress_str = f' | Last: "{note_text[:60]}"'
            except (json.JSONDecodeError, TypeError):
                pass

            # Proposal outcome summary
            proposal_str = ""
            try:
                summary = await ego_crud.get_goal_proposal_summary(self._db, goal_id)
                if summary:
                    parts = [f"{count} {status}" for status, count in summary.items()]
                    proposal_str = f" | Proposals: {', '.join(parts)}"
            except Exception:
                pass

            # Render
            header = (
                f"- [{priority.upper()}] **{title}** "
                f"(id={goal_id}, {category})"
            )
            detail_parts = [p for p in [stale_str, progress_str, proposal_str] if p]
            if detail_parts:
                detail = "".join(detail_parts).lstrip(" |")
                lines.append(f"{header}\n  {detail}")
            else:
                lines.append(header)

        lines.append("")
        return "\n".join(lines)

    async def _user_directives_section(self) -> str:
        """User directives — explicit user instructions for the ego.

        Only rendered if there are active directives. Returns empty string
        otherwise to avoid polluting context with empty sections.
        """
        try:
            from genesis.db.crud import ego as ego_crud

            directives = await ego_crud.list_active_directives(
                self._db, ego_target="user_ego", limit=5,
            )
        except Exception:
            logger.debug("Failed to query ego directives", exc_info=True)
            return ""

        if not directives:
            return ""

        lines = ["## User Directives\n"]
        lines.append(
            "*The user flagged these as important. Factor them into your "
            "thinking — but you decide what to propose.*\n"
        )

        from datetime import UTC, datetime

        now = datetime.now(UTC)
        for d in directives:
            priority = d.get("priority", "normal").upper()
            content = d.get("content", "?")[:200]
            directive_id = d.get("id", "?")
            created_at = d.get("created_at", "")
            # Compute age
            age_str = ""
            if created_at:
                try:
                    created = datetime.fromisoformat(created_at)
                    delta = now - created
                    if delta.days > 0:
                        age_str = f"{delta.days}d ago"
                    else:
                        hours = int(delta.total_seconds() / 3600)
                        age_str = f"{hours}h ago" if hours > 0 else "just now"
                except (ValueError, TypeError):
                    pass
            age_part = f", {age_str}" if age_str else ""
            lines.append(
                f"- [{priority}] {content}\n"
                f"  (id={directive_id}{age_part})"
            )

        lines.append("")
        return "\n".join(lines)

    async def _world_snapshot_section(self) -> str:
        """Synthesized view of the user's world — events, contacts, signals.

        Replaces the raw user-world observations section with a structured
        world snapshot that connects goals to events, contacts, and signals.
        """
        lines = ["## User's World\n"]

        try:
            from genesis.ego.world_snapshot import build as build_snapshot
            snapshot = await build_snapshot(self._db)
            rendered = snapshot.render()
            lines.append(rendered)
        except Exception:
            logger.debug("Failed to build world snapshot", exc_info=True)
            lines.append("*World snapshot not available.*\n")

        return "\n".join(lines)

    # Signals that track user activity — used to filter awareness tick
    # signals for the user ego's activity pulse section.
    _USER_FACING_SIGNALS = frozenset({
        "conversations_since_reflection",
        "task_completion_quality",
        "recon_findings_pending",
        "stale_pending_items",
        "user_goal_staleness",
        "user_session_pattern",
    })

    # Human-readable interpretations for user-facing signal ranges.
    _SIGNAL_INTERPRETATIONS: dict[str, list[tuple[float, str]]] = {
        "user_goal_staleness": [
            (0.7, "oldest pending goal is significantly stale (14+ days)"),
            (0.3, "oldest pending goal is moderately stale (7-14 days)"),
            (0.0, "pending goals are relatively fresh"),
        ],
        "user_session_pattern": [
            (0.7, "user activity is significantly below their baseline"),
            (0.3, "user activity is somewhat below their baseline"),
            (0.0, "user activity is near their normal pattern"),
        ],
        "stale_pending_items": [
            (0.7, "several pending items are aging (3+ days)"),
            (0.3, "some pending items are aging"),
            (0.0, "pending items are fresh"),
        ],
    }

    @classmethod
    def _interpret_signal(cls, name: str, value: float) -> str | None:
        """Return a human-readable interpretation, or None to skip."""
        thresholds = cls._SIGNAL_INTERPRETATIONS.get(name)
        if thresholds:
            for threshold, text in thresholds:
                if value >= threshold:
                    return text
            return None
        # Default: just show the raw value for known user-facing signals
        if isinstance(value, float):
            return f"{value:.2f}"
        return str(value)

    async def _user_activity_pulse_section(self) -> str:
        """User activity signals — interpreted for ego decision-making.

        Queries the latest awareness tick and surfaces user-facing signal
        values as prose. Signals at 0.0 are skipped (nothing noteworthy).
        """
        lines = ["## User Activity Pulse\n"]

        try:
            cursor = await self._db.execute(
                "SELECT signals_json, created_at "
                "FROM awareness_ticks "
                "ORDER BY created_at DESC LIMIT 1"
            )
            row = await cursor.fetchone()
        except Exception:
            logger.error("Failed to query awareness_ticks", exc_info=True)
            lines.append("*No awareness data available.*\n")
            return "\n".join(lines)

        if not row:
            lines.append("*No awareness ticks recorded.*\n")
            return "\n".join(lines)

        signals_json, created_at = row

        try:
            signals = json.loads(signals_json) if signals_json else {}
        except (json.JSONDecodeError, TypeError):
            signals = {}

        # Normalize list format to dict (same as genesis_context.py)
        if isinstance(signals, list):
            signals = {
                s["name"]: s
                for s in signals
                if isinstance(s, dict) and "name" in s
            }

        # Filter to user-facing signals with non-zero values
        pulse_items: list[str] = []
        for sig_name in sorted(self._USER_FACING_SIGNALS):
            sig_info = signals.get(sig_name)
            if not sig_info:
                continue
            value = sig_info.get("value", 0.0) if isinstance(sig_info, dict) else sig_info
            if not isinstance(value, (int, float)) or value == 0.0:
                continue
            interpretation = self._interpret_signal(sig_name, float(value))
            if interpretation:
                label = sig_name.replace("_", " ").title()
                pulse_items.append(
                    f"- **{label}**: {interpretation} (signal: {value:.2f})"
                )

        if not pulse_items:
            lines.append("*All user activity signals nominal.*\n")
        else:
            ts = created_at[:16] if created_at else "?"
            lines.append(f"*From latest awareness tick ({ts}):*\n")
            lines.extend(pulse_items)

        lines.append("")
        return "\n".join(lines)

    async def _recent_conversations_section(self) -> str:
        """What the user has been working on — from cc_sessions."""
        lines = ["## Recent Conversations (48h)\n"]

        try:
            cursor = await self._db.execute(
                "SELECT topic, source_tag, model, started_at "
                "FROM cc_sessions "
                "WHERE source_tag = 'foreground' "
                "AND started_at >= datetime('now', '-48 hours') "
                "AND topic != '' "
                "ORDER BY started_at DESC "
                "LIMIT 10"
            )
            rows = await cursor.fetchall()
        except Exception:
            logger.error("Failed to query cc_sessions", exc_info=True)
            lines.append("*Could not query recent sessions.*\n")
            return "\n".join(lines)

        if not rows:
            lines.append("*No foreground sessions in last 48h.*\n")
            return "\n".join(lines)

        lines.append(f"**{len(rows)} sessions** (most recent first):\n")
        for topic, _source_tag, model, started_at in rows:
            short_topic = topic[:200] + "..." if len(topic) > 200 else topic
            short_topic = short_topic.replace("\n", " ")
            ts = started_at[:16] if started_at else "?"
            domain = " [Genesis internal]" if _is_genesis_internal(short_topic) else ""
            lines.append(f"- [{ts}] ({model}) {short_topic}{domain}")

        lines.append(
            "\nSessions marked [Genesis internal] are system development work — "
            "the Genesis ego's domain. Focus on user-facing threads only.\n"
        )
        return "\n".join(lines)

    async def _backlog_summary_section(self) -> str:
        """Inbox, recon, and pending item backlogs."""
        lines = ["## Backlogs\n"]

        counts: list[tuple[str, int, str | None]] = []  # (label, count, oldest)

        # Inbox: pending/processing items
        try:
            cursor = await self._db.execute(
                "SELECT COUNT(*), MIN(created_at) FROM inbox_items "
                "WHERE status NOT IN ('completed', 'failed')"
            )
            row = await cursor.fetchone()
            if row and row[0] > 0:
                age = self._days_ago(row[1])
                counts.append(("Inbox", row[0], age))
        except Exception:
            pass  # Table may not exist

        # Recon findings: unresolved
        try:
            cursor = await self._db.execute(
                "SELECT COUNT(*), MIN(created_at) FROM observations "
                "WHERE type = 'finding' AND resolved = 0"
            )
            row = await cursor.fetchone()
            if row and row[0] > 0:
                age = self._days_ago(row[1])
                counts.append(("Recon findings", row[0], age))
        except Exception:
            pass

        # Follow-ups awaiting user input
        try:
            cursor = await self._db.execute(
                "SELECT COUNT(*), MIN(created_at) FROM follow_ups "
                "WHERE status = 'pending' "
                "AND strategy = 'user_input_needed'"
            )
            row = await cursor.fetchone()
            if row and row[0] > 0:
                age = self._days_ago(row[1])
                counts.append(("Awaiting user input", row[0], age))
        except Exception:
            pass

        if not counts:
            lines.append("*All backlogs clear.*\n")
        else:
            for label, count, oldest in counts:
                age_str = f" (oldest: {oldest})" if oldest else ""
                lines.append(f"- **{label}**: {count} pending{age_str}")

        lines.append("")
        return "\n".join(lines)

    @staticmethod
    def _days_ago(iso_timestamp: str | None) -> str | None:
        """Convert ISO timestamp to 'Xd ago' string."""
        if not iso_timestamp:
            return None
        try:
            dt = datetime.fromisoformat(iso_timestamp)
            if dt.tzinfo is None:
                dt = dt.replace(tzinfo=UTC)
            days = (datetime.now(UTC) - dt).days
            if days == 0:
                return "today"
            if days == 1:
                return "1d ago"
            return f"{days}d ago"
        except (ValueError, TypeError):
            return None

    async def _genesis_escalations_section(self) -> str:
        """Escalations from the Genesis ego that need user ego attention."""
        lines = ["## Genesis Ego Escalations\n"]

        try:
            cursor = await self._db.execute(
                "SELECT source, content, priority, created_at "
                "FROM observations "
                "WHERE resolved = 0 "
                "AND type = 'escalation_to_user_ego' "
                "AND created_at >= datetime('now', '-48 hours') "
                "ORDER BY "
                "  CASE priority "
                "    WHEN 'critical' THEN 0 "
                "    WHEN 'high' THEN 1 "
                "    WHEN 'medium' THEN 2 "
                "    ELSE 3 "
                "  END, "
                "  created_at DESC "
                "LIMIT 10"
            )
            rows = await cursor.fetchall()
        except Exception:
            logger.error("Failed to query escalations", exc_info=True)
            lines.append("*Could not query escalations.*\n")
            return "\n".join(lines)

        if not rows:
            lines.append("*No escalations from Genesis ego.*\n")
            return "\n".join(lines)

        lines.append(f"**{len(rows)} escalations** needing your attention:\n")
        for _source, content, priority, _created_at in rows:
            short = content[:300] + "..." if len(content) > 300 else content
            short = short.replace("\n", " ")
            lines.append(f"- [{priority}] {short}")

        lines.append(
            "\nThese are issues the Genesis ego couldn't resolve alone. "
            "Decide: fix it, escalate to user, or defer.\n"
        )
        return "\n".join(lines)

    async def _capabilities_section(self) -> str:
        """What Genesis CAN do — prompts stretch thinking."""
        lines = ["## Genesis Capabilities\n"]

        if not self._capabilities:
            lines.append("*No capabilities registered.*\n")
            return "\n".join(lines)

        # Just list capabilities briefly — the ego can explore via MCP
        for name, description in sorted(self._capabilities.items()):
            lines.append(f"- **{name}**: {description}")

        lines.append(
            "\nThink about which capabilities could serve the user "
            "that aren't being used enough.\n"
        )
        return "\n".join(lines)

    async def _follow_ups_section(self) -> str:
        """Pending follow-ups the user ego should track."""
        lines = ["## Open Threads\n"]

        try:
            from genesis.db.crud import follow_ups as follow_up_crud

            actionable = await follow_up_crud.get_actionable(self._db)
        except Exception:
            logger.error("Failed to query follow-ups", exc_info=True)
            lines.append("*Could not query follow-ups.*\n")
            return "\n".join(lines)

        if not actionable:
            lines.append("*No follow-ups requiring attention.*\n")
            return "\n".join(lines)

        # Filter to user-relevant follow-ups (pinned items always shown)
        user_relevant = [
            fu for fu in actionable
            if fu.get("strategy") in ("ego_judgment", "user_input_needed")
            or fu.get("priority") in ("high", "critical")
            or fu.get("pinned")
        ]

        if not user_relevant:
            lines.append(
                f"*{len(actionable)} follow-ups exist but none require "
                f"user ego attention.*\n"
            )
            return "\n".join(lines)

        lines.append(f"**{len(user_relevant)} items** needing attention:\n")
        for fu in user_relevant[:10]:
            fid = fu.get("id", "?")
            status = fu.get("status", "?")
            priority = fu.get("priority", "medium")
            content = fu.get("content", "")[:200]
            content = content.replace("\n", " ")
            strategy = fu.get("strategy", "?")
            blocked = fu.get("blocked_reason", "")

            pin_tag = " [pinned]" if fu.get("pinned") else ""
            line = f"- [id:{fid}] [{priority}/{status}]{pin_tag} **{strategy}**: {content}"
            if blocked:
                line += f" — BLOCKED: {blocked[:100]}"
            lines.append(line)

        lines.append("")
        return "\n".join(lines)

    _NEUTRAL_STATUS = NEUTRAL_STATUS  # single source of truth in ego.types

    async def _proposal_history_section(self) -> str:
        """Recent proposal topics with neutral status for context.

        Split into Active and Recently Tried so active proposals are always
        visible. Shows WHAT was proposed and its lifecycle outcome using
        neutral labels (no judgment language). Also shows realist annotations
        when available. No aggregate scores, no user_response text —
        those trigger deference bias.
        """
        lines = ["## Active Proposals\n"]
        table_header = (
            "| Action | Topic | Outcome | Realist |\n"
            "|--------|-------|---------|---------|"
        )

        def _format_row(row) -> str:
            action_type = row["action_type"]
            content = row["content"]
            short = content[:100] + "..." if len(content) > 100 else content
            short = short.replace("\n", " ").replace("|", "/")
            status = self._NEUTRAL_STATUS.get(row["status"], row["status"])
            realist = ""
            if row["realist_verdict"]:
                realist = row["realist_verdict"]
                if row["realist_reasoning"]:
                    reason_short = row["realist_reasoning"][:60]
                    reason_short = reason_short.replace("|", "/")
                    realist = f"{realist}: {reason_short}"
            return f"| {action_type} | {short} | {status} | {realist} |"

        try:
            # Section 1: Active proposals
            cursor = await self._db.execute(
                "SELECT action_type, content, status, realist_verdict, "
                "realist_reasoning, created_at "
                "FROM ego_proposals "
                "WHERE created_at >= datetime('now', '-7 days') "
                "AND status IN ('pending', 'approved', 'executed') "
                "ORDER BY created_at DESC "
                "LIMIT 15",
            )
            active_rows = await cursor.fetchall()

            if not active_rows:
                lines.append("*No active proposals.*\n")
            else:
                lines.append(table_header)
                for row in active_rows:
                    lines.append(_format_row(row))
                lines.append("")

            # Section 2: Recently tried
            lines.append("## Recently Tried (do not re-propose)\n")
            cursor2 = await self._db.execute(
                "SELECT action_type, content, status, realist_verdict, "
                "realist_reasoning, created_at "
                "FROM ego_proposals "
                "WHERE created_at >= datetime('now', '-7 days') "
                "AND status IN ('withdrawn', 'tabled', 'rejected', 'failed', 'expired') "
                "ORDER BY created_at DESC "
                "LIMIT 10",
            )
            tried_rows = await cursor2.fetchall()

            if not tried_rows:
                lines.append("*No recently tried proposals.*\n")
            else:
                lines.append(table_header)
                for row in tried_rows:
                    lines.append(_format_row(row))
                lines.append("")

        except Exception:
            lines.append("*No proposal history available.*\n")

        return "\n".join(lines)

    async def _proposal_board_section(self) -> str:
        """Current board state + approved proposals ready for execution."""
        from genesis.db.crud import ego as ego_crud

        lines = ["## Proposal Board\n"]

        try:
            board = await ego_crud.get_board(self._db, board_size=10)
        except Exception:
            logger.error("Failed to query proposal board", exc_info=True)
            lines.append("*Could not query proposal board.*\n")
            return "\n".join(lines)

        if not board:
            lines.append("*Board is empty — no pending proposals.*\n")
        else:
            lines.append(f"**{len(board)} pending proposals** (your active board):\n")
            for p in board:
                rank = p.get("rank")
                rank_str = f"#{rank}" if rank else "unranked"
                content = (p.get("content") or "")[:150]
                content = content.replace("\n", " ")
                lines.append(
                    f"- [{rank_str}] **{p.get('action_type', '?')}** "
                    f"(id:{p['id']}): {content}"
                )

        # Approved proposals ready for execution
        try:
            approved = await ego_crud.list_proposals(self._db, status="approved", limit=5)
        except Exception:
            approved = []

        if approved:
            lines.append(f"\n**{len(approved)} approved proposals** (ready for execution):\n")
            for p in approved:
                content = (p.get("content") or "")[:150]
                content = content.replace("\n", " ")
                lines.append(
                    f"- **{p.get('action_type', '?')}** (id:{p['id']}): {content}"
                )
            lines.append(
                "\nYou can execute these by outputting execution_briefs "
                "with the proposal_id and dispatch instructions.\n"
            )
        else:
            lines.append("\n*No approved proposals awaiting execution.*\n")

        # Deferred (tabled) proposals — ego can resurface or withdraw
        try:
            tabled = await ego_crud.get_tabled(self._db)
        except Exception:
            tabled = []

        if tabled:
            shown = tabled[:8]
            lines.append(f"\n**Deferred ({len(tabled)} tabled)**:\n")
            for p in shown:
                content = (p.get("content") or "")[:120]
                content = content.replace("\n", " ")
                lines.append(
                    f"- (id:{p['id']}) **{p.get('action_type', '?')}**: {content}"
                )
            if len(tabled) > 8:
                lines.append(f"- ... and {len(tabled) - 8} more")
            lines.append(
                "\nReview deferred items each cycle. Withdraw if stale, "
                "resurface if conditions changed.\n"
            )

        lines.append("")
        return "\n".join(lines)

    async def _execution_outcomes_section(self) -> str:
        """Recent outcomes from ego-dispatched background sessions."""
        lines = ["## Recent Execution Outcomes (48h)\n"]

        try:
            cursor = await self._db.execute(
                "SELECT content, priority, created_at FROM observations "
                "WHERE type = 'execution_outcome' AND source = 'ego_dispatch' "
                "AND created_at >= datetime('now', '-48 hours') "
                "ORDER BY created_at DESC LIMIT 10"
            )
            rows = await cursor.fetchall()
        except Exception:
            logger.error("Failed to query execution outcomes", exc_info=True)
            lines.append("*Could not query execution outcomes.*\n")
            return "\n".join(lines)

        if not rows:
            lines.append("*No ego dispatch outcomes in last 48h.*\n")
            return "\n".join(lines)

        lines.append(f"**{len(rows)} outcomes**:\n")
        for content, priority, created_at in rows:
            short = (content or "")[:200].replace("\n", " ")
            ts = (created_at or "?")[:16]
            lines.append(f"- [{ts}] [{priority}] {short}")

        lines.append("")
        return "\n".join(lines)

    async def _goal_progress_section(self) -> str:
        """Goal pursuit progress — executed proposals grouped by goal."""
        lines = ["## Goal Progress (7d)\n"]

        try:
            cursor = await self._db.execute(
                "SELECT p.goal_id, g.title, p.content, p.status, "
                "  p.user_response, p.created_at "
                "FROM ego_proposals p "
                "JOIN user_goals g ON p.goal_id = g.id "
                "WHERE p.goal_id IS NOT NULL "
                "AND p.status = 'executed' "
                "AND p.created_at >= datetime('now', '-7 days') "
                "ORDER BY p.created_at DESC "
                "LIMIT 15"
            )
            rows = await cursor.fetchall()
        except Exception:
            logger.error("Failed to query goal progress", exc_info=True)
            lines.append("*Could not query goal progress.*\n")
            return "\n".join(lines)

        if not rows:
            lines.append("*No goal-linked proposals executed recently.*\n")
            return "\n".join(lines)

        # Group by goal
        by_goal: dict[str, dict] = {}
        for goal_id, title, content, status, response, created in rows:
            by_goal.setdefault(goal_id, {"title": title, "items": []})
            # user_response format: "session:{id}|completed:{summary}" or "|failed:{summary}"
            resp = response or ""
            outcome = "done" if "|completed:" in resp else "failed" if "|failed:" in resp else status
            by_goal[goal_id]["items"].append(
                f"  - [{outcome}] {(content or '')[:100]} ({(created or '')[:10]})"
            )

        for gid, info in by_goal.items():
            lines.append(f"**{info['title']}** (id={gid})")
            lines.extend(info["items"])

        lines.append("")
        return "\n".join(lines)

    async def _capability_performance_section(self) -> str:
        """Your track record — domain confidence from multiple data sources.

        Framed as context for better proposals, NOT as a limiter.
        The realist gate handles feasibility; this informs confidence calibration.
        """
        lines = ["## Your Track Record\n"]

        try:
            from genesis.db.crud import capability_map as cap_crud

            entries = await cap_crud.get_all(self._db)
        except Exception:
            return ""

        if not entries:
            lines.append("*No performance data yet.*\n")
            return "\n".join(lines)

        _TREND_ICONS = {"improving": "+", "declining": "-", "stable": "="}

        lines.append("| Domain | Confidence | Trend | Evidence |")
        lines.append("|--------|-----------|-------|----------|")
        for e in entries[:12]:
            domain = e.get("domain", "?")
            conf = e.get("confidence", 0.0)
            trend = e.get("trend", "stable")
            evidence = (e.get("evidence_summary") or "")[:80].replace("|", "/")
            icon = _TREND_ICONS.get(trend, "=")
            lines.append(f"| {domain} | {conf:.0%} | {icon} | {evidence} |")

        lines.append(
            "\nUse this to calibrate confidence on proposals. High-confidence "
            "domains deserve ambitious proposals. Low-confidence domains may "
            "benefit from smaller, incremental actions.\n"
        )
        return "\n".join(lines)

    async def _autonomy_readiness_section(self) -> str:
        """Show autonomy posteriors — informs promotion proposals."""
        from genesis.db.crud import autonomy as autonomy_crud
        from genesis.db.crud.autonomy import bayesian_level, bayesian_posterior

        try:
            states = await autonomy_crud.list_all(self._db)
        except Exception:
            return ""
        if not states:
            return ""

        lines = ["## Autonomy Readiness\n"]
        for row in states:
            cat = row["category"]
            level = row["current_level"]
            earned = row["earned_level"]
            successes = row["total_successes"]
            corrections = row["total_corrections"]
            posterior = bayesian_posterior(successes, corrections)
            target = bayesian_level(successes, corrections)
            status = f"L{level}"
            if earned > level:
                status += f" (earned L{earned})"
            readiness = ""
            if target > level:
                readiness = f" — **ready for L{min(level + 1, target)}** (posterior={posterior:.3f})"
            else:
                readiness = f" (posterior={posterior:.3f})"
            lines.append(f"- {cat}: {status}{readiness} [{successes}S/{corrections}C]")

        lines.append("")
        return "\n".join(lines)

    async def _recurring_patterns_section(self) -> str:
        """Detect recurring observation patterns (3+ occurrences in 72h).

        Groups unresolved observations by (type, category) and surfaces
        clusters that may warrant automation or systematic response.
        """
        lines = ["## Recurring Patterns (72h)\n"]

        try:
            cursor = await self._db.execute(
                "SELECT type, category, COUNT(*) AS cnt, "
                "  MAX(content) AS sample, MAX(created_at) AS latest "
                "FROM observations "
                "WHERE created_at >= datetime('now', '-3 days') "
                "  AND resolved = 0 "
                "GROUP BY type, category "
                "HAVING cnt >= 3 "
                "ORDER BY cnt DESC "
                "LIMIT 5"
            )
            rows = await cursor.fetchall()
        except Exception:
            logger.error("Failed to query recurring patterns", exc_info=True)
            lines.append("*Could not query patterns.*\n")
            return "\n".join(lines)

        if not rows:
            lines.append("*No recurring patterns detected.*\n")
            return "\n".join(lines)

        lines.append(
            f"**{len(rows)} pattern(s)** appearing 3+ times "
            f"(may warrant automation):\n"
        )
        for obs_type, category, cnt, sample, _latest in rows:
            cat_str = f"/{category}" if category else ""
            short = (sample or "")[:100].replace("\n", " ")
            if len(sample or "") > 100:
                short += "\u2026"
            lines.append(f"- **[{obs_type}{cat_str}]** \u00d7{cnt} \u2014 {short}")

        lines.append("")
        return "\n".join(lines)

    @staticmethod
    def _output_contract_section() -> str:
        """Remind the ego of its required output format."""
        return (
            "## Output Contract\n\n"
            "You MUST respond with valid JSON matching this structure:\n"
            "```json\n"
            "{\n"
            '  "proposals": [\n'
            "    {\n"
            '      "action_type": "investigate|outreach|maintenance|dispatch|config|recurring_pattern",\n'
            '      "action_category": "category for tracking",\n'
            '      "content": "what you want to do (specific and actionable)",\n'
            '      "rationale": "why this helps the user",\n'
            '      "confidence": 0.85,\n'
            '      "urgency": "low|normal|high|critical",\n'
            '      "alternatives": "what else you considered",\n'
            '      "memory_basis": "non-obvious memory that informed this (optional)"\n'
            "    }\n"
            "  ],\n"
            '  "focus_summary": "one-line: what you are focused on for the user",\n'
            '  "follow_ups": [],\n'
            '  "resolved_follow_ups": [{"id": "follow_up_id", "resolution": "why resolved"}],\n'
            '  "resolved_directives": [{"id": "directive_id", "resolution": "what you decided"}],\n'
            '  "morning_report": "only if this is a morning trigger"\n'
            "}\n"
            "```\n\n"
            "If you have nothing to propose, return an empty proposals "
            "array. The focus_summary is always required.\n"
        )
