core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

fix(ask): preserve boundary-first order for ASK delivery

Browse source
This commit is contained in:
Shay 2026-06-09 10:39:58 -07:00
parent de13082ce0
commit c4ddc69b87
2 changed files with 109 additions and 26 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

44
generate/contemplation/pass_manager.py
View file

@ -94,6 +94,7 @@ def _handle_ask_delivery(
attempts: tuple[ComprehensionAttempt, ...],
text: str,
proposal_root: Path | None,
question_root: Path | None,
exercise_ask: bool,
selected_organ: str | None = None,
) -> ContemplationResult:
@ -103,7 +104,7 @@ def _handle_ask_delivery(
import json
from core.epistemic_questions.delivery import question_path
path = question_path(outcome.question, None)
path = question_path(outcome.question, question_root)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps(outcome.question.to_json_dict(), indent=2, sort_keys=True),
@ -142,6 +143,7 @@ def contemplate(
options: dict[str, Any] | None = None,
answer_key: str | None = None,
proposal_root: Path | None = None,
question_root: Path | None = None,
case_id: str | None = None,
exercise_ask: bool = False,
) -> ContemplationResult:
@ -171,11 +173,11 @@ def contemplate(
if route.status == "routed":
assert route.selected is not None
if route.selected.organ == "r2_constraints":
return _solve_and_verify_r2(text, options, answer_key, findings, attempts, proposal_root, exercise_ask)
return _solve_and_verify_r2(text, options, answer_key, findings, attempts, proposal_root, question_root, exercise_ask)
if route.selected.organ == "r3_rate":
return _solve_and_verify_r3(text, options, answer_key, findings, attempts, proposal_root, exercise_ask)
return _solve_and_verify_r3(text, options, answer_key, findings, attempts, proposal_root, question_root, exercise_ask)
if route.selected.organ == "r4_combined_rate":
return _solve_and_verify_cmb(text, options, answer_key, findings, attempts, proposal_root, exercise_ask)
return _solve_and_verify_cmb(text, options, answer_key, findings, attempts, proposal_root, question_root, exercise_ask)
findings.append(Finding("solve", "r1 admissible setup (numeric answer is the eval lane in v0)"))
findings.append(Finding("terminal", Terminal.SOLVED_VERIFIED.value))
return ContemplationResult(
@ -183,7 +185,7 @@ def contemplate(
)
# route.status == "all_refused"
return _classify_all_refused(text, attempts, findings, proposal_root, exercise_ask)
return _classify_all_refused(text, attempts, findings, proposal_root, question_root, exercise_ask)
def _solve_and_verify_r2(
@ -193,6 +195,7 @@ def _solve_and_verify_r2(
findings: list[Finding],
attempts: tuple[ComprehensionAttempt, ...],
proposal_root: Path | None,
question_root: Path | None,
exercise_ask: bool,
) -> ContemplationResult:
problem = read_constraint_problem(text)
@ -208,7 +211,7 @@ def _solve_and_verify_r2(
if assessment.resolution_action == "ask_question":
if exercise_ask:
return _handle_ask_delivery(
assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
assessment, family_obj.name, findings, attempts, text, proposal_root, question_root, exercise_ask,
selected_organ="r2_constraints"
)
findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value))
@ -249,6 +252,7 @@ def _solve_and_verify_r3(
findings: list[Finding],
attempts: tuple[ComprehensionAttempt, ...],
proposal_root: Path | None,
question_root: Path | None,
exercise_ask: bool,
) -> ContemplationResult:
problem = read_rate_problem(text)
@ -264,7 +268,7 @@ def _solve_and_verify_r3(
if assessment.resolution_action == "ask_question":
if exercise_ask:
return _handle_ask_delivery(
assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
assessment, family_obj.name, findings, attempts, text, proposal_root, question_root, exercise_ask,
selected_organ="r3_rate"
)
findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value))
@ -305,6 +309,7 @@ def _solve_and_verify_cmb(
findings: list[Finding],
attempts: tuple[ComprehensionAttempt, ...],
proposal_root: Path | None,
question_root: Path | None,
exercise_ask: bool,
) -> ContemplationResult:
problem = read_combined_rate_problem(text)
@ -324,7 +329,7 @@ def _solve_and_verify_cmb(
if assessment.resolution_action == "ask_question":
if exercise_ask:
return _handle_ask_delivery(
assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
assessment, family_obj.name, findings, attempts, text, proposal_root, question_root, exercise_ask,
selected_organ="r4_combined_rate"
)
findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value))
@ -363,6 +368,7 @@ def _classify_all_refused(
attempts: tuple[ComprehensionAttempt, ...],
findings: list[Finding],
proposal_root: Path | None,
question_root: Path | None,
exercise_ask: bool,
) -> ContemplationResult:
# CMB-over-R3 precedence (family side): when CMB substantively recognized the text, R3's broader
@ -372,19 +378,9 @@ def _classify_all_refused(
if cmb_is_authoritative(attempts):
considered = tuple(a for a in attempts if a.organ != "r3_rate")
# Check for ASK delivery.
for attempt in considered:
from core.epistemic_disclosure.limitation import assess_from_attempt
assessment = assess_from_attempt(attempt)
if assessment is not None and assessment.resolution_action == "ask_question":
if exercise_ask:
return _handle_ask_delivery(
assessment, assessment.blocking_reason, findings, attempts, text, proposal_root, exercise_ask
)
families = [(a, family_for_reason(a.refusal_reason)) for a in considered]
# Boundary-first: a substantive recognized boundary blocks any proposal.
# Boundary-first: a substantive recognized boundary blocks any proposal or ASK.
for _attempt, family in families:
if family is not None and family.must_remain_refused and family.name != _NOT_MY_DOMAIN:
terminal = (
@ -395,6 +391,16 @@ def _classify_all_refused(
findings.append(Finding("terminal", f"{terminal.value} via {family.name}"))
return ContemplationResult(terminal, tuple(findings), attempts, family=family.name)
# Check for ASK delivery only after substantive boundaries are ruled out.
for attempt in considered:
from core.epistemic_disclosure.limitation import assess_from_attempt
assessment = assess_from_attempt(attempt)
if assessment is not None and assessment.resolution_action == "ask_question":
if exercise_ask:
return _handle_ask_delivery(
assessment, assessment.blocking_reason, findings, attempts, text, proposal_root, question_root, exercise_ask
)
# No substantive boundary: a genuine growth surface may emit a proposal-only artifact.
for _attempt, family in families:
if family is not None and family.proposal_allowed:

91
tests/test_ask_pass_manager_delivery.py
View file

@ -139,8 +139,18 @@ def test_renderable_ask_path_returns_question_needed_under_exercise_ask(monkeypa
monkeypatch.setattr(pm, "route_setup", lambda text, case_id=None: RouteResult((attempt,), None, "all_refused"))
question_root = tmp_path / "teaching" / "questions"
proposal_root = tmp_path / "teaching" / "proposals"
repo_questions_dir = Path(__file__).resolve().parents[1] / "teaching" / "questions"
before_files = set(repo_questions_dir.glob("**/*")) if repo_questions_dir.exists() else set()
# 1. Assert that without exercise_ask, it falls back to PROPOSAL_EMITTED (due to carve-out)
res_normal = contemplate("chickens", proposal_root=tmp_path)
res_normal = contemplate(
"chickens",
proposal_root=proposal_root,
question_root=question_root,
)
assert res_normal.terminal == Terminal.PROPOSAL_EMITTED
# 2. Assert that with exercise_ask=True, it returns QUESTION_NEEDED
@ -151,7 +161,12 @@ def test_renderable_ask_path_returns_question_needed_under_exercise_ask(monkeypa
return orig(assessment)
monkeypatch.setattr(pm, "_delivery_outcome_for_limitation", wrapped)
res_ask = contemplate("chickens", proposal_root=tmp_path, exercise_ask=True)
res_ask = contemplate(
"chickens",
proposal_root=proposal_root,
question_root=question_root,
exercise_ask=True,
)
assert res_ask.terminal == Terminal.QUESTION_NEEDED
assert len(calls) == 1 # Verify it is called exactly once! No double delivery / double render!
@ -159,9 +174,15 @@ def test_renderable_ask_path_returns_question_needed_under_exercise_ask(monkeypa
assert res_ask.proposal_path is not None
artifact_path = Path(res_ask.proposal_path)
assert artifact_path.exists()
# Confirm it is NOT written under proposal_root / teaching/proposals
assert "teaching/questions" in res_ask.proposal_path
assert str(tmp_path) not in res_ask.proposal_path
# Assert question artifact is under question_root
assert question_root in artifact_path.parents
# Assert question artifact is not under proposal_root
assert proposal_root not in artifact_path.parents
# Assert no repo-local teaching/questions artifact is created during tests
after_files = set(repo_questions_dir.glob("**/*")) if repo_questions_dir.exists() else set()
assert before_files == after_files
def test_unrenderable_ask_falls_back_in_pass_manager(monkeypatch, tmp_path) -> None:
@ -182,7 +203,15 @@ def test_unrenderable_ask_falls_back_in_pass_manager(monkeypatch, tmp_path) -> N
lambda assessment: DeliveryOutcome(Terminal.PROPOSAL_EMITTED, None, "multi_slot_not_supported")
)
res = contemplate("chickens", proposal_root=tmp_path, exercise_ask=True)
question_root = tmp_path / "teaching" / "questions"
proposal_root = tmp_path / "teaching" / "proposals"
res = contemplate(
"chickens",
proposal_root=proposal_root,
question_root=question_root,
exercise_ask=True,
)
assert res.terminal == Terminal.PROPOSAL_EMITTED
assert res.terminal is not Terminal.QUESTION_NEEDED
assert res.proposal_path is not None
@ -212,5 +241,53 @@ def test_family_none_does_not_crash_ask_branch(monkeypatch, tmp_path) -> None:
monkeypatch.setattr(pm, "route_setup", lambda text, case_id=None: RouteResult((attempt,), None, "all_refused"))
monkeypatch.setattr(lim_mod, "assess_from_attempt", lambda att: fake_assessment)
res = contemplate("chickens", proposal_root=tmp_path, exercise_ask=True)
question_root = tmp_path / "teaching" / "questions"
proposal_root = tmp_path / "teaching" / "proposals"
res = contemplate(
"chickens",
proposal_root=proposal_root,
question_root=question_root,
exercise_ask=True,
)
assert res.terminal == Terminal.NO_PROGRESS
def test_boundary_wins_over_ask_in_pass_manager(monkeypatch, tmp_path) -> None:
from core.comprehension_attempt import ComprehensionAttempt, RouteResult
import generate.contemplation.pass_manager as pm
from generate.binding_graph.model import SourceSpanLink
span = SourceSpanLink(source_id="src", start=0, end=8, text="chickens")
attempt_boundary = ComprehensionAttempt(
organ="r2_constraints",
outcome="setup_refused",
refusal_reason="too_many_categories", # maps to unsupported_system_size (must_remain_refused = True)
evidence=(span,),
)
attempt_ask = ComprehensionAttempt(
organ="r2_constraints",
outcome="setup_refused",
refusal_reason="missing_total_count", # ask carve-out
evidence=(span,),
)
monkeypatch.setattr(
pm,
"route_setup",
lambda text, case_id=None: RouteResult((attempt_boundary, attempt_ask), None, "all_refused"),
)
question_root = tmp_path / "teaching" / "questions"
proposal_root = tmp_path / "teaching" / "proposals"
res = contemplate(
"chickens",
proposal_root=proposal_root,
question_root=question_root,
exercise_ask=True,
)
# Assert terminal remains REFUSED_UNSUPPORTED_FAMILY or REFUSED_KNOWN_BOUNDARY, not QUESTION_NEEDED
assert res.terminal == Terminal.REFUSED_UNSUPPORTED_FAMILY
# Ensure no question is written under question_root
assert not question_root.exists() or len(list(question_root.glob("**/*"))) == 0