SILVER
/
NODEDC_1C
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Уточнить агентный pack-loop для ограниченных срезов

This commit is contained in:
dctouch 2026-06-03 07:38:22 +03:00
parent 477e550b0c
commit 4d32b7fad5
2 changed files with 119 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

60
scripts/domain_case_loop.py
View File

@ -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,

60
tests/test_domain_case_loop.py
View File

@ -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",