"""Canonical connections index per corpus (`connections.json` SSOT)."""

from __future__ import annotations

import hashlib
import json
import os
import re
from typing import Any

from persistence import secure_atomic_write
from utils.knowledge_topic_labels import build_theme_index, normalize_theme_label

CONNECTIONS_SCHEMA = "mugi.knowledge.connections.v1.4"
CONNECTIONS_SCHEMA_V13 = "mugi.knowledge.connections.v1.3"
CONNECTIONS_SCHEMA_V12 = "mugi.knowledge.connections.v1.2"
CONNECTIONS_SCHEMA_LEGACY = "mugi.knowledge.connections.v1"
_ACCEPTED_SCHEMAS = frozenset(
    {
        CONNECTIONS_SCHEMA,
        CONNECTIONS_SCHEMA_V13,
        CONNECTIONS_SCHEMA_V12,
        "mugi.knowledge.connections.v1.1",
        CONNECTIONS_SCHEMA_LEGACY,
    }
)
_MAX_SEMANTIC_THEMES = 12
_MAX_ENTITIES = 40
_MAX_CLAIMS = 30
_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{0,63}$")


def _semantic_themes_row_cap() -> int:
    try:
        from utils.knowledge_themes import get_themes_config

        raw = get_themes_config().get("max_clusters", _MAX_SEMANTIC_THEMES)
        cap = int(raw)
        return max(1, min(24, cap))
    except (TypeError, ValueError):
        return _MAX_SEMANTIC_THEMES


def _entities_row_cap() -> int:
    try:
        from utils.knowledge_entities import get_entities_config

        raw = get_entities_config().get("max_entities", _MAX_ENTITIES)
        cap = int(raw)
        return max(1, min(80, cap))
    except (TypeError, ValueError):
        return _MAX_ENTITIES


def _claims_row_cap() -> int:
    try:
        from utils.knowledge_entities import get_entities_config

        raw = get_entities_config().get("max_claims", _MAX_CLAIMS)
        cap = int(raw)
        return max(0, min(60, cap))
    except (TypeError, ValueError):
        return _MAX_CLAIMS


def _claim_max_chars() -> int:
    try:
        from utils.knowledge_entities import get_entities_config

        raw = get_entities_config().get("claim_max_chars", 200)
        return max(40, min(500, int(raw)))
    except (TypeError, ValueError):
        return 200
_MAX_SYNTHESIS_EDGES = 40
_MAX_BRIDGE_EDGES = 80


def _validate_slug(slug: str) -> str:
    raw = (slug or "").strip().lower()
    if not raw or not _SLUG_RE.match(raw):
        raise ValueError("invalid corpus slug")
    return raw


def connections_path(slug: str) -> str:
    from tools.knowledge_tools import _knowledge_root

    safe = _validate_slug(slug)
    return os.path.join(_knowledge_root(), safe, "connections.json")


def load_connections_index(slug: str) -> dict[str, Any] | None:
    path = connections_path(slug)
    if not os.path.isfile(path):
        return None
    try:
        with open(path, encoding="utf-8") as f:
            data = json.load(f)
        if not isinstance(data, dict):
            return None
        for field in ("entities", "claims", "semantic_themes", "contradictions"):
            if not isinstance(data.get(field), list):
                data[field] = []
        return data
    except (OSError, json.JSONDecodeError):
        return None


def preserved_semantic_themes(slug: str) -> list[dict[str, Any]]:
    """Return on-disk semantic_themes (used when discovery is disabled or failed)."""
    data = load_connections_index(slug)
    if not isinstance(data, dict):
        return []
    rows = data.get("semantic_themes")
    if not isinstance(rows, list):
        return []
    return [dict(row) for row in rows if isinstance(row, dict)]


def preserved_entities(slug: str) -> list[dict[str, Any]]:
    """Return on-disk entities (used when discovery is disabled or failed)."""
    data = load_connections_index(slug)
    if not isinstance(data, dict):
        return []
    rows = data.get("entities")
    if not isinstance(rows, list):
        return []
    return [dict(row) for row in rows if isinstance(row, dict)]


