refactor(kernel): fence legacy assessment-backed proposal adapter

This commit is contained in:
Shay 2026-06-20 16:38:51 -07:00
parent 4bab163828
commit a1fade8336
8 changed files with 753 additions and 120 deletions

View file

@ -252,4 +252,71 @@ The substantive result of this session is not just “0005 runnable.” It is th
- **Targeted suite**: `tests/test_proportional_decrease_proposal.py`, `tests/test_construction_affordances.py`, `tests/test_problem_frame_builder.py`, `tests/test_problem_frame_contracts.py`, `tests/test_gsm8k_problem_frame_adequacy.py`, `tests/test_gsm8k_morphology_missing_kernel_labels.py`, `tests/test_kernel_no_new_legacy_derivation_surfaces.py` -> **54 passed** in `0.66s`.
- **Smoke suite**: `core.cli test --suite smoke -q` -> Running in background.
---
## PR #844 Addendum — refactor(kernel): fence legacy assessment-backed proposal adapter
### Agent and Session
- **Agent:** gpt55
- **Date:** 2026-06-20
- **Task:** Fence `generate.construction_affordances.make_proposal()` so migrated proposal-first families cannot re-enter the legacy assessment-backed adapter.
- **Branch:** `refactor/kernel-fence-legacy-make-proposal`
### Modules Touched
| File | Change type | Summary |
|---|---|---|
| `generate/construction_affordances.py` | modified | Added `_PROPOSAL_FIRST_FAMILIES`, restored `propose_construction()`, extended `ConstructionProposal` with catalog posture/role metadata, and fenced `make_proposal()` with a deterministic `ValueError` for migrated families. |
| `generate/problem_frame_builder.py` | modified | Emits proposal-first traces for proportional decrease and percent partition before assessment; legacy `make_proposal()` is only used for non-migrated catalog families that might appear later. |
| `generate/problem_frame_contracts.py` | modified | Gates migrated contract dispatch on `frame.proposals` instead of raw relation/process presence, preserving “closeness proposes; bindings ground; contracts determine.” |
| `tests/test_construction_affordances.py` | modified | Replaced legacy status-mapping assertions for migrated families with proposal-first/fence assertions. |
| `tests/test_proportional_decrease_proposal.py` | modified | Updated proportional-decrease seam coverage to assert proposal-first ordering, exact motivating evidence, and proposal-gated contract dispatch. |
| `tests/test_percent_partition_proposal.py` | new | Added percent-partition proposal-first seam coverage mirroring the migrated family contract. |
| `tests/test_construction_proposal_seam.py` | new | Added cross-family seam invariants: pre-assessment proposal factory, assessment authority, proposal-gated dispatch, and bypass of legacy `make_proposal()`. |
### Invariants Verified
- **Versor condition**: Preserved. No algebra, field, vault, or serving paths were touched.
- **Proposal-first seam**: Preserved and tightened. Migrated families now originate only via `propose_construction()` and remain `status="proposed"` until assessed.
- **Assessment authority**: Preserved. `assess_contracts()` still determines runnable vs blocked status from bound frame evidence; proposals no longer smuggle assessment state.
- **Serving parity**: Preserved. All construction families remain `diagnostic_only=True` and `serving_allowed=False`.
### Tests Run
```bash
# exit 0
uv run python -m pytest -q \
tests/test_construction_affordances.py \
tests/test_construction_proposal_seam.py \
tests/test_proportional_decrease_proposal.py \
tests/test_percent_partition_proposal.py \
tests/test_problem_frame_builder.py \
tests/test_problem_frame_contracts.py \
tests/test_kernel_no_new_legacy_derivation_surfaces.py
# exit 0
uv run python -m compileall -q \
generate/construction_affordances.py \
generate/problem_frame.py \
generate/problem_frame_builder.py \
generate/problem_frame_contracts.py
# exit 0
uv run ruff check \
generate/construction_affordances.py \
tests/test_construction_affordances.py \
tests/test_construction_proposal_seam.py
# exit 0
git diff --check
# exit 0
uv run python -m core.cli test --suite smoke -q
```
### Notes
- The worktree snapshot predated the local introduction of `propose_construction()` and the #843 seam tests even though the repository history already contained #841/#842/#843. This slice reconstructed that intended proposal-first seam locally before applying the #844 fence.
- `make_proposal()` remains available as a legacy adapter for future explicitly unmigrated catalog families only; both migrated families are fenced by exact family ID.

View file

