from __future__ import annotations import asyncio import json from gap_pipeline.clients import ScriptedClient from gap_pipeline.kernel_models import MethodPlan, ProofDAG, ReplacementPlan from gap_pipeline.kernel_prompts import diffusion_user from gap_pipeline.paper_pipeline import PaperKernelPipeline, PaperPipelineConfig from gap_pipeline.prompts import JUDGE_SYSTEM_PROMPT from gap_pipeline.store import RunStore def _dag() -> dict: return { "nodes": [ { "node_id": "n1", "claim": "(a-1)^2 >= 0", "dependencies": [], }, { "node_id": "n2", "claim": "a+1/a >= 2", "dependencies": ["n1"], }, ], "terminal_node_id": "n2", } def _methods() -> dict: return { "nodes": [ { "node_id": "n1", "method_label": "use nonnegativity of a square", }, { "node_id": "n2", "method_label": "expand and divide by a positive quantity", }, ] } def _replacement() -> dict: return { "changes": [ { "slot_id": "slot1", "source_node_id": "n1", "description": "positive equality value", "original_value": "1", "replacement_value": "2", "guard_condition": "replacement is positive", "guard_justification": "2 is positive", } ], "closure_statement": "No undeclared changes.", } def _diffused() -> dict: return { "nodes": [ { "node_id": "n1", "dependencies": [], "method_label": "use nonnegativity of a square", "instantiated_claim": "(x-2)^2 >= 0", "justification": "squares are nonnegative", }, { "node_id": "n2", "dependencies": ["n1"], "method_label": "expand and divide by a positive quantity", "instantiated_claim": "x+4/x >= 4", "justification": "expand and divide by x>0", }, ], "terminal_node_id": "n2", "terminal_answer": "4", } def _variant(suffix: str = "") -> dict: return { "question": f"Let x>0. Prove that x+4/x >= 4.{suffix}", "solution": ( "[n1] Since (x-2)^2 >= 0. " "[n2] Expand and divide by x>0 to obtain x+4/x >= 4." ), "node_order": ["n1", "n2"], "terminal_answer": "4", } def _accept() -> dict: return { "verdict": "accept", "step_by_step_check": "n1 is valid; n2 is valid", "blocking_issues": "", "patch_suggestion": "", } def _reject() -> dict: return { "verdict": "reject", "step_by_step_check": "n1 is valid; n2 needs a wording correction", "blocking_issues": "the terminal wording is ambiguous", "patch_suggestion": "clarify the terminal wording", } def test_five_stage_repair_uses_prior_bundle_and_appendix_judge( tmp_path, item, ) -> None: proposer = ScriptedClient( { f"{item.item_id}.stage1.dag": _dag(), f"{item.item_id}.stage2.methods": _methods(), f"{item.item_id}.stage3.replacement": [ _replacement(), _replacement(), ], f"{item.item_id}.stage4.diffusion": [ _diffused(), _diffused(), ], f"{item.item_id}.stage5.render": [ _variant(), _variant(" The requested bound is explicit."), ], } ) judges = [] for judge_id in range(1, 6): first = _reject() if judge_id == 1 else _accept() judges.append( ScriptedClient( {f"{item.item_id}.verify": [first, _accept(), _accept()]} ) ) run_root = tmp_path / "run" result = asyncio.run( PaperKernelPipeline( proposer=proposer, judges=judges, store=RunStore(run_root, item.item_id), config=PaperPipelineConfig( proposer_model="scripted", judge_model="scripted", ), ).run(item) ) assert result.status == "accepted" assert [row.pass_streak_after for row in result.iterations] == [0, 1, 2] assert result.iterations[0].bundle_sha256 != result.iterations[1].bundle_sha256 assert result.iterations[1].bundle_sha256 == result.iterations[2].bundle_sha256 calls_dir = run_root / "items" / item.item_id / "calls" repair_stage_calls = [ json.loads( ( calls_dir / f"{item.item_id}.stage{stage}.{name}.v02.json" ).read_text() ) for stage, name in [ (3, "replacement"), (4, "diffusion"), (5, "render"), ] ] assert "PREVIOUS REPLACEMENT PLAN" in repair_stage_calls[0]["user_prompt"] assert "PREVIOUS DIFFUSED PROOF" in repair_stage_calls[1]["user_prompt"] assert "PREVIOUS RENDERED VARIANT" in repair_stage_calls[2]["user_prompt"] assert all( "the terminal wording is ambiguous" in call["user_prompt"] for call in repair_stage_calls ) judge_calls = sorted(calls_dir.glob(f"{item.item_id}.verify.*.json")) assert len(judge_calls) == 15 assert not any(".a1." in path.name for path in judge_calls) first_judge_call = json.loads(judge_calls[0].read_text()) assert first_judge_call["system_prompt"] == JUDGE_SYSTEM_PROMPT assert "METHOD-LABEL SEQUENCE (abstract plan):" in first_judge_call["user_prompt"] assert "SOURCE PROOF DAG:" not in first_judge_call["user_prompt"] def test_stage4_retries_identical_prompt_when_required_field_is_missing( tmp_path, item, ) -> None: invalid = _diffused() for node in invalid["nodes"]: node.pop("justification") request_prefix = f"{item.item_id}.stage4.diffusion.v01" proposer = ScriptedClient({request_prefix: [invalid, _diffused()]}) pipeline = PaperKernelPipeline( proposer=proposer, judges=[ScriptedClient({}) for _ in range(5)], store=RunStore(tmp_path / "run", item.item_id), config=PaperPipelineConfig( proposer_model="scripted", judge_model="scripted", ), ) dag = ProofDAG.model_validate(_dag()) methods = MethodPlan.model_validate(_methods()) replacements = ReplacementPlan.model_validate(_replacement()) result = asyncio.run( pipeline.diffuse_dag( item, dag, methods, replacements, version=1, ) ) assert all(node.justification for node in result.nodes) assert proposer.calls == [ request_prefix, f"{request_prefix}.validation_retry01", ] calls_dir = tmp_path / "run" / "items" / item.item_id / "calls" first_call = json.loads((calls_dir / f"{request_prefix}.json").read_text()) retry_call = json.loads( ( calls_dir / f"{request_prefix}.validation_retry01.json" ).read_text() ) expected_prompt = diffusion_user(item, dag, methods, replacements) assert first_call["user_prompt"] == expected_prompt assert retry_call["user_prompt"] == expected_prompt stage = json.loads( ( tmp_path / "run" / "items" / item.item_id / "stages" / "04_diffused_proof_v01.json" ).read_text() ) assert stage["request_id"] == f"{request_prefix}.validation_retry01"