def preserved_claims(slug: str) -> list[dict[str, Any]]:
    """Return on-disk claims (used when discovery is disabled or failed)."""
    data = load_connections_index(slug)
    if not isinstance(data, dict):
        return []
    rows = data.get("claims")
    if not isinstance(rows, list):
        return []
    return [dict(row) for row in rows if isinstance(row, dict)]


def preserved_contradictions(slug: str) -> list[dict[str, Any]]:
    """Return on-disk contradictions (used when detection is disabled or failed)."""
    data = load_connections_index(slug)
    if not isinstance(data, dict):
        return []
    rows = data.get("contradictions")
    if not isinstance(rows, list):
        return []
    return [dict(row) for row in rows if isinstance(row, dict)]


def _contradictions_row_cap() -> int:
    try:
        from utils.knowledge_contradictions import get_contradictions_config

        raw = get_contradictions_config().get("max_contradictions", 20)
        cap = int(raw)
        return max(0, min(40, cap))
    except (TypeError, ValueError):
        return 20


def save_connections_index(slug: str, data: dict[str, Any]) -> None:
    path = connections_path(slug)
    os.makedirs(os.path.dirname(path), exist_ok=True)
    secure_atomic_write(path, json.dumps(data, indent=2, ensure_ascii=False) + "\n")


def _slug_to_source_map(sources: list[dict[str, Any]]) -> dict[str, dict[str, Any]]:
    out: dict[str, dict[str, Any]] = {}
    for src in sources:
        if not isinstance(src, dict):
            continue
        sid = str(src.get("slug") or "").strip()
        if not sid or ".." in sid or "/" in sid:
            continue
        out[sid] = src
    return out


def _sources_for_slugs(
    slug_map: dict[str, dict[str, Any]],
    slugs: list[str],
    corpus_slug: str,
) -> list[dict[str, str]]:
    from utils.knowledge_topic_labels import build_source_entry

    rows: list[dict[str, str]] = []
    for sid in slugs:
        src = slug_map.get(sid)
        if not src:
            continue
        rows.append(build_source_entry(src, corpus_slug))
    return rows


def _cap_semantic_theme_rows(rows: list[dict[str, Any]], *, max_rows: int) -> list[dict[str, Any]]:
    capped: list[dict[str, Any]] = []
    for row in rows[:max_rows]:
        if not isinstance(row, dict):
            continue
        item = dict(row)
        item["source_slugs"] = list(item.get("source_slugs") or [])[:40]
        item["representative_row_ids"] = list(item.get("representative_row_ids") or [])[:40]
        item["keywords"] = list(item.get("keywords") or [])[:12]
        capped.append(item)
    return capped


def _cap_entity_rows(rows: list[dict[str, Any]], *, max_rows: int) -> list[dict[str, Any]]:
    capped: list[dict[str, Any]] = []
    for row in rows[:max_rows]:
        if not isinstance(row, dict):
            continue
        item = dict(row)
        item["source_slugs"] = list(item.get("source_slugs") or [])[:40]
        item["chunk_node_ids"] = list(item.get("chunk_node_ids") or [])[:40]
        item["keywords"] = list(item.get("keywords") or [])[:12]
        capped.append(item)
    return capped


def _cap_claim_rows(rows: list[dict[str, Any]], *, max_rows: int, max_chars: int) -> list[dict[str, Any]]:
    capped: list[dict[str, Any]] = []
    for row in rows[:max_rows]:
        if not isinstance(row, dict):
            continue
        item = dict(row)
        text = str(item.get("claim_text") or "")
        if len(text) > max_chars:
            item["claim_text"] = text[: max_chars - 3].rstrip() + "..."
        item["entity_keys"] = list(item.get("entity_keys") or [])[:40]
        item["entity_labels"] = list(item.get("entity_labels") or [])[:40]
        item["chunk_node_ids"] = list(item.get("chunk_node_ids") or [])[:40]
        capped.append(item)
    return capped