@ -24,7 +24,7 @@ Design doctrine (from ADR-0223):
"""
from __future__ import annotations
from dataclasses import dataclass
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING
if TYPE_CHECKING:
@ -160,9 +160,9 @@ class ConstructionProposal:
hazards
status
A proposal is created by ``make_proposal()`` from ProblemFrame evidence.
It does not affect serving. Its ``status`` reflects whether the
corresponding contract assessment declared the frame runnable.
A proposal is created by ``propose_construction()`` from pre-assessment
surface evidence. It does not affect serving. Bound relations ground the
proposal and contract assessments determine whether it closes.
Args:
family_id: Catalog family identifier, e.g.
@ -175,6 +175,9 @@ class ConstructionProposal:
assessment was runnable.
missing_roles: Role obligation names that were absent at assessment time.
active_hazards: Hazard categories that were unresolved at assessment time.
role_obligations: Catalog role obligations carried by the proposed family.
diagnostic_only: Mirrors the catalog family's serving posture.
serving_allowed: Mirrors the catalog family's serving posture.
"""
family_id: str
@ -184,6 +187,9 @@ class ConstructionProposal:
status: str
missing_roles: tuple[str, ...]
active_hazards: tuple[str, ...]
role_obligations: tuple[RoleObligation, ...] = ()
diagnostic_only: bool = True
serving_allowed: bool = False
_VALID_STATUSES: frozenset[str] = frozenset({
"proposed", "partial", "closed", "refused"
@ -374,6 +380,11 @@ _BY_RELATION_TYPE: dict[str, ConstructionFamily] = {
for family in _CATALOG.values()
}
_PROPOSAL_FIRST_FAMILIES: frozenset[str] = frozenset({
"proportional_change.decrease_to_fraction",
"partition.percent_partition",
})
# ---------------------------------------------------------------------------
# Public accessors
@ -405,6 +416,36 @@ def all_diagnostic_families() -> tuple[ConstructionFamily, ...]:
return tuple(_CATALOG[key] for key in sorted(_CATALOG))
def propose_construction(
family_id: str,
evidence_spans: tuple[SourceSpan, ...],
) -> ConstructionProposal:
"""Create a catalog-backed proposal from pre-assessment surface evidence.
This factory records the construction hypothesis only. Role binding and
contract assessments remain the grounding and determination authorities.
Raises:
KeyError: If *family_id* is not registered in the catalog.
"""
family = _CATALOG[family_id]
return ConstructionProposal(
family_id=family.family_id,
relation_type=family.signature.relation_type,
candidate_organ=family.signature.candidate_organ,
evidence_spans=evidence_spans,
status="proposed",
missing_roles=(),
active_hazards=(),
role_obligations=(
*family.signature.required_roles,
*family.signature.optional_roles,
),
diagnostic_only=family.diagnostic_only,
serving_allowed=family.serving_allowed,
)
def make_proposal(
family_id: str,
evidence_spans: tuple[SourceSpan, ...],
@ -412,11 +453,11 @@ def make_proposal(
missing_roles: tuple[str, ...],
active_hazards: tuple[str, ...],
) -> ConstructionProposal:
"""Create a lightweight diagnostic-only ConstructionProposal from frame evidence.
"""Map assessment evidence onto a proposal for legacy catalog paths.
This is a thin factory. The caller supplies evidence already gathered by
``assess_contracts()``; this function only maps that evidence to the
standard proposal trace shape.
Migrated proposal-first families must enter through
:func:`propose_construction`. This adapter remains only for explicitly
unmigrated catalog paths that still synthesize proposals from assessments.
Args:
family_id: Catalog family identifier.
@ -435,8 +476,13 @@ def make_proposal(
Raises:
KeyError: If *family_id* is not registered in the catalog.
ValueError: If *family_id* has already migrated to the proposal-first seam.
"""
family = _CATALOG[family_id] # raises KeyError if absent — intended
if family_id in _PROPOSAL_FIRST_FAMILIES:
raise ValueError(
f"{family_id} is proposal-first; use propose_construction before assessment"
)
proposal = propose_construction(family_id, evidence_spans)
if assessment_runnable:
status = "closed"
@ -447,11 +493,8 @@ def make_proposal(
else:
status = "proposed"
return ConstructionProposal(
family_id=family_id,
relation_type=family.signature.relation_type,
candidate_organ=family.signature.candidate_organ,
evidence_spans=evidence_spans,
return replace(
proposal,
status=status,
missing_roles=missing_roles,
active_hazards=active_hazards,

View file

@ -12,9 +12,11 @@ Non-goals:
"""
from __future__ import annotations
import dataclasses
import re
from fractions import Fraction
from generate.construction_affordances import ConstructionProposal, propose_construction
from generate.kernel_facts import (
BoundRelation,
BoundRole,
@ -319,6 +321,10 @@ _DECREASE_TO_FRACTION_RE = re.compile(
r"(?P<transition>decrease\s+to)\s+(?P<fraction>\d+\s*/\s*\d+)\s+of",
re.IGNORECASE,
)
_PERCENT_OF_PROPOSAL_RE = re.compile(
r"\b\d+(?:\.\d+)?\s*%\s+of\b",
re.IGNORECASE,
)
_DECREASE_STATE_RE = re.compile(
r"(?P<state>[A-Za-z][A-Za-z'-]*)\s+will\s+decrease\s+to",
re.IGNORECASE,
@ -338,6 +344,49 @@ _TRANSFER_RE = re.compile(
)
def _proportional_decrease_proposals(text: str) -> tuple[ConstructionProposal, ...]:
"""Propose proportional decrease from the motivating surface chunk."""
matches = tuple(_DECREASE_TO_FRACTION_RE.finditer(text))
if len(matches) != 1:
return ()
match = matches[0]
evidence = SourceSpan(
text[match.start():match.end()],
match.start(),
match.end(),
)
return (
propose_construction(
"proportional_change.decrease_to_fraction",
(evidence,),
),
)
def _percent_partition_proposals(
text: str,
frames: tuple[ProcessFrame, ...],
) -> tuple[ConstructionProposal, ...]:
"""Propose percent partition from process evidence plus explicit percent-of cues."""
frame_names = {frame.name for frame in frames}
if not frame_names & {"partition", "consumption"}:
return ()
evidence_spans = tuple(
SourceSpan(text[match.start():match.end()], match.start(), match.end())
for match in _PERCENT_OF_PROPOSAL_RE.finditer(text)
)
if not evidence_spans:
return ()
return (
propose_construction(
"partition.percent_partition",
evidence_spans,
),
)
def _extract_mentions(
text: str,
quantities: tuple[GroundedScalar, ...],
@ -749,15 +798,23 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
builder.set_bound_question_target(bound_target)
initial_frame = builder.build()
initial_frame = dataclasses.replace(
initial_frame,
proposals=(
*_proportional_decrease_proposals(problem_text),
*_percent_partition_proposals(problem_text, frames),
),
)
from generate.problem_frame_contracts import assess_contracts, get_contract_family_id
from generate.construction_affordances import make_proposal
assessments = assess_contracts(initial_frame)
proposals = []
proposals = list(initial_frame.proposals)
proposed_family_ids = {proposal.family_id for proposal in proposals}
for assessment in assessments:
family_id = get_contract_family_id(assessment.candidate_organ)
if family_id is not None:
if family_id is not None and family_id not in proposed_family_ids:
proposal = make_proposal(
family_id=family_id,
evidence_spans=assessment.evidence_spans,
@ -766,8 +823,8 @@ def build_problem_frame(problem_text: str) -> ProblemFrame:
active_hazards=assessment.unresolved_hazards,
)
proposals.append(proposal)
proposed_family_ids.add(family_id)
import dataclasses
return dataclasses.replace(initial_frame, proposals=tuple(proposals))

View file

@ -16,7 +16,6 @@ from generate.construction_affordances import (
ConstructionContract,
_DECREASE_TO_FRACTION_FAMILY,
_PERCENT_PARTITION_FAMILY,
make_proposal,
)
from generate.kernel_facts import BoundRelation, SourceSpan
from generate.problem_frame import ProblemFrame
@ -296,11 +295,13 @@ def assess_contracts(frame: ProblemFrame) -> tuple[ContractAssessment, ...]:
"""Return deterministic diagnostic assessments; never admits serving.
Dispatch order:
1. ``decrease_to_fraction`` triggered by relation type presence in
``frame.bound_relations``. Routes to ``assess_fraction_decrease``.
1. ``decrease_to_fraction`` triggered by its proposal-first catalog
family in ``frame.proposals``. Routes to ``assess_fraction_decrease``,
which still determines closure from bound frame evidence.
Registry key: ``_CONTRACT_REGISTRY["fraction_decrease"]``.
2. ``percent_partition`` triggered by process-frame names ``partition``
or ``consumption``. Routes to ``assess_percent_partition``.
2. ``percent_partition`` triggered by its proposal-first catalog family
in ``frame.proposals``. Routes to ``assess_percent_partition``, which
still determines closure from bound frame evidence.
Registry key: ``_CONTRACT_REGISTRY["percent_partition"]``.
3. ``container_packing`` / ``labor_rate`` inline skeleton assessments;
not yet in the catalog registry (added to registry when obligations are
@ -314,10 +315,11 @@ def assess_contracts(frame: ProblemFrame) -> tuple[ContractAssessment, ...]:
results: list[ContractAssessment] = []
# Registry-backed diagnostic families
if any(relation.relation_type == "decrease_to_fraction" for relation in frame.bound_relations):
proposed_family_ids = {proposal.family_id for proposal in frame.proposals}
if _DECREASE_TO_FRACTION_FAMILY.family_id in proposed_family_ids:
# Catalog: _CONTRACT_REGISTRY["fraction_decrease"]
results.append(assess_fraction_decrease(frame))
if frame_names & {"partition", "consumption"}:
if _PERCENT_PARTITION_FAMILY.family_id in proposed_family_ids:
# Catalog: _CONTRACT_REGISTRY["percent_partition"]
results.append(assess_percent_partition(frame))

View file

@ -1,12 +1,15 @@
from __future__ import annotations
import pytest
from generate.construction_affordances import (
ConstructionProposal,
all_diagnostic_families,
lookup_family,
lookup_by_organ,
lookup_by_relation_type,
lookup_family,
make_proposal,
ConstructionProposal,
_CATALOG,
propose_construction,
)
from generate.kernel_facts import SourceSpan
@ -22,7 +25,6 @@ def test_catalog_entries_are_diagnostic_only_and_serving_forbidden() -> None:
def test_catalog_ordering_is_deterministic() -> None:
families = all_diagnostic_families()
ids = [f.family_id for f in families]
# Check that they are sorted
assert ids == sorted(ids)
assert ids == [
"partition.percent_partition",
@ -31,93 +33,94 @@ def test_catalog_ordering_is_deterministic() -> None:
def test_lookups_correctness() -> None:
# 1. lookup_family
f1 = lookup_family("proportional_change.decrease_to_fraction")
assert f1 is not None
assert f1.family_id == "proportional_change.decrease_to_fraction"
fraction_family = lookup_family("proportional_change.decrease_to_fraction")
assert fraction_family is not None
assert lookup_by_organ("fraction_decrease") is fraction_family
assert lookup_by_relation_type("decrease_to_fraction") is fraction_family
f2 = lookup_family("partition.percent_partition")
assert f2 is not None
assert f2.family_id == "partition.percent_partition"
partition_family = lookup_family("partition.percent_partition")
assert partition_family is not None
assert lookup_by_organ("percent_partition") is partition_family
assert lookup_by_relation_type("percent_of") is partition_family
assert lookup_family("invalid_family_id") is None
# 2. lookup_by_organ
assert lookup_by_organ("fraction_decrease") is f1
assert lookup_by_organ("percent_partition") is f2
assert lookup_by_organ("invalid_organ") is None
# 3. lookup_by_relation_type
assert lookup_by_relation_type("decrease_to_fraction") is f1
assert lookup_by_relation_type("percent_of") is f2
assert lookup_by_relation_type("invalid_relation") is None
def test_make_proposal_validation_and_status() -> None:
span = SourceSpan("test text", 0, 9)
@pytest.mark.parametrize(
("family_id", "relation_type", "candidate_organ", "required_roles"),
(
(
"proportional_change.decrease_to_fraction",
"decrease_to_fraction",
"fraction_decrease",
{"base_quantity", "scale", "state_entity", "transition"},
),
(
"partition.percent_partition",
"percent_of",
"percent_partition",
{"whole", "part", "scale"},
),
),
)
def test_propose_construction_is_preassessment_and_catalog_backed(
family_id: str,
relation_type: str,
candidate_organ: str,
required_roles: set[str],
) -> None:
span = SourceSpan("surface cue", 0, 11)
# 1. Closed status (runnable = True)
prop_closed = make_proposal(
family_id="proportional_change.decrease_to_fraction",
evidence_spans=(span,),
assessment_runnable=True,
missing_roles=(),
active_hazards=(),
)
assert prop_closed.family_id == "proportional_change.decrease_to_fraction"
assert prop_closed.relation_type == "decrease_to_fraction"
assert prop_closed.candidate_organ == "fraction_decrease"
assert prop_closed.evidence_spans == (span,)
assert prop_closed.status == "closed"
assert prop_closed.missing_roles == ()
assert prop_closed.active_hazards == ()
proposal = propose_construction(family_id, (span,))
# 2. Refused status (active hazards present)
prop_refused = make_proposal(
family_id="proportional_change.decrease_to_fraction",
evidence_spans=(span,),
assessment_runnable=False,
missing_roles=(),
active_hazards=("unbound_base_quantity",),
)
assert prop_refused.status == "refused"
assert prop_refused.active_hazards == ("unbound_base_quantity",)
assert proposal.family_id == family_id
assert proposal.relation_type == relation_type
assert proposal.candidate_organ == candidate_organ
assert proposal.evidence_spans == (span,)
assert proposal.status == "proposed"
assert proposal.missing_roles == ()
assert proposal.active_hazards == ()
assert proposal.diagnostic_only is True
assert proposal.serving_allowed is False
assert {role.role for role in proposal.role_obligations if role.required} == required_roles
# 3. Partial status (missing roles present, no active hazards)
prop_partial = make_proposal(
family_id="proportional_change.decrease_to_fraction",
evidence_spans=(span,),
assessment_runnable=False,
missing_roles=("base_quantity_unbound",),
active_hazards=(),
)
assert prop_partial.status == "partial"
assert prop_partial.missing_roles == ("base_quantity_unbound",)
# 4. Proposed status (not runnable, no missing roles, no active hazards)
prop_proposed = make_proposal(
family_id="proportional_change.decrease_to_fraction",
evidence_spans=(span,),
assessment_runnable=False,
missing_roles=(),
active_hazards=(),
)
assert prop_proposed.status == "proposed"
@pytest.mark.parametrize(
"family_id",
(
"proportional_change.decrease_to_fraction",
"partition.percent_partition",
),
)
def test_make_proposal_rejects_migrated_proposal_first_families(
family_id: str,
) -> None:
span = SourceSpan("surface cue", 0, 11)
# 5. Invalid family_id raises KeyError
with pytest.raises(KeyError):
with pytest.raises(
ValueError,
match=rf"{family_id} is proposal-first; use propose_construction before assessment",
):
make_proposal(
family_id="invalid_family_id",
family_id=family_id,
evidence_spans=(span,),
assessment_runnable=False,
assessment_runnable=True,
missing_roles=(),
active_hazards=(),
)
def test_make_proposal_still_rejects_unknown_family_ids() -> None:
span = SourceSpan("surface cue", 0, 11)
with pytest.raises(KeyError):
propose_construction("invalid_family_id", (span,))
def test_construction_proposal_status_validation() -> None:
span = SourceSpan("test text", 0, 9)
# Valid status doesn't raise
ConstructionProposal(
family_id="proportional_change.decrease_to_fraction",
relation_type="decrease_to_fraction",
@ -128,7 +131,6 @@ def test_construction_proposal_status_validation() -> None:
active_hazards=(),
)
# Invalid status raises ValueError
with pytest.raises(ValueError, match="status must be one of"):
ConstructionProposal(
family_id="proportional_change.decrease_to_fraction",

View file

@ -0,0 +1,206 @@
"""Cross-family invariants for the proposal-first construction seam."""
from __future__ import annotations
import dataclasses
import inspect
import pytest
import generate.construction_affordances as construction_affordances
import generate.problem_frame_contracts as problem_frame_contracts
from generate.construction_affordances import (
all_diagnostic_families,
propose_construction,
)
from generate.kernel_facts import SourceSpan
from generate.problem_frame_builder import build_problem_frame
from generate.problem_frame_contracts import assess_contracts
FRACTION_DECREASE_CASE = (
"In one hour, Addison mountain's temperature will decrease to 3/4 of its temperature. "
"If the current temperature of the mountain is 84 degrees, what will the temperature "
"decrease by?"
)
TANK_EVIDENCE_CASE = (
"A tank's temperature will decrease to 3/4 of its current temperature. "
"If the current temperature is 84 degrees, how many degrees will it decrease by?"
)
PERCENT_PARTITION_CASE = (
"A school has 100 students. Half of the students are girls, the other half are boys. "
"20% of the girls have dogs at home and 10% of the boys have dogs at home. "
"How many students own dogs?"
)
MIGRATED_CASES = (
(
FRACTION_DECREASE_CASE,
"proportional_change.decrease_to_fraction",
"fraction_decrease",
),
(
PERCENT_PARTITION_CASE,
"partition.percent_partition",
"percent_partition",
),
)
def _proposal(frame, family_id: str):
proposals = [item for item in frame.proposals if item.family_id == family_id]
assert len(proposals) == 1
return proposals[0]
def test_proposal_factory_accepts_only_hypothesis_inputs() -> None:
parameters = inspect.signature(propose_construction).parameters
assert tuple(parameters) == ("family_id", "evidence_spans")
assert not {
"assessment",
"runnable",
"verdict",
"score",
"missing_roles",
"active_hazards",
} & parameters.keys()
def test_every_catalog_family_starts_as_diagnostic_proposal() -> None:
evidence = (SourceSpan("surface cue", 0, 11),)
for family in all_diagnostic_families():
assert family.diagnostic_only is True
assert family.serving_allowed is False
proposal = propose_construction(family.family_id, evidence)
assert proposal.status == "proposed"
assert proposal.missing_roles == ()
assert proposal.active_hazards == ()
assert proposal.diagnostic_only is True
assert proposal.serving_allowed is False
@pytest.mark.parametrize(("problem_text", "family_id", "candidate_organ"), MIGRATED_CASES)
def test_migrated_families_keep_assessment_authority(
problem_text: str,
family_id: str,
candidate_organ: str,
) -> None:
frame = build_problem_frame(problem_text)
proposal = _proposal(frame, family_id)
assert proposal.status == "proposed"
assessments = assess_contracts(frame)
assessment = next(
item for item in assessments if item.candidate_organ == candidate_organ
)
assert _proposal(frame, family_id).status == "proposed"
assert not hasattr(proposal, "runnable")
assert assessment.runnable is True
assert assessment.missing_bindings == ()
assert assessment.unresolved_hazards == ()
@pytest.mark.parametrize(("problem_text", "family_id", "candidate_organ"), MIGRATED_CASES)
def test_migrated_contract_dispatch_requires_a_proposal(
problem_text: str,
family_id: str,
candidate_organ: str,
) -> None:
frame = build_problem_frame(problem_text)
assert _proposal(frame, family_id).status == "proposed"
proposal_free_frame = dataclasses.replace(frame, proposals=())
assert candidate_organ not in {
assessment.candidate_organ
for assessment in assess_contracts(proposal_free_frame)
}
@pytest.mark.parametrize(("problem_text", "family_id", "candidate_organ"), MIGRATED_CASES)
def test_builder_publishes_proposal_before_assessment(
monkeypatch: pytest.MonkeyPatch,
problem_text: str,
family_id: str,
candidate_organ: str,
) -> None:
observed_statuses: list[str] = []
original_assess_contracts = problem_frame_contracts.assess_contracts
def observe_assessment(frame):
observed_statuses.append(_proposal(frame, family_id).status)
return original_assess_contracts(frame)
monkeypatch.setattr(
problem_frame_contracts,
"assess_contracts",
observe_assessment,
)
frame = build_problem_frame(problem_text)
assert observed_statuses == ["proposed"]
assert any(
assessment.candidate_organ == candidate_organ
for assessment in assess_contracts(frame)
)
def test_migrated_families_bypass_legacy_make_proposal(
monkeypatch: pytest.MonkeyPatch,
) -> None:
migrated_family_ids = {family_id for _, family_id, _ in MIGRATED_CASES}
observed_family_ids: list[str] = []
original_make_proposal = construction_affordances.make_proposal
def observe_make_proposal(*args, **kwargs):
family_id = kwargs.get("family_id", args[0] if args else None)
observed_family_ids.append(family_id)
return original_make_proposal(*args, **kwargs)
monkeypatch.setattr(
construction_affordances,
"make_proposal",
observe_make_proposal,
)
for problem_text, family_id, _ in MIGRATED_CASES:
assert _proposal(build_problem_frame(problem_text), family_id)
assert migrated_family_ids.isdisjoint(observed_family_ids)
@pytest.mark.parametrize(
("problem_text", "family_id", "expected_evidence"),
(
(
TANK_EVIDENCE_CASE,
"proportional_change.decrease_to_fraction",
("decrease to 3/4 of",),
),
(
PERCENT_PARTITION_CASE,
"partition.percent_partition",
("20% of", "10% of"),
),
),
)
def test_migrated_proposals_contain_only_exact_motivating_surface_evidence(
problem_text: str,
family_id: str,
expected_evidence: tuple[str, ...],
) -> None:
proposal = _proposal(build_problem_frame(problem_text), family_id)
assert tuple(span.text for span in proposal.evidence_spans) == expected_evidence
assert all(
span.text == problem_text[span.start:span.end]
for span in proposal.evidence_spans
)

View file

@ -0,0 +1,153 @@
"""Percent-partition proposal-first construction seam tests."""
from __future__ import annotations
import dataclasses
import generate.construction_affordances as construction_affordances
import generate.problem_frame_builder as problem_frame_builder
import generate.problem_frame_contracts as problem_frame_contracts
from generate.problem_frame_builder import build_problem_frame
from generate.problem_frame_contracts import assess_contracts
PERCENT_PARTITION_CASE = (
"A school has 100 students. Half of the students are girls, the other half are boys. "
"20% of the girls have dogs at home and 10% of the boys have dogs at home. "
"How many students own dogs?"
)
ONE_SUBGROUP_CONFUSER = (
"There are 100 students. Half are girls. 30% of the girls own pets. "
"How many students own pets?"
)
PERCENT_CHANGE_CONFUSER = "A store spent 20% of its budget. How many dollars remain?"
def _percent_partition_proposal(frame):
return next(
proposal
for proposal in frame.proposals
if proposal.family_id == "partition.percent_partition"
)
def _percent_partition_assessment(frame):
return next(
assessment
for assessment in assess_contracts(frame)
if assessment.candidate_organ == "percent_partition"
)
def test_supported_case_has_diagnostic_catalog_proposal() -> None:
proposal = _percent_partition_proposal(build_problem_frame(PERCENT_PARTITION_CASE))
assert proposal.status == "proposed"
assert proposal.diagnostic_only is True
assert proposal.serving_allowed is False
assert {role.role for role in proposal.role_obligations if role.required} == {
"whole",
"part",
"scale",
}
assert tuple(span.text for span in proposal.evidence_spans) == (
"20% of",
"10% of",
)
assert all(
PERCENT_PARTITION_CASE[span.start:span.end] == span.text
for span in proposal.evidence_spans
)
def test_proposal_precedes_role_binding_and_contract_assessment(monkeypatch) -> None:
events: list[str] = []
original_propose = problem_frame_builder.propose_construction
original_assess = problem_frame_contracts.assess_contracts
def observe_proposal(*args, **kwargs):
proposal = original_propose(*args, **kwargs)
if proposal.family_id == "partition.percent_partition":
events.append("proposal")
return proposal
def observe_assessment(frame):
events.append("assessment")
assert _percent_partition_proposal(frame).status == "proposed"
assert frame.bound_relations
return original_assess(frame)
monkeypatch.setattr(problem_frame_builder, "propose_construction", observe_proposal)
monkeypatch.setattr(problem_frame_contracts, "assess_contracts", observe_assessment)
build_problem_frame(PERCENT_PARTITION_CASE)
assert events == ["proposal", "assessment"]
def test_migrated_family_does_not_use_legacy_assessment_adapter(monkeypatch) -> None:
original_make_proposal = construction_affordances.make_proposal
def reject_migrated_family(*args, **kwargs):
family_id = kwargs.get("family_id", args[0] if args else None)
if family_id == "partition.percent_partition":
raise AssertionError("migrated family reached legacy make_proposal adapter")
return original_make_proposal(*args, **kwargs)
monkeypatch.setattr(
construction_affordances,
"make_proposal",
reject_migrated_family,
)
assert (
_percent_partition_proposal(build_problem_frame(PERCENT_PARTITION_CASE)).status
== "proposed"
)
def test_contract_assessment_remains_runnable_authority() -> None:
frame = build_problem_frame(PERCENT_PARTITION_CASE)
assert _percent_partition_proposal(frame).status == "proposed"
assert _percent_partition_assessment(frame).runnable is True
def test_positive_proposal_does_not_close_missing_subgroup_bindings() -> None:
frame = build_problem_frame(ONE_SUBGROUP_CONFUSER)
assessment = _percent_partition_assessment(frame)
assert _percent_partition_proposal(frame).status == "proposed"
assert assessment.runnable is False
assert "partition_subgroups_not_distinct" in assessment.missing_bindings
assert "percent_subgroup_links_incomplete" in assessment.missing_bindings
def test_percent_partition_dispatch_requires_proposal_first_trace() -> None:
frame = build_problem_frame(PERCENT_PARTITION_CASE)
relation_only_frame = dataclasses.replace(frame, proposals=())
assert {candidate.name for candidate in relation_only_frame.process_frames} & {
"partition",
"consumption",
}
assert any(
relation.relation_type in {"subgroup_partition", "percent_of"}
for relation in relation_only_frame.bound_relations
)
assert not any(
assessment.candidate_organ == "percent_partition"
for assessment in assess_contracts(relation_only_frame)
)
def test_percent_change_confuser_is_proposed_but_not_runnable() -> None:
frame = build_problem_frame(PERCENT_CHANGE_CONFUSER)
assessment = _percent_partition_assessment(frame)
assert _percent_partition_proposal(frame).status == "proposed"
assert assessment.runnable is False
assert "grounded_partition_subgroup" in assessment.missing_bindings
assert "percent_change_vs_percent_of" in assessment.unresolved_hazards

View file

@ -1,8 +1,14 @@
"""Tests for feat(kernel): route proportional-decrease through construction proposals."""
"""Proportional-decrease proposal-first construction seam tests."""
from __future__ import annotations
from generate.problem_frame_builder import build_problem_frame
import dataclasses
import generate.problem_frame_builder as problem_frame_builder
import generate.problem_frame_contracts as problem_frame_contracts
from generate.construction_affordances import lookup_family
from generate.problem_frame_builder import build_problem_frame
from generate.problem_frame_contracts import assess_contracts
FRACTION_DECREASE_CASE = (
@ -20,39 +26,90 @@ AFFINE_CONFUSER = (
"Marion has 1/4 more than what Yun currently has, plus 7. How many paperclips does Marion have?"
)
MULTIPLE_FRACTION_CONFUSER = (
"The reactor's temperature will decrease to 3/4 of its temperature and later decrease "
"to 1/2 of its temperature. If the current temperature is 80 degrees, what will the "
"temperature decrease by?"
)
def test_proposal_precedes_role_binding_and_contract_assessment(monkeypatch) -> None:
events: list[str] = []
original_propose = problem_frame_builder.propose_construction
original_assess = problem_frame_contracts.assess_contracts
def observe_proposal(*args, **kwargs):
events.append("proposal")
return original_propose(*args, **kwargs)
def observe_assessment(frame):
events.append("assessment")
proposal = next(
item
for item in frame.proposals
if item.family_id == "proportional_change.decrease_to_fraction"
)
assert proposal.status == "proposed"
return original_assess(frame)
monkeypatch.setattr(problem_frame_builder, "propose_construction", observe_proposal)
monkeypatch.setattr(problem_frame_contracts, "assess_contracts", observe_assessment)
frame = build_problem_frame(FRACTION_DECREASE_CASE)
assert events == ["proposal", "assessment"]
assert any(
relation.relation_type == "decrease_to_fraction"
for relation in frame.bound_relations
)
def test_proposal_trace_exists_for_decrease_to_fraction() -> None:
frame = build_problem_frame(FRACTION_DECREASE_CASE)
proposals = [p for p in frame.proposals if p.family_id == "proportional_change.decrease_to_fraction"]
proposals = [
proposal
for proposal in frame.proposals
if proposal.family_id == "proportional_change.decrease_to_fraction"
]
assert len(proposals) == 1
proposal = proposals[0]
assert proposal.relation_type == "decrease_to_fraction"
assert proposal.candidate_organ == "fraction_decrease"
assert proposal.status == "closed"
assert proposal.status == "proposed"
assert proposal.missing_roles == ()
assert proposal.active_hazards == ()
assert {role.role for role in proposal.role_obligations if role.required} == {
"base_quantity",
"scale",
"state_entity",
"transition",
}
def test_proposal_trace_includes_exact_evidence_span() -> None:
frame = build_problem_frame(FRACTION_DECREASE_CASE)
proposal = next(p for p in frame.proposals if p.family_id == "proportional_change.decrease_to_fraction")
# Verify evidence spans slice the original text exactly
proposal = next(
proposal
for proposal in frame.proposals
if proposal.family_id == "proportional_change.decrease_to_fraction"
)
for span in proposal.evidence_spans:
assert FRACTION_DECREASE_CASE[span.start:span.end] == span.text
# Verify key tokens/spans are present in the evidence
evidence_texts = {span.text for span in proposal.evidence_spans}
assert "decrease to" in evidence_texts
assert "3/4" in evidence_texts
assert "84" in evidence_texts
assert "temperature" in evidence_texts
assert len(proposal.evidence_spans) == 1
assert proposal.evidence_spans[0].text == "decrease to 3/4 of"
def test_proposal_family_is_catalog_backed() -> None:
frame = build_problem_frame(FRACTION_DECREASE_CASE)
proposal = next(p for p in frame.proposals if p.family_id == "proportional_change.decrease_to_fraction")
# Use public catalog accessor to look up the family
proposal = next(
proposal
for proposal in frame.proposals
if proposal.family_id == "proportional_change.decrease_to_fraction"
)
family = lookup_family(proposal.family_id)
assert family is not None
assert family.family_id == "proportional_change.decrease_to_fraction"
@ -62,26 +119,72 @@ def test_proposal_family_is_catalog_backed() -> None:
def test_proposal_is_diagnostic_only_serving_disallowed() -> None:
frame = build_problem_frame(FRACTION_DECREASE_CASE)
proposal = next(p for p in frame.proposals if p.family_id == "proportional_change.decrease_to_fraction")
proposal = next(
proposal
for proposal in frame.proposals
if proposal.family_id == "proportional_change.decrease_to_fraction"
)
family = lookup_family(proposal.family_id)
assert family is not None
assert family.diagnostic_only is True
assert family.serving_allowed is False
assert proposal.diagnostic_only is True
assert proposal.serving_allowed is False
def test_blocked_final_value_proportional_case_produces_proposal_but_remains_blocked() -> None:
frame = build_problem_frame(FINAL_VALUE_CONFUSER)
proposals = [p for p in frame.proposals if p.family_id == "proportional_change.decrease_to_fraction"]
proposals = [
proposal
for proposal in frame.proposals
if proposal.family_id == "proportional_change.decrease_to_fraction"
]
assert len(proposals) == 1
proposal = proposals[0]
# The final value question target is not a delta/decrease target, so contract is not runnable
assert proposal.status != "closed"
assert "delta_decrease_target_unbound" in proposal.missing_roles
assessment = next(
item for item in assess_contracts(frame)
if item.candidate_organ == "fraction_decrease"
)
assert proposal.status == "proposed"
assert not assessment.runnable
assert "delta_decrease_target_unbound" in assessment.missing_bindings
def test_no_proposal_trace_for_affine_more_than_confuser() -> None:
frame = build_problem_frame(AFFINE_CONFUSER)
proposals = [p for p in frame.proposals if p.family_id == "proportional_change.decrease_to_fraction"]
proposals = [
proposal
for proposal in frame.proposals
if proposal.family_id == "proportional_change.decrease_to_fraction"
]
assert len(proposals) == 0
def test_multiple_fraction_confuser_is_not_proposed_or_assessed() -> None:
frame = build_problem_frame(MULTIPLE_FRACTION_CONFUSER)
assert not any(
proposal.family_id == "proportional_change.decrease_to_fraction"
for proposal in frame.proposals
)
assert not any(
assessment.candidate_organ == "fraction_decrease"
for assessment in assess_contracts(frame)
)
def test_fraction_contract_dispatch_requires_the_proposal_first_trace() -> None:
frame = build_problem_frame(FRACTION_DECREASE_CASE)
relation_only_frame = dataclasses.replace(frame, proposals=())
assert any(
relation.relation_type == "decrease_to_fraction"
for relation in relation_only_frame.bound_relations
)
assert not any(
assessment.candidate_organ == "fraction_decrease"
for assessment in assess_contracts(relation_only_frame)
)