el/elp/projector/plan.py

"""plan.py — PLAN stage: geometry region -> document skeleton (a DAG/outline).

The manifold becomes the skeleton. We extract faithful propositions from the
region's nodes (the proven neuron-talk extractor, SACRED polarity preserved),
apply a quality floor, then GROUP them into sections. Grouping is by source
node — each engram node is one coherent topic, so one salient node becomes one
section. The section ORDER is the node ranking (importance/salience): the
geometry decides the outline, not a template.

Output: a DocumentIR whose sections carry seed node ids and empty blocks. REALIZE
fills the blocks; the plan owns the structure.
"""
from __future__ import annotations

import os
import re
import sys

_NT = os.path.expanduser("~/Desktop/neuron-talk")
_LR = os.path.expanduser("~/Desktop/lang-realizers")
for _p in (_NT, _LR):
    if _p not in sys.path:
        sys.path.insert(0, _p)

import propositions  # noqa: E402  (the proven, faithful extractor)

from document_ir import DocumentIR, Section  # noqa: E402
from geometry import Region  # noqa: E402


# --------------------------------------------------------------------------- #
#  Proposition quality — keep only clean, well-grounded claims.
# --------------------------------------------------------------------------- #
_JUNK_RE = re.compile(r"[.][a-z]{1,3}\b|[^A-Za-z0-9 '\-]")  # ".o", stray symbols


def _has_banner_token(s: str) -> bool:
    """True if any word is an ALLCAPS banner token (DHARMA, ENGRAM, MEASURED)."""
    for w in (s or "").split():
        core = w.strip(".,:;'\"-")
        if len(core) > 2 and core.isupper():
            return True
    return False


def _clean_prop(p, floor: float) -> bool:
    if p.confidence < floor:
        return False
    if not p.subject or not (p.object or (p.obj_np is not None)):
        return False
    subj = (p.subject or "").strip()
    obj = (p.object or "").strip()
    if len(subj) < 2:
        return False
    # banner-derived shouty fragments read as garbage in prose
    if _has_banner_token(subj) or _has_banner_token(obj):
        return False
    if propositions._is_shouty(p.sentence or ""):
        return False
    # junk tokens: file-extension fragments (".o"), stray non-word symbols
    if _JUNK_RE.search(subj) or _JUNK_RE.search(obj):
        return False
    # a proposition whose object repeats the subject is usually a parse artifact
    if obj and subj.lower() == obj.lower():
        return False
    # a bare copula with no real complement ("X is it") reads as noise
    if p.predicate == "be" and obj.lower() in ("it", "no", "nothing", "empty", ""):
        return False
    return True


def _dedup(props):
    """Drop duplicate claims. Two axes: (a) identical (pred,obj,polarity), and
    (b) same (subject,predicate) — which collapses a mis-split compound like
    "detection is post-hoc eval" -> "Detection is post/hoc/eval" into one claim
    (keep the highest-confidence surface)."""
    props = sorted(props, key=lambda p: p.confidence, reverse=True)
    seen_po, seen_sp, out = set(), set(), []
    for p in props:
        subj = (p.subject or "").lower()
        po = (p.predicate, (p.object or "").lower(), p.polarity)
        sp = (subj, p.predicate, p.polarity)
        if po in seen_po or sp in seen_sp:
            continue
        seen_po.add(po)
        seen_sp.add(sp)
        out.append(p)
    return out


# --------------------------------------------------------------------------- #
#  Heading derivation — a clean human heading from a node.
# --------------------------------------------------------------------------- #
_HEADING_RE = re.compile(r"^\s*#{1,4}\s+(.{2,70})\s*$", re.M)
# node-type / system labels that are NOT topical headings
_NONTOPIC_LABEL = re.compile(r"^(memory|node|knowledge|doc|session)[:/]", re.I)


def _titlecase_banner(s: str) -> str:
    """A shouty banner ("CHRONOCEPTION — SCALE-INVARIANCE") makes a fine title
    once Title-cased. Keep short acronyms uppercase."""
    def fix(w):
        core = w.strip("—-:,.")
        if len(core) <= 3 and core.isupper():
            return w  # acronym
        return w.capitalize()
    return " ".join(fix(w) for w in s.split())


def _clean_heading(text: str) -> str | None:
    """First line only, no markdown, capped, banner Title-cased. None if unusable."""
    if not text:
        return None
    line = text.strip().splitlines()[0]
    line = re.sub(r"^#+\s*", "", line).strip().strip("#").strip()
    # cut at a natural break so a long banner heading stays a heading, not a para
    for sep in (" — ", " – ", ": ", ". "):
        if sep in line and len(line) > 48:
            line = line.split(sep)[0].strip()
            break
    if not (3 <= len(line) <= 64):
        return None
    if propositions._is_shouty(line):
        line = _titlecase_banner(line)
    return line or None


def _heading_for(node: dict, fallback: str) -> str:
    label = (node.get("label") or "").strip()
    content = node.get("content") or ""
    candidates: list[str] = []
    # a node-type label ("memory:remembered") is never a topic — skip it
    if label and not _NONTOPIC_LABEL.match(label):
        candidates.append(label)
    m = _HEADING_RE.search(content)
    if m:
        candidates.append(m.group(1))
    # the leading banner/first sentence of the content is often the real title
    first = re.split(r"(?<=[.\n])", content.strip(), maxsplit=1)[0] if content.strip() else ""
    candidates.append(first)
    for c in candidates:
        h = _clean_heading(c)
        if h:
            return h
    return fallback


def plan_document(region: Region, *, title: str, subtitle: str = "",
                  format_spec: dict | None = None,
                  conf_floor: float = 0.55,
                  max_sections: int = 8,
                  max_claims_per_section: int = 6) -> DocumentIR:
    """Region -> DocumentIR skeleton. The geometry dictates the outline."""
    format_spec = format_spec or {}
    doc = DocumentIR(title=title, subtitle=subtitle,
                     seed_id=region.nodes[0]["id"] if region.nodes else None,
                     format_spec=format_spec)

    made = 0
    seen_headings: set[str] = set()
    for node in region.nodes:
        if made >= max_sections:
            break
        props = propositions.extract(node.get("content") or "",
                                     node_id=node.get("id"),
                                     node_importance=float(node.get("importance") or 0.0),
                                     max_sentences=10)
        props = [p for p in props if _clean_prop(p, conf_floor)]
        props = _dedup(props)
        props.sort(key=lambda p: p.confidence, reverse=True)
        props = props[:max_claims_per_section]
        if not props:
            continue
        heading = _heading_for(node, fallback=f"Region {made + 1}")
        # cross-section dedup: a topic appears once. Distinguish by top claim
        # subject, else drop the collision so the outline stays clean.
        if heading.lower() in seen_headings:
            subj = (props[0].subject or "").strip().title()
            alt = f"{heading}: {subj}" if subj and subj.lower() not in heading.lower() else None
            if alt and alt.lower() not in seen_headings and len(alt) <= 64:
                heading = alt
            else:
                continue
        seen_headings.add(heading.lower())
        sec = Section(heading=heading, level=2, seed_ids=[node["id"]])
        # stash the planned propositions on the section for REALIZE
        sec.__dict__["_planned_props"] = props
        sec.__dict__["_node"] = node
        doc.sections.append(sec)
        made += 1

    return doc