def _cap_contradiction_rows(rows: list[dict[str, Any]], *, max_rows: int, max_chars: int) -> list[dict[str, Any]]:
    from utils.knowledge_themes import _looks_secret

    capped: list[dict[str, Any]] = []
    for row in rows[:max_rows]:
        if not isinstance(row, dict):
            continue
        item = dict(row)
        block_a = item.get("claim_a")
        block_b = item.get("claim_b")
        if not isinstance(block_a, dict) or not isinstance(block_b, dict):
            continue
        sub_a = dict(block_a)
        sub_b = dict(block_b)
        text_a = str(sub_a.get("text") or "").strip()
        text_b = str(sub_b.get("text") or "").strip()
        if not text_a or not text_b or _looks_secret(text_a) or _looks_secret(text_b):
            continue
        if len(text_a) > max_chars:
            sub_a["text"] = text_a[: max_chars - 3].rstrip() + "..."
        if len(text_b) > max_chars:
            sub_b["text"] = text_b[: max_chars - 3].rstrip() + "..."
        item["claim_a"] = sub_a
        item["claim_b"] = sub_b
        capped.append(item)
    return capped


def _contradiction_matches_filter(
    row: dict[str, Any],
    *,
    topic_low: str,
    query_low: str,
) -> bool:
    from utils.knowledge_themes import _tokenize_for_themes

    entity_key = str(row.get("entity_key") or "").lower()
    entity_label = str(row.get("entity_label") or "").lower()
    entity_keys_raw = row.get("entity_keys") or []
    entity_keys_low: list[str] = []
    if isinstance(entity_keys_raw, list):
        entity_keys_low = [str(k).lower() for k in entity_keys_raw if str(k).strip()]

    def _entity_term_match(term_low: str) -> bool:
        if _term_matches_field(term_low, entity_key) or _term_matches_field(term_low, entity_label):
            return True
        return any(_term_matches_field(term_low, ek) for ek in entity_keys_low)

    if topic_low:
        if _entity_term_match(topic_low):
            return True
        for side in ("claim_a", "claim_b"):
            block = row.get(side)
            if not isinstance(block, dict):
                continue
            text = str(block.get("text") or "").lower()
            if _single_term_matches_text(text, set(_tokenize_for_themes(text)), topic_low):
                return True
        return False
    if query_low:
        if _entity_term_match(query_low):
            return True
        for side in ("claim_a", "claim_b"):
            block = row.get(side)
            if not isinstance(block, dict):
                continue
            text = str(block.get("text") or "").lower()
            if _single_term_matches_text(text, set(_tokenize_for_themes(text)), query_low):
                return True
        return False
    return True


def _single_term_matches_text(text_low: str, text_tokens: set[str], term_low: str) -> bool:
    """Match a single-token query without substring false positives (e.g. art in start)."""
    if not term_low:
        return False
    if " " in term_low:
        return term_low in text_low
    return term_low in text_tokens


def _term_matches_field(term_low: str, field_low: str) -> bool:
    """Match a filter term against a canonical key, label, or entity_key."""
    if not term_low or not field_low:
        return False
    if term_low == field_low:
        return True
    if " " in term_low:
        return term_low in field_low
    from utils.knowledge_themes import _tokenize_for_themes

    return _single_term_matches_text(
        field_low,
        set(_tokenize_for_themes(field_low)),
        term_low,
    )


def _claim_matches_filter(
    row: dict[str, Any],
    *,
    topic_low: str,
    query_low: str,
) -> bool:
    from utils.knowledge_themes import _tokenize_for_themes

    claim_text = str(row.get("claim_text") or "").lower()
    claim_tokens = set(_tokenize_for_themes(claim_text))
    labels_raw = row.get("entity_labels") or []
    labels_low: list[str] = []
    if isinstance(labels_raw, list):
        labels_low = [str(l).lower() for l in labels_raw if str(l).strip()]
    keys_raw = row.get("entity_keys") or []
    keys_low: list[str] = []
    if isinstance(keys_raw, list):
        keys_low = [str(k).lower() for k in keys_raw if str(k).strip()]
    if topic_low:
        if _single_term_matches_text(claim_text, claim_tokens, topic_low):
            return True
        if any(_term_matches_field(topic_low, k) for k in keys_low):
            return True
        return any(
            _single_term_matches_text(lbl, set(_tokenize_for_themes(lbl)), topic_low)
            for lbl in labels_low
        )
    if query_low:
        if _single_term_matches_text(claim_text, claim_tokens, query_low):
            return True
        if any(_term_matches_field(query_low, k) for k in keys_low):
            return True
        return any(
            _single_term_matches_text(lbl, set(_tokenize_for_themes(lbl)), query_low)
            for lbl in labels_low
        )
    return True


