rune/scripts/research_memory_port.py

"""research_memory_port.py — T21.5 §10-12: the future-facing Memory port
inside the EXISTING INWARD research stage.

BOUNDARIES (G9/G10/G11):
  - This is a PORT, not a memory system. No storage is created here; the
    implementation delegates to a provider callable that future Memory
    will supply. Tests inject doubles; INWARD never imports Memory.
  - Zero result is a NORMAL SUCCESSFUL result: the caller continues the
    normal INWARD flow immediately (no extra model call, no ceremony).
  - Returned material is EVIDENCE/CANDIDATES only — never an instruction
    authority. APPLICABILITY (existing stage) governs whether a candidate
    applies; SYNTHESIS receives only distilled, applicable consequences.
  - Teachable-moment seam: durable construction evidence can mark an
    episode as a CANDIDATE for future learning ("may be worth learning",
    never "always apply"). Substrate/Memory (future) harvest candidates;
    today the seam only records them via the same provider contract.
"""
from __future__ import annotations

import os
from typing import Any, Callable

# The provider contract: ONE callable, independently replaceable/mockable.
# Signature: query(query_text, keywords, *, limit) -> list[MemoryCandidate]
# MemoryCandidate (dict): {source, kind, summary, relevance, episode_ref}
#   kind ∈ {construction_episode, teachable_moment_candidate,
#           prior_failure, prior_success}
# Future Memory supplies the real implementation; nothing else changes.

MemoryProvider = Callable[..., list[dict]]

_DEFAULT_PROVIDER: MemoryProvider | None = None


def set_memory_provider(provider: MemoryProvider | None) -> None:
    """Bind the Memory-side implementation (production or test double).
    None disables the port — INWARD continues without it (fast path)."""
    global _DEFAULT_PROVIDER
    _DEFAULT_PROVIDER = provider


def _provider() -> MemoryProvider | None:
    global _DEFAULT_PROVIDER
    if _DEFAULT_PROVIDER is not None:
        return _DEFAULT_PROVIDER
    env = os.environ.get("RUNE_MEMORY_PORT_PROVIDER")
    if env:
        import importlib.util
        spec = importlib.util.spec_from_file_location("mprov", env)
        if spec is not None and spec.loader is not None:
            mod = importlib.util.module_from_spec(spec)
            spec.loader.exec_module(mod)
            provider = getattr(mod, "query", None)
            if provider is not None:
                _DEFAULT_PROVIDER = provider
    return _DEFAULT_PROVIDER


MAX_CANDIDATES = 5
MIN_RELEVANCE = 0.3


def memory_port_of_call(issue: dict, keywords: list[str],
                        *, provider: MemoryProvider | None = None) -> list[dict]:
    """The ONE internal port of call inside INWARD. Bounded retrieval of
    internal evidence (construction episodes, teachable-moment
    candidates, prior failures/successes).

    ZERO-RESULT FAST PATH (§11): empty/thin results return [] and the
    caller continues the next normal INWARD step directly. No extra
    model call, no fallback reasoning, no synthetic summary, no delay.
    """
    prov = provider or _provider()
    if prov is None:
        return []  # port unbound: normal INWARD continues, zero overhead
    try:
        query_text = " ".join([
            str(issue.get("title", "")),
            str(issue.get("body", ""))[:400],
        ]).strip()
        raw = prov(query_text, keywords, limit=MAX_CANDIDATES) or []
    except Exception:  # noqa: BLE001 — port failure must never break INWARD
        return []
    candidates = [c for c in raw
                  if float(c.get("relevance", 0.0)) >= MIN_RELEVANCE]
    return candidates[:MAX_CANDIDATES]


def applicability_filter(candidates: list[dict],
                         context_keywords: list[str]) -> tuple[list[dict],
                                                              list[dict]]:
    """Existing APPLICABILITY governs transfer: candidates relevant to
    the CURRENT context pass; irrelevant ones are rejected WITHOUT
    contaminating downstream SYNTHESIS. Returns (applicable, rejected)."""
    applicable, rejected = [], []
    for c in candidates:
        text = (str(c.get("summary", "")) + " " +
                " ".join(str(k) for k in c.get("keywords", []))).lower()
        overlap = sum(1 for k in context_keywords if k and k.lower() in text)
        if overlap > 0 or c.get("explicitly_applicable") is True:
            applicable.append(c)
        else:
            rejected.append(c)
    return applicable, rejected


# ── §12: teachable-moment future seam ──────────────────────────────────────

CANDIDATE_TRIGGERS = (
    "completed_release", "capability_completed", "failure_recovered",
    "rollback_executed", "user_correction", "prediction_observation_mismatch",
    "procedure_improved",
)


def mark_teachable_moment_candidate(
        episode_ref: str, *, trigger: str, summary: str,
        emit: Callable[[dict], Any] | None = None) -> dict | None:
    """Record durable evidence that an episode MAY be worth future
    learning — a candidate, never a rule. Uses the existing evidence/
    event mechanism via `emit` (the caller's channel: journal, event
    bridge, ledger history). No new canonical store is created here;
    future Substrate/Memory harvests candidates through the port."""
    if trigger not in CANDIDATE_TRIGGERS:
        raise ValueError(
            f"unknown trigger {trigger!r} — must be one of "
            f"{CANDIDATE_TRIGGERS}")
    record = {
        "candidate": "teachable_moment",
        "episode_ref": episode_ref,
        "trigger": trigger,
        "summary": summary[:200],
        "status": "CANDIDATE",   # never authority, never auto-applied
        "note": "may be worth future learning; APPLICABILITY governs "
                "any future transfer",
    }
    if emit is not None:
        emit(record)
    return record