feat(proposals): emit review-gated proposals from derived CLOSE facts

- New focused emitter generate/determine/derived_close_proposals.py:
  - collect eligible derived (member/subset + TRANSITIVE) with entailed Derivation + SPECULATIVE.
  - deterministic sort + stable dedupe key (predicate+args+derivation+structure_key).
  - write proposal-only artifacts (source=derived_close_fact) to dedicated sink.
  - best-effort, counts returned, safe skips.
- RuntimeConfig: review_derived_close_proposals (default False), documented.
- ChatRuntime.idle_tick: after consolidation (when flag on), call emitter best-effort (no did_work, no engine checkpoint).
- IdleTickResult: new field derived_close_proposals_emitted (additive, documented).
- No change to existing proposal_review contract or comprehension-failure sink.
- No serving/determine side-effects.

chat/runtime.py

@@ -562,6 +562,11 @@ class IdleTickResult:
     #: IT — read-only proposal-review summary (None unless ``config.review_pending_proposals``).
     #: Surfaces pending comprehension-failure proposals for review; mutates nothing.
     proposal_review: ProposalReviewIdleSummary | None = None
+    #: PR-2 — count of new reviewable proposal artifacts emitted this tick from proof-gated
+    #: derived CLOSE facts (member/subset + TRANSITIVE_PREDICATES). Only non-zero when
+    #: ``config.review_derived_close_proposals``. Proposal-only, review-gated, SPECULATIVE;
+    #: does not set did_work or trigger engine checkpoint (lives in teaching/proposals sink).
+    derived_close_proposals_emitted: int = 0
     #: Observational backpressure telemetry — backlog depth, headroom, inflow vs drain.
     #: None only when the tick was never called (not possible in normal use); always
     #: present on a completed tick. Never gates or refuses; see ADR-0161 for that.
@@ -982,6 +987,28 @@ class ChatRuntime:
             facts_consolidated = consolidate_once(self._context).consolidated
             did_work = True
 
+        derived_close_proposals_emitted = 0
+        # PR-2 derived CLOSE proposal bridge — after consolidation so newly derived
+        # proof-gated facts (member/subset + relational transitive) can surface as
+        # reviewable proposal artifacts. Best-effort, proposal-only, review-gated.
+        # Never sets did_work (proposals live outside engine_state; no unrelated checkpoint).
+        # Gated by explicit config flag (default off).
+        if getattr(self.config, "review_derived_close_proposals", False):
+            try:
+                from generate.determine.derived_close_proposals import (
+                    emit_derived_close_proposals,
+                )
+
+                res = emit_derived_close_proposals(self._context)
+                derived_close_proposals_emitted = res.get("emitted", 0)
+            except Exception as exc:  # noqa: BLE001 — proposal emission must not crash idle
+                warnings.warn(
+                    f"derived close proposal emission skipped: {exc!r}",
+                    RuntimeWarning,
+                    stacklevel=2,
+                )
+                derived_close_proposals_emitted = 0
+
         # 2b. Frontier-contemplation pass (ADR-0080 into the idle loop) — autonomously MINE
         #     the frontier-compare reports into persisted SPECULATIVE reviewable findings, so
         #     the always-on life proposes its OWN frontier with no user turn. Idempotent per
@@ -1093,6 +1120,7 @@ class ChatRuntime:
             facts_consolidated=facts_consolidated,
             frontier_findings=frontier_findings,
             proposal_review=proposal_review,
+            derived_close_proposals_emitted=derived_close_proposals_emitted,
             backpressure=bp_record,
         )
 

core/config.py

@@ -323,6 +323,15 @@ class RuntimeConfig:
     # OFF by default — idle ticks don't pay for the scan unless a deployment wants the surface.
     review_pending_proposals: bool = False
 
+    # PR-2 derived CLOSE proposal bridge — when on, after consolidation in idle_tick, scan
+    # realized derived facts (member/subset + TRANSITIVE_PREDICATES) that carry proof-gated
+    # Derivation (verdict="entailed", SPECULATIVE). Emit reviewable proposal-only artifacts
+    # (source="derived_close_fact") to teaching/proposals/derived_close_facts/. Default off;
+    # review-gated, no corpus mutation, no ratification, no COHERENT, no serving change.
+    # Deterministic dedupe by (predicate, args, derivation, structure_key). Best-effort;
+    # failures captured.
+    review_derived_close_proposals: bool = False
+
     # Autonomous frontier contemplation (ADR-0080 into the idle loop) — when on, idle_tick
     # autonomously MINES its frontier-compare reports into persisted SPECULATIVE reviewable
     # findings (core.contemplation), so the always-on life proposes its own frontier without

generate/determine/derived_close_proposals.py