def materialize_connections_for_corpus(
    slug: str,
    label: str,
    synthesis_text: str,
    sources: list[dict[str, Any]],
    *,
    broad_topics: list[str] | None = None,
    corpus_topics: list[str] | None = None,
    semantic_themes: list[dict[str, Any]] | None = None,
    entities: list[dict[str, Any]] | None = None,
    claims: list[dict[str, Any]] | None = None,
    contradictions: list[dict[str, Any]] | None = None,
    ingest_profile: str = "docs",
) -> dict[str, Any]:
    """Build connections.json from digest outputs."""
    from tools.knowledge_tools import parse_synthesis_connections_section

    safe_slug = _validate_slug(slug)
    slug_map = _slug_to_source_map(sources)

    topics = build_theme_index(
        corpus_slug=safe_slug,
        sources=sources,
        broad_topics=broad_topics,
        corpus_topics=corpus_topics,
        ingest_profile=ingest_profile,
    )

    canonical_to_slugs: dict[str, list[str]] = {
        str(t.get("canonical_key") or ""): list(t.get("source_slugs") or [])
        for t in topics
        if str(t.get("canonical_key") or "")
    }

    synthesis_edges: list[dict[str, Any]] = []
    for bullet in parse_synthesis_connections_section(synthesis_text):
        text = (bullet or "").strip()
        if not text:
            continue
        eid = hashlib.sha256(f"{safe_slug}:syn:{text}".encode()).hexdigest()[:16]
        synthesis_edges.append(
            {
                "id": eid,
                "kind": "synthesis_link",
                "src": "",
                "dst": "",
                "rationale": text[:1200],
                "confidence": 0.85,
                "sources": [],
                "source_count": 0,
            }
        )

    bridge_edges: list[dict[str, Any]] = []
    for row in topics:
        canonical = str(row.get("canonical_key") or "")
        slugs = list(row.get("source_slugs") or [])
        if len(slugs) < 2:
            continue
        display = str(row.get("label") or canonical)
        pair_key = "|".join(sorted(slugs)[:4])
        eid = hashlib.sha256(f"{safe_slug}:topic:{canonical}:{pair_key}".encode()).hexdigest()[:16]
        source_rows = _sources_for_slugs(slug_map, slugs, safe_slug)
        bridge_edges.append(
            {
                "id": eid,
                "kind": "topic_cooccurrence",
                "src": canonical,
                "dst": pair_key,
                "theme_key": canonical,
                "rationale": f"Theme “{display}” appears in {len(slugs)} sources.",
                "confidence": min(0.55 + 0.08 * len(slugs), 0.95),
                "sources": source_rows[:12],
                "source_count": len(slugs),
            }
        )

    bridge_edges.sort(key=lambda e: (-int(e.get("source_count") or 0), str(e.get("src") or "")))
    edges = synthesis_edges[:_MAX_SYNTHESIS_EDGES] + bridge_edges[:_MAX_BRIDGE_EDGES]

    return {
        "schema": CONNECTIONS_SCHEMA,
        "corpus_slug": safe_slug,
        "corpus_label": label,
        "topics": topics,
        "edges": edges,
        "semantic_themes": _cap_semantic_theme_rows(
            list(semantic_themes or []),
            max_rows=_semantic_themes_row_cap(),
        ),
        "entities": _cap_entity_rows(
            list(entities or []),
            max_rows=_entities_row_cap(),
        ),
        "claims": _cap_claim_rows(
            list(claims or []),
            max_rows=_claims_row_cap(),
            max_chars=_claim_max_chars(),
        ),
        "contradictions": _cap_contradiction_rows(
            list(contradictions or []),
            max_rows=_contradictions_row_cap(),
            max_chars=_claim_max_chars(),
        ),
    }


