Merge pull request #631 from AssetOverflow/feat/contemplation-pack-d

core/comprehension_attempt/__init__.py

@@ -15,12 +15,20 @@ from core.comprehension_attempt.failure_family import (
     family_for_reason,
 )
 from core.comprehension_attempt.model import ComprehensionAttempt, Organ, Outcome
+from core.comprehension_attempt.proposal import (
+    FailureProposal,
+    build_proposal,
+    emit_proposal,
+)
 from core.comprehension_attempt.router import RouteResult, RouteStatus, route_setup
 
 __all__ = [
     "REGISTRY",
     "ComprehensionAttempt",
     "FailureFamily",
+    "FailureProposal",
+    "build_proposal",
+    "emit_proposal",
     "Organ",
     "Outcome",
     "RouteResult",

core/comprehension_attempt/proposal.py

@@ -0,0 +1,148 @@
+"""Proposal-only failure-artifact emitter (N5) — deliberately toothless.
+
+When the contemplation pass (N6) meets a **growth-surface** failure family (``proposal_allowed
+= True``), it may emit a single content-addressed JSON artifact under
+``teaching/proposals/comprehension_failures/<hash>.json``. That artifact can *propose* a next
+fixture/rule for human review. It can do nothing else:
+
+```text
+status is always "proposal_only"
+mounted is always false
+requires_review is always true
+serving never reads these files
+no reader/test is modified
+```
+
+This routes into the existing proposal-only teaching flywheel (ADR-0055/0056/0057) — it is NOT a
+parallel correction path (CLAUDE.md teaching-safety). The alignment is
+``failure -> classification -> proposal -> review -> ratification``, never ``failure -> self-patch``.
+
+Content-addressing: the filename is ``sha256(failure_family : sha256(problem_text))`` — so the
+same failure on the same text always writes the same path (idempotent), and the raw problem text
+is **hashed, never stored**. Deterministic; no clock, no randomness.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from core.comprehension_attempt.failure_family import FailureFamily
+from core.comprehension_attempt.model import ComprehensionAttempt
+
+_PROPOSAL_STATUS = "proposal_only"
+
+
+def default_proposal_root() -> Path:
+    """``<repo>/teaching/proposals/comprehension_failures`` — the write-only proposal sink."""
+    return Path(__file__).resolve().parents[2] / "teaching" / "proposals" / "comprehension_failures"
+
+
+def _sha256(text: str) -> str:
+    return hashlib.sha256(text.encode("utf-8")).hexdigest()
+
+
+def _attempt_to_dict(attempt: ComprehensionAttempt) -> dict[str, Any]:
+    return {
+        "organ": attempt.organ,
+        "outcome": attempt.outcome,
+        "refusal_reason": attempt.refusal_reason,
+        "family": attempt.family,
+        "setup_signature": attempt.setup_signature,
+        "answer": attempt.answer,
+    }
+
+
+@dataclass(frozen=True, slots=True)
+class FailureProposal:
+    """A proposal-only artifact. The invariant fields (``status``/``requires_review``/``mounted``)
+    are enforced in ``__post_init__`` so even a hand-constructed proposal cannot be made
+    serving-mountable."""
+
+    failure_family: str
+    problem_text_sha256: str
+    observed_attempts: tuple[dict[str, Any], ...]
+    status: str = _PROPOSAL_STATUS
+    suggested_next_fixture: None = None  # v0: always None — a human authors the fixture on review
+    requires_review: bool = True
+    mounted: bool = False
+
+    def __post_init__(self) -> None:
+        if self.status != _PROPOSAL_STATUS:
+            raise ValueError(f"proposal status must be {_PROPOSAL_STATUS!r}; got {self.status!r}")
+        if self.mounted:
+            raise ValueError("a proposal can never be mounted")
+        if not self.requires_review:
+            raise ValueError("a proposal always requires review")
+
+    def content_hash(self) -> str:
+        """Deterministic content address: same failure on same text -> same hash."""
+        return hashlib.sha256(
+            f"{self.failure_family}:{self.problem_text_sha256}".encode("utf-8")
+        ).hexdigest()
+
+    def to_json_dict(self) -> dict[str, Any]:
+        return {
+            "status": self.status,
+            "failure_family": self.failure_family,
+            "problem_text_sha256": self.problem_text_sha256,
+            "observed_attempts": list(self.observed_attempts),
+            "suggested_next_fixture": self.suggested_next_fixture,
+            "requires_review": self.requires_review,
+            "mounted": self.mounted,
+        }
+
+
+def build_proposal(
+    text: str, family: FailureFamily, attempts: tuple[ComprehensionAttempt, ...]
+) -> FailureProposal | None:
+    """Build a proposal for a growth-surface family, or ``None`` for a correct wrong=0 boundary.
+
+    A family with ``proposal_allowed = False`` (every must-remain-refused boundary) yields NO
+    proposal — the loop never proposes against a faithful refusal.
+    """
+    if not family.proposal_allowed:
+        return None
+    return FailureProposal(
+        failure_family=family.name,
+        problem_text_sha256=_sha256(text),
+        observed_attempts=tuple(_attempt_to_dict(a) for a in attempts),
+    )
+
+
+def proposal_path(proposal: FailureProposal, root: Path | None = None) -> Path:
+    base = root if root is not None else default_proposal_root()
+    return base / f"{proposal.content_hash()}.json"
+
+
+def emit_proposal(
+    text: str,
+    family: FailureFamily,
+    attempts: tuple[ComprehensionAttempt, ...],
+    *,
+    root: Path | None = None,
+) -> Path | None:
+    """Write a proposal-only artifact for a growth-surface family; return its path, or ``None``.
+
+    Idempotent: the same failure on the same text writes the same content-addressed path with
+    byte-identical content (``sort_keys``). Creates the sink directory on demand.
+    """
+    proposal = build_proposal(text, family, attempts)
+    if proposal is None:
+        return None
+    path = proposal_path(proposal, root)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(proposal.to_json_dict(), indent=2, sort_keys=True), encoding="utf-8")
+    return path
+
+
+__all__ = [
+    "FailureProposal",
+    "build_proposal",
+    "default_proposal_root",
+    "emit_proposal",
+    "proposal_path",
+]

teaching/proposals/comprehension_failures/README.md

@@ -0,0 +1,45 @@
+# Comprehension failure proposals (proposal-only)
+
+This directory is the **write-only sink** for the contemplation pass (N5/N6). When the loop meets
+a *growth-surface* failure family (`proposal_allowed = True` in the N4 registry), it may emit one
+content-addressed artifact here:
+
+```
+<sha256(failure_family : sha256(problem_text))>.json
+```
+
+Every artifact is **deliberately toothless**:
+
+```json
+{
+  "status": "proposal_only",
+  "mounted": false,
+  "requires_review": true,
+  "suggested_next_fixture": null,
+  ...
+}
+```
+
+## Hard rules (enforced by tests)
+
+- `status` is always `proposal_only`; `mounted` is always `false`; `requires_review` is always `true`.
+- **Serving never reads these files.** No `generate/derivation`, `core/reliability_gate`,
+  `generate/stream.py`, `field/propagate.py`, or `vault/store.py` references this path.
+- The raw problem text is **hashed, never stored** (`problem_text_sha256`).
+- The emitter never proposes against a correct wrong=0 boundary (`must_remain_refused` families
+  produce no artifact).
+- Filenames are content-addressed and deterministic — the same failure writes the same path.
+
+## What happens next
+
+A proposal is an *input to human review*, not a change. The aligned flow is
+
+```
+failure -> classification -> proposal -> review -> ratification
+```
+
+never `failure -> self-patch`. Ratification (authoring the gold fixture / reader rule) happens
+only via a human-reviewed PR through the existing teaching flywheel (ADR-0055/0056/0057). The
+engine cannot mount, ratify, or apply anything written here.
+
+Generated artifacts are not committed; this README is the only tracked file in the directory.

tests/test_failure_proposal.py

@@ -0,0 +1,87 @@
+"""Tests for the proposal-only failure-artifact emitter (N5).
+
+Pins the toothless guarantees: proposals are only emitted for growth-surface families, are
+content-addressed + deterministic + idempotent, carry `status=proposal_only` / `mounted=false`,
+hash (never store) the problem text, and are never referenced by any serving-path module.
+"""
+
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from core.comprehension_attempt import classify_r1, classify_r2, family_by_name
+from core.comprehension_attempt.proposal import (
+    FailureProposal,
+    build_proposal,
+    emit_proposal,
+)
+
+_MISSING_TOTAL = "Each large bus holds 50 students and each small bus holds 30 students. The buses carry 260 students in total. How many large buses are there?"
+
+
+def _attempts(text: str):
+    return (classify_r1(text), classify_r2(text))
+
+
+def test_emit_writes_proposal_only_for_growth_surface(tmp_path: Path) -> None:
+    family = family_by_name("missing_total_count")
+    path = emit_proposal(_MISSING_TOTAL, family, _attempts(_MISSING_TOTAL), root=tmp_path)
+    assert path is not None and path.exists()
+    doc = json.loads(path.read_text())
+    assert doc["status"] == "proposal_only"
+    assert doc["mounted"] is False and doc["requires_review"] is True
+    assert doc["failure_family"] == "missing_total_count"
+    assert doc["suggested_next_fixture"] is None
+
+
+def test_no_proposal_for_a_correct_boundary(tmp_path: Path) -> None:
+    family = family_by_name("unsupported_system_size")  # too_many_categories — must stay refused
+    text = "A lot has 10 vehicles. Each car has 4 wheels, each motorcycle has 2 wheels, and each truck has 6 wheels. Together the vehicles have 34 wheels. How many cars are there?"
+    assert build_proposal(text, family, _attempts(text)) is None
+    assert emit_proposal(text, family, _attempts(text), root=tmp_path) is None
+    assert list(tmp_path.glob("*.json")) == []
+
+
+def test_content_addressed_filename_is_deterministic(tmp_path: Path) -> None:
+    family = family_by_name("missing_total_count")
+    a = emit_proposal(_MISSING_TOTAL, family, _attempts(_MISSING_TOTAL), root=tmp_path)
+    b = emit_proposal(_MISSING_TOTAL, family, _attempts(_MISSING_TOTAL), root=tmp_path)
+    assert a == b  # same failure -> same path
+    assert len(list(tmp_path.glob("*.json"))) == 1  # idempotent: one file
+    assert a.read_text() == b.read_text()  # byte-identical
+
+
+def test_problem_text_is_hashed_not_stored(tmp_path: Path) -> None:
+    family = family_by_name("missing_total_count")
+    path = emit_proposal(_MISSING_TOTAL, family, _attempts(_MISSING_TOTAL), root=tmp_path)
+    raw = path.read_text()
+    assert "large bus" not in raw and "students" not in raw  # the prose is not embedded
+    assert json.loads(raw)["problem_text_sha256"]
+
+
+def test_proposal_invariants_cannot_be_overridden() -> None:
+    with pytest.raises(ValueError):
+        FailureProposal("missing_total_count", "abc", (), status="ratified")
+    with pytest.raises(ValueError):
+        FailureProposal("missing_total_count", "abc", (), mounted=True)
+    with pytest.raises(ValueError):
+        FailureProposal("missing_total_count", "abc", (), requires_review=False)
+
+
+def test_serving_path_never_references_proposals() -> None:
+    # No serving-path module reads the proposal sink. Scan the forbidden/serving sources.
+    repo = Path(__file__).resolve().parents[1]
+    serving = [
+        repo / "generate" / "stream.py",
+        repo / "field" / "propagate.py",
+        repo / "vault" / "store.py",
+        repo / "generate" / "derivation",
+        repo / "core" / "reliability_gate",
+    ]
+    for target in serving:
+        files = [target] if target.is_file() else sorted(target.rglob("*.py")) if target.exists() else []
+        for f in files:
+            assert "comprehension_failures" not in f.read_text(encoding="utf-8"), f