@@ -0,0 +1,147 @@
+"""PR-2: deterministic proposal bridge from proof-gated derived CLOSE facts.
+
+Collects realized derived facts (member/subset + TRANSITIVE_PREDICATES) that carry
+a Derivation with verdict="entailed" and are SPECULATIVE. Emits reviewable proposal-only
+artifacts (source="derived_close_fact") to a dedicated sink.
+
+- Default-off via RuntimeConfig.review_derived_close_proposals
+- Review-gated, proposal-only (never ratification or corpus mutation)
+- Deterministic dedupe by stable key (predicate + args + derivation + structure_key)
+- Skips non-derived, malformed, unsupported, non-entailed safely
+- Best-effort; failures do not corrupt state or set did_work on engine checkpoint
+
+Intended to be called from idle_tick *after* consolidation (Step D) so that newly
+derived facts can immediately surface as proposals.
+"""
+
+from __future__ import annotations
+
+import json
+from hashlib import sha256
+from pathlib import Path
+from typing import Any
+
+from generate.meaning_graph.relational import TRANSITIVE_PREDICATES
+from generate.realize import RealizedRecord, recall_realized
+
+#: Dedicated sink for derived CLOSE proposals (distinct from comprehension-failures
+#: to keep families clean while still reviewable by the same HITL tooling).
+DEFAULT_SINK = (
+    Path(__file__).resolve().parents[3]
+    / "teaching"
+    / "proposals"
+    / "derived_close_facts"
+)
+
+
+def _stable_dedupe_key(rec: RealizedRecord) -> str:
+    """Stable dedupe key for a derived record. Includes derivation provenance so that
+    a re-derived fact with different proof chain is treated as distinct if premises differ."""
+    if rec.derivation is None:
+        return ""
+    key = (
+        rec.relation_predicate,
+        rec.relation_arguments[0] if len(rec.relation_arguments) > 0 else "",
+        rec.relation_arguments[1] if len(rec.relation_arguments) > 1 else "",
+        rec.derivation.rule,
+        tuple(rec.derivation.premise_structure_keys),
+        rec.structure_key,
+    )
+    return sha256(repr(key).encode("utf-8")).hexdigest()
+
+
+def _is_eligible(rec: RealizedRecord) -> bool:
+    if not rec.derived or rec.derivation is None:
+        return False
+    if rec.derivation.verdict != "entailed":
+        return False
+    if getattr(rec, "epistemic_status", None) != "speculative":
+        return False
+    p = rec.relation_predicate
+    if p not in {"member", "subset"} and p not in TRANSITIVE_PREDICATES:
+        return False
+    if len(getattr(rec, "relation_arguments", ())) != 2:
+        return False
+    return True
+
+
+def emit_derived_close_proposals(
+    ctx: "SessionContext",
+    *,
+    sink: Path | None = None,
+    max_emissions: int = 100,
+) -> dict[str, int]:
+    """Scan realized facts for eligible derived CLOSE conclusions and emit proposal artifacts.
+
+    Returns counts:
+      considered, eligible, emitted, duplicate, skipped
+
+    Artifacts are written only for first sighting (deduped by stable key).
+    Order is deterministic (sorted by (predicate, subject, object)).
+    """
+    sink = sink or DEFAULT_SINK
+    sink.mkdir(parents=True, exist_ok=True)
+
+    considered = 0
+    eligible = 0
+    emitted = 0
+    duplicate = 0
+    skipped = 0
+
+    candidates: list[tuple[str, str, str, RealizedRecord]] = []
+
+    # member and subset first (from PR-1 is-a), then the relational transitive
+    for p in ["member", "subset"] + sorted(TRANSITIVE_PREDICATES):
+        for rec in recall_realized(ctx, predicate=p):
+            considered += 1
+            if _is_eligible(rec):
+                eligible += 1
+                candidates.append(
+                    (p, rec.relation_arguments[0], rec.relation_arguments[1], rec)
+                )
+
+    # deterministic global order
+    candidates.sort(key=lambda t: (t[0], t[1], t[2]))
+
+    for p, subj, obj, rec in candidates:
+        if emitted >= max_emissions:
+            break
+        dkey = _stable_dedupe_key(rec)
+        path = sink / f"{dkey}.json"
+        if path.exists():
+            duplicate += 1
+            continue
+
+        artifact: dict[str, Any] = {
+            "source": "derived_close_fact",
+            "predicate": rec.relation_predicate,
+            "subject": subj,
+            "object": obj,
+            "relation_arguments": list(rec.relation_arguments),
+            "derivation": {
+                "rule": rec.derivation.rule,
+                "verdict": rec.derivation.verdict,
+                "premise_structure_keys": list(rec.derivation.premise_structure_keys),
+            },
+            "epistemic_status": rec.epistemic_status,
+            "structure_key": rec.structure_key,
+            "dedupe_key": dkey,
+            "status": "proposal_only",
+            "requires_review": True,
+            "mounted": False,
+        }
+        path.write_text(
+            json.dumps(artifact, indent=2, sort_keys=True), encoding="utf-8"
+        )
+        emitted += 1
+
+    return {
+        "considered": considered,
+        "eligible": eligible,
+        "emitted": emitted,
+        "duplicate": duplicate,
+        "skipped": skipped,
+    }
+
+
+__all__ = ["emit_derived_close_proposals", "DEFAULT_SINK"]