def _topic_matches_filter(
    row: dict[str, Any],
    *,
    topic_low: str,
    query_low: str,
) -> bool:
    from utils.knowledge_themes import _tokenize_for_themes

    label = str(row.get("label") or "")
    label_low = label.lower()
    canonical = str(row.get("canonical_key") or label_low).lower()
    entity_key = str(row.get("entity_key") or canonical).lower()
    label_tokens = set(_tokenize_for_themes(label_low))
    canonical_tokens = set(_tokenize_for_themes(canonical))
    entity_tokens = set(_tokenize_for_themes(entity_key))
    keywords_raw = row.get("keywords") or []
    keywords_low: list[str] = []
    if isinstance(keywords_raw, list):
        keywords_low = [str(k).lower() for k in keywords_raw if str(k).strip()]
    if topic_low:
        if topic_low in (canonical, label_low, entity_key):
            return True
        if (
            _term_matches_field(topic_low, canonical)
            or _term_matches_field(topic_low, label_low)
            or _term_matches_field(topic_low, entity_key)
        ):
            return True
        return any(
            topic_low == kw or _term_matches_field(topic_low, kw) for kw in keywords_low
        )
    if query_low:
        if (
            _single_term_matches_text(label_low, label_tokens, query_low)
            or _single_term_matches_text(canonical, canonical_tokens, query_low)
            or _single_term_matches_text(entity_key, entity_tokens, query_low)
        ):
            return True
        return any(
            _single_term_matches_text(kw, set(_tokenize_for_themes(kw)), query_low) for kw in keywords_low
        )
    return True


def _topic_source_slug_keys(
    intra_themes: list[dict[str, Any]],
    *,
    topic_canonical: str,
    topic_term: str,
) -> set[tuple[str, str]]:
    """``(corpus_slug, source_slug)`` for digest themes matching the active ``topic=`` filter."""
    keys: set[tuple[str, str]] = set()
    term_low = topic_term.strip().lower()
    canonical_low = (topic_canonical or "").strip().lower()
    if not term_low and not canonical_low:
        return keys
    topic_low = canonical_low or term_low
    for row in intra_themes:
        if not isinstance(row, dict):
            continue
        if not _topic_matches_filter(row, topic_low=topic_low, query_low=""):
            continue
        corpus = str(row.get("corpus_slug") or "").strip()
        if not corpus:
            continue
        for sid in row.get("source_slugs") or []:
            s = str(sid).strip()
            if s:
                keys.add((corpus, s))
    return keys


def _row_shares_topic_sources(
    row: dict[str, Any],
    *,
    corpus_slug: str,
    topic_source_keys: set[tuple[str, str]],
) -> bool:
    if not topic_source_keys:
        return False
    for sid in row.get("source_slugs") or []:
        s = str(sid).strip()
        if s and (corpus_slug, s) in topic_source_keys:
            return True
    return False


def _sidebar_row_matches_active_topic(
    row: dict[str, Any],
    *,
    corpus_slug: str,
    topic_source_keys: set[tuple[str, str]],
    topic_low: str,
    query_low: str,
    claim_row: bool = False,
) -> bool:
    """Label/keyword match, or shared sources with the selected digest theme."""
    if not topic_low and not query_low:
        return True
    if claim_row:
        if _claim_matches_filter(row, topic_low=topic_low, query_low=query_low):
            return True
        if topic_source_keys:
            sid = str(row.get("source_slug") or "").strip()
            if sid and (corpus_slug, sid) in topic_source_keys:
                return True
        return False
    if _topic_matches_filter(row, topic_low=topic_low, query_low=query_low):
        return True
    if topic_source_keys and topic_low:
        return _row_shares_topic_sources(
            row, corpus_slug=corpus_slug, topic_source_keys=topic_source_keys
        )
    return False


def _filter_corpora_rows(
    corpora: list[dict[str, Any]],
    corpus_filter: str | None,
) -> list[dict[str, Any]]:
    if not corpus_filter:
        return corpora
    cf = corpus_filter.strip().lower()
    if _SLUG_RE.match(cf):
        return [c for c in corpora if str(c.get("slug") or "").lower() == cf]
    return [c for c in corpora if cf in str(c.get("label") or "").lower()]


