diff --git a/scripts/domain_case_loop.py b/scripts/domain_case_loop.py index 6b38cfe..5357205 100644 --- a/scripts/domain_case_loop.py +++ b/scripts/domain_case_loop.py @@ -4894,6 +4894,12 @@ def build_deterministic_repair_targets( "execution_status": pack_state.get("execution_status"), "acceptance_status": pack_state.get("acceptance_status"), "final_status": pack_state.get("final_status"), + "coverage_mode": pack_state.get("coverage_mode"), + "max_scenarios": pack_state.get("max_scenarios"), + "total_scenario_count": pack_state.get("total_scenario_count"), + "selected_scenario_count": pack_state.get("selected_scenario_count"), + "selected_scenario_ids": pack_state.get("selected_scenario_ids") or [], + "unexecuted_scenario_ids": pack_state.get("unexecuted_scenario_ids") or [], "target_count": len(targets), "severity_counts": severity_counts, "priority_foci": priority_foci, @@ -5206,6 +5212,22 @@ def analyst_target_contradicts_pack_acceptance_status( ) +def analyst_target_outside_limited_slice( + repair_targets: dict[str, Any], + target: dict[str, Any], +) -> bool: + coverage_mode = str(repair_targets.get("coverage_mode") or "").strip() + if coverage_mode != "limited_slice": + return False + unexecuted_scenario_ids = { + str(item or "").strip() + for item in normalize_string_list(repair_targets.get("unexecuted_scenario_ids")) + if str(item or "").strip() + } + scenario_id = str(target.get("scenario_id") or "").strip() + return bool(scenario_id and scenario_id in unexecuted_scenario_ids) + + def merge_analyst_priority_repair_targets( repair_targets: dict[str, Any], analyst_verdict: dict[str, Any], @@ -5239,6 +5261,19 @@ def merge_analyst_priority_repair_targets( analyst_count += 1 target_id = str(analyst_target.get("target_id") or "").strip() step_snapshot = step_validation_index.get(target_id) if isinstance(step_validation_index, dict) else None + if analyst_target_outside_limited_slice(repair_targets, analyst_target): + suppressed_analyst_targets.append( + { + "target_id": target_id, + "reason": "analyst_target_outside_limited_slice", + "coverage": { + "coverage_mode": repair_targets.get("coverage_mode"), + "selected_scenario_ids": repair_targets.get("selected_scenario_ids") or [], + "unexecuted_scenario_ids": repair_targets.get("unexecuted_scenario_ids") or [], + }, + } + ) + continue if analyst_target_contradicts_validated_step(analyst_target, step_snapshot): suppressed_analyst_targets.append( { @@ -5733,6 +5768,10 @@ def build_analyst_loop_prompt( Rules: - `accepted` is allowed only if quality_score >= {target_score}, unresolved_p0_count = 0, and regression_detected = false; - `accepted` is forbidden if the evidence bundle shows `pack_state.acceptance_status != accepted` (or, for old artifacts without `acceptance_status`, `pack_state.final_status != accepted`) or the deterministic repair targets still contain any `P0` or `P1` items; + - if `pack_state.coverage_mode = limited_slice`, judge only `pack_state.selected_scenario_ids` and the scenario artifacts that actually exist in the evidence bundle; do not create priority targets, broken edges, or P0/P1 failures for `pack_state.unexecuted_scenario_ids` merely because they were intentionally outside `--max-scenarios`; + - for a limited slice, mention missing scenarios only as a coverage limitation / next replay scope, not as a semantic defect, unless an executed step actually proves a cross-scenario regression; + - for a limited slice, `pack_state.acceptance_status=partial` can be caused solely by the intentional scenario limit; do not lower `direct_answer_ok`, `business_usefulness_ok`, `temporal_honesty_ok`, or `field_truth_ok` because unexecuted scenarios are absent; + - for a limited slice, set `direct_answer_ok`, `business_usefulness_ok`, `temporal_honesty_ok`, and `field_truth_ok` from the executed selected scenarios only; if those answers are semantically clean, these booleans should be true even when `loop_decision` remains `partial` for full-pack coverage; - `pack_state.final_status=partial` with `acceptance_status=accepted` means the run still had partial execution evidence; treat it as a proof-strength signal, not as a presentation bug by itself; - do not create a priority target solely to change `pack_state.final_status=partial` when `pack_state.acceptance_status=accepted`; only create a pack-status target if `acceptance_status` is missing/not accepted or if the user-facing scenario semantics are actually wrong; - `accepted` also requires `direct_answer_ok = true`, `business_usefulness_ok = true`, `temporal_honesty_ok = true`, and `field_truth_ok = true`; @@ -6585,6 +6624,21 @@ def handle_run_pack(args: argparse.Namespace) -> int: scenario_results: list[dict[str, Any]] = [] max_scenarios = max(0, int(args.max_scenarios)) if args.max_scenarios is not None else None scenarios_to_run = pack["scenarios"][:max_scenarios] if max_scenarios else pack["scenarios"] + total_scenario_count = len(pack.get("scenarios") or []) + selected_scenario_ids = [ + str(scenario.get("scenario_id") or "").strip() + for scenario in scenarios_to_run + if isinstance(scenario, dict) and str(scenario.get("scenario_id") or "").strip() + ] + selected_scenario_id_set = set(selected_scenario_ids) + unexecuted_scenario_ids = [ + str(scenario.get("scenario_id") or "").strip() + for scenario in pack.get("scenarios", []) + if isinstance(scenario, dict) + and str(scenario.get("scenario_id") or "").strip() + and str(scenario.get("scenario_id") or "").strip() not in selected_scenario_id_set + ] + coverage_mode = "limited_slice" if len(selected_scenario_ids) < total_scenario_count else "full_pack" for scenario in scenarios_to_run: scenario_manifest = normalize_scenario_manifest( scenario, @@ -6622,6 +6676,12 @@ def handle_run_pack(args: argparse.Namespace) -> int: "title": pack["title"], "analysis_context": pack.get("analysis_context") or {}, "bindings": pack.get("bindings") or {}, + "coverage_mode": coverage_mode, + "max_scenarios": max_scenarios, + "total_scenario_count": total_scenario_count, + "selected_scenario_count": len(selected_scenario_ids), + "selected_scenario_ids": selected_scenario_ids, + "unexecuted_scenario_ids": unexecuted_scenario_ids, "scenario_results": scenario_results, "execution_status": execution_status, "acceptance_status": acceptance_status, diff --git a/tests/test_domain_case_loop.py b/tests/test_domain_case_loop.py index dbd3e9b..58e4a04 100644 --- a/tests/test_domain_case_loop.py +++ b/tests/test_domain_case_loop.py @@ -7,6 +7,7 @@ from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from scripts.domain_case_loop import ( + build_analyst_loop_prompt, build_coder_loop_prompt, build_coder_snapshot_paths, build_deterministic_repair_targets, @@ -18,6 +19,7 @@ from scripts.domain_case_loop import ( evaluate_deterministic_loop_gate, load_scenario_pack, load_shared_local_llm_defaults, + merge_analyst_priority_repair_targets, merge_scenario_date_scope, select_primary_repair_focus, restore_line_collapsed_files_from_snapshot, @@ -840,7 +842,7 @@ def test_evaluate_deterministic_loop_gate_rejects_partial_pack_even_without_targ ) assert gate_ok is False - assert reason == "pack_final_status=partial" + assert reason == "pack_acceptance_status=partial;pack_final_status=partial" def test_evaluate_deterministic_loop_gate_rejects_remaining_p1_targets() -> None: @@ -863,6 +865,62 @@ def test_evaluate_deterministic_loop_gate_accepts_clean_pack_without_remaining_p assert reason == "deterministic_gate_passed" +def test_build_analyst_loop_prompt_treats_limited_slice_as_coverage_scope(tmp_path: Path) -> None: + prompt = build_analyst_loop_prompt( + loop_dir=tmp_path / "loop", + iteration_dir=tmp_path / "loop" / "iterations" / "iteration_00", + pack_dir=tmp_path / "loop" / "iterations" / "iteration_00" / "pack_output" / "pack_run", + repair_targets_path=tmp_path / "loop" / "iterations" / "iteration_00" / "repair_targets.json", + previous_pack_dir=None, + previous_verdict_path=None, + target_score=90, + review_bundle_json=json.dumps( + { + "pack_state": { + "coverage_mode": "limited_slice", + "selected_scenario_ids": ["inventory_snapshot_roots"], + "unexecuted_scenario_ids": ["inventory_selected_item_provenance"], + "acceptance_status": "partial", + } + }, + ensure_ascii=False, + ), + repair_targets_json=json.dumps({"targets": []}), + previous_verdict_json=None, + ) + + assert "pack_state.coverage_mode = limited_slice" in prompt + assert "do not create priority targets" in prompt + assert "pack_state.unexecuted_scenario_ids" in prompt + + +def test_merge_analyst_priority_targets_suppresses_unexecuted_limited_slice_scenarios() -> None: + merged = merge_analyst_priority_repair_targets( + { + "coverage_mode": "limited_slice", + "selected_scenario_ids": ["inventory_snapshot_roots"], + "unexecuted_scenario_ids": ["inventory_selected_item_provenance"], + "targets": [], + "step_validation_index": {}, + }, + { + "priority_targets": [ + { + "scenario_id": "inventory_selected_item_provenance", + "step_id": "step_02_selected_item_supplier_ui", + "severity": "P1", + "problem_type": "loop_coverage_gap", + "fix_goal": "run the selected item scenario", + } + ] + }, + ) + + assert merged["targets"] == [] + assert merged["suppressed_analyst_priority_target_count"] == 1 + assert merged["suppressed_analyst_priority_targets"][0]["reason"] == "analyst_target_outside_limited_slice" + + def test_build_scenario_step_state_uses_effective_analysis_context_from_turn_artifact() -> None: step_state = build_scenario_step_state( scenario_id="inventory_snapshot_roots",