def _enrich_connections_index(
    data: dict[str, Any],
    slug: str,
    sources: list[dict[str, Any]],
) -> dict[str, Any]:
    """Upgrade legacy v1 rows in memory (canonical keys, source lists on bridges)."""
    slug_map = _slug_to_source_map(sources)
    topics = data.get("topics") or []
    for row in topics:
        if not isinstance(row, dict):
            continue
        label = str(row.get("label") or "")
        if not row.get("canonical_key"):
            _, canonical = normalize_theme_label(label)
            row["canonical_key"] = canonical or label.lower()
        slugs = list(row.get("source_slugs") or [])
        if "source_count" not in row:
            row["source_count"] = len(slugs)
        if not row.get("sources") and slugs:
            row["sources"] = _sources_for_slugs(slug_map, slugs, slug)

    topic_by_canonical = {
        str(t.get("canonical_key") or "").lower(): t
        for t in topics
        if isinstance(t, dict) and str(t.get("canonical_key") or "")
    }
    for edge in data.get("edges") or []:
        if not isinstance(edge, dict):
            continue
        if str(edge.get("kind") or "") != "topic_cooccurrence":
            continue
        theme_key = str(edge.get("theme_key") or edge.get("src") or "").lower()
        if not edge.get("theme_key") and theme_key:
            edge["theme_key"] = theme_key
        if edge.get("sources"):
            continue
        topic_row = topic_by_canonical.get(theme_key)
        if not topic_row:
            continue
        slugs = list(topic_row.get("source_slugs") or [])
        edge["sources"] = _sources_for_slugs(slug_map, slugs, slug)
        edge["source_count"] = len(slugs)
    return data


def corpora_missing_connections_index(
    *,
    corpus_filter: str | None = None,
) -> list[dict[str, str]]:
    """Corpora with digest output but no connections.json (operator should rebuild)."""
    from tools.knowledge_tools import list_knowledge_corpora

    gaps: list[dict[str, str]] = []
    corpora = _filter_corpora_rows(list_knowledge_corpora(), corpus_filter)
    for c in corpora:
        slug = str(c.get("slug") or "")
        if not slug:
            continue
        status = str(c.get("digest_status") or "").lower()
        if status not in {"complete", "partial"}:
            continue
        if load_connections_index(slug):
            continue
        gaps.append({"slug": slug, "label": str(c.get("label") or slug)})
    return gaps


def build_shared_topics_from_connections(
    *,
    corpus_filter: str | None = None,
    shared_filter: str = "",
    max_corpora: int = 10,
) -> list[dict[str, Any]]:
    """Cross-corpus themes keyed by connections.json canonical_key (not manifest topics)."""
    from tools.knowledge_tools import _load_manifest, list_knowledge_corpora

    index: dict[str, dict[str, Any]] = {}
    corpora = _filter_corpora_rows(list_knowledge_corpora(), corpus_filter)
    filter_low = shared_filter.strip().lower()

    for c in corpora:
        slug = str(c.get("slug") or "")
        label = str(c.get("label") or slug)
        if not slug:
            continue
        data = load_connections_index(slug)
        if not data or str(data.get("schema") or "") not in _ACCEPTED_SCHEMAS:
            continue
        manifest = _load_manifest(slug) or {}
        sources = manifest.get("sources") if isinstance(manifest.get("sources"), list) else []
        data = _enrich_connections_index(data, slug, sources)
        for row in data.get("topics") or []:
            if not isinstance(row, dict):
                continue
            canonical = str(row.get("canonical_key") or "").lower()
            if not canonical:
                continue
            display = str(row.get("label") or canonical)
            bucket = index.setdefault(
                canonical,
                {"topic": display, "canonical_key": canonical, "corpora": []},
            )
            if len(display) > len(str(bucket.get("topic") or "")):
                bucket["topic"] = display
            entries: list[dict[str, str]] = bucket["corpora"]
            if not any(str(e.get("slug") or "") == slug for e in entries):
                entries.append({"slug": slug, "label": label})

    shared: list[dict[str, Any]] = []
    for canonical in sorted(index.keys()):
        bucket = index[canonical]
        entries = bucket.get("corpora") or []
        if len(entries) < 2:
            continue
        topic_label = str(bucket.get("topic") or canonical)
        if filter_low and filter_low not in canonical and filter_low not in topic_label.lower():
            continue
        shared.append(
            {
                "topic": topic_label,
                "canonical_key": canonical,
                "corpora": entries[:max_corpora],
            }
        )
        if len(shared) >= max_corpora:
            break
    return shared


def source_rows_for_theme(
    theme_row: dict[str, Any],
    corpus_slug: str,
    manifest_sources: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Resolve theme source list from v1.1 ``sources`` or legacy ``source_slugs``."""
    from utils.knowledge_topic_labels import build_source_entry

    rows: list[dict[str, Any]] = []
    for src in theme_row.get("sources") or []:
        if not isinstance(src, dict):
            continue
        row = dict(src)
        if not row.get("corpus_slug"):
            row["corpus_slug"] = corpus_slug
        if row.get("slug"):
            rows.append(row)
    if rows:
        return rows
    slug_map = _slug_to_source_map(manifest_sources)
    for sid in theme_row.get("source_slugs") or []:
        sid_str = str(sid or "").strip()
        if not sid_str:
            continue
        src = slug_map.get(sid_str)
        if src:
            rows.append(build_source_entry(src, corpus_slug))
        else:
            rows.append({"slug": sid_str, "filename": sid_str, "corpus_slug": corpus_slug})
    return rows


def _edge_matches_filter(
    edge: dict[str, Any],
    *,
    topic_low: str,
    query_low: str,
) -> bool:
    rationale = str(edge.get("rationale") or "")
    src = str(edge.get("src") or "")
    theme_key = str(edge.get("theme_key") or src or "")
    rationale_low = rationale.lower()
    src_low = src.lower()
    theme_low = theme_key.lower()
    if topic_low:
        kind = str(edge.get("kind") or "")
        if kind == "topic_cooccurrence":
            return topic_low in {src_low, theme_low}
        return False
    if query_low and query_low not in rationale_low and query_low not in src_low and query_low not in theme_low:
        return False
    return True


def list_all_connection_edges(
    *,
    corpus_filter: str | None = None,
    topic_filter: str | None = None,
    query_filter: str | None = None,
) -> tuple[list[dict[str, Any]], list[dict[str, Any]]]:
    """Return (intra_corpus_themes, edges) aggregated from connections.json files."""
    from tools.knowledge_tools import list_knowledge_corpora

    themes: list[dict[str, Any]] = []
    edges: list[dict[str, Any]] = []
    from tools.knowledge_tools import _load_manifest

    corpora = _filter_corpora_rows(list_knowledge_corpora(), corpus_filter)

    topic_raw = (topic_filter or "").strip()
    topic_low = topic_raw.lower()
    if topic_raw and not topic_low:
        topic_low = ""
    elif topic_raw:
        _, canonical = normalize_theme_label(topic_raw)
        if canonical:
            topic_low = canonical

    query_low = (query_filter or "").strip().lower()

    for c in corpora:
        slug = str(c.get("slug") or "")
        label = str(c.get("label") or slug)
        if not slug:
            continue
        data = load_connections_index(slug)
        if not data:
            continue
        if str(data.get("schema") or "") not in _ACCEPTED_SCHEMAS:
            continue
        manifest = _load_manifest(slug) or {}
        sources = manifest.get("sources") if isinstance(manifest.get("sources"), list) else []
        data = _enrich_connections_index(data, slug, sources)
        for row in data.get("topics") or []:
            if not isinstance(row, dict):
                continue
            if not _topic_matches_filter(row, topic_low=topic_low, query_low=query_low):
                continue
            theme_row = dict(row)
            theme_row["corpus_slug"] = slug
            theme_row["corpus_label"] = label
            if "source_count" not in theme_row:
                slugs = theme_row.get("source_slugs") or []
                theme_row["source_count"] = len(slugs)
            themes.append(theme_row)
        for edge in data.get("edges") or []:
            if not isinstance(edge, dict):
                continue
            if topic_low:
                if not _edge_matches_filter(edge, topic_low=topic_low, query_low=""):
                    continue
            row = dict(edge)
            row["corpus_slug"] = slug
            row["corpus_label"] = label
            edges.append(row)

    themes.sort(
        key=lambda r: (
            -int(r.get("source_count") or len(r.get("source_slugs") or [])),
            str(r.get("label") or "").lower(),
        )
    )
    return themes[:80], edges[:80]
