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

Усилить агентные проверки selected-object сценариев

This commit is contained in:
dctouch 2026-06-02 00:23:47 +03:00
parent 908d011743
commit c17495d1bb
7 changed files with 432 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
scripts/agent_detector_runner.py
View File

@ -166,6 +166,60 @@ def read_text_or_empty(path: Path) -> str:
return ""
def assistant_text_from_turn_path(path: Path | None) -> str:
if path is None or not path.exists():
return ""
payload = read_json_object(path)
assistant_message = payload.get("assistant_message") if isinstance(payload.get("assistant_message"), dict) else {}
text = str(assistant_message.get("text") or "").strip()
return text
def current_answer_text_for_output(output: dict[str, Any]) -> str:
turn_path = output.get("turn_path")
if isinstance(turn_path, Path):
turn_answer = assistant_text_from_turn_path(turn_path)
if turn_answer:
return turn_answer
return str(output.get("text") or "")
CURRENT_TRACE_FIELD_NAMES = {
"detected_mode",
"query_shape",
"detected_intent",
"selected_recipe",
"requested_result_mode",
"result_mode",
"response_type",
"capability_id",
"capability_layer",
"capability_route_mode",
"capability_route_reason",
"execution_lane",
"fallback_type",
"match_failure_stage",
"match_failure_reason",
}
def current_trace_text_from_turn_path(path: Path) -> str:
payload = read_json_object(path)
debug = payload.get("technical_debug_payload") if isinstance(payload.get("technical_debug_payload"), dict) else {}
scoped: dict[str, Any] = {}
for field_name in CURRENT_TRACE_FIELD_NAMES:
if field_name in debug:
scoped[field_name] = debug.get(field_name)
for prefix in ("route_expectation_", "capability_route_"):
for field_name, value in debug.items():
if isinstance(field_name, str) and field_name.startswith(prefix):
scoped[field_name] = value
assistant_message = payload.get("assistant_message") if isinstance(payload.get("assistant_message"), dict) else {}
if assistant_message.get("reply_type") is not None:
scoped["assistant_reply_type"] = assistant_message.get("reply_type")
return json.dumps(scoped, ensure_ascii=False, sort_keys=True)
def output_turn_path(output_path: Path) -> Path | None:
name = output_path.name
if name == "output.md":
@ -305,7 +359,7 @@ def evaluate_forbidden_regex(
return build_result(detector_name, detector, "skipped", "no output.md-style artifacts matched detector scope")
evidence: list[dict[str, Any]] = []
for output in outputs:
text = str(output.get("text") or "")
text = current_answer_text_for_output(output)
if prefix_line_count is not None:
lines = [line for line in text.splitlines() if line.strip()]
text = "\n".join(lines[:prefix_line_count])
@ -336,7 +390,7 @@ def evaluate_required_any(
missing: list[dict[str, Any]] = []
matched: list[dict[str, Any]] = []
for output in outputs:
text = str(output.get("text") or "")
text = current_answer_text_for_output(output)
match = next((pattern.search(text) for pattern in patterns if pattern.search(text)), None)
if match:
matched.append({"path": output["repo_path"], "match": match.group(0)[:240]})
@ -360,7 +414,7 @@ def evaluate_limited_next_action(
failures: list[dict[str, Any]] = []
limited_outputs = 0
for output in outputs:
text = str(output.get("text") or "")
text = current_answer_text_for_output(output)
limited_match = next((pattern.search(text) for pattern in limited_patterns if pattern.search(text)), None)
if not limited_match:
continue
@ -388,7 +442,11 @@ def evaluate_trace_guard(
return build_result(detector_name, detector, "skipped", "no turn.json-style artifacts matched detector scope")
evidence: list[dict[str, Any]] = []
for turn in turns:
text = str(turn.get("text") or "").lower()
path = turn.get("path")
if isinstance(path, Path):
text = current_trace_text_from_turn_path(path).lower()
else:
text = str(turn.get("text") or "").lower()
for marker in markers:
if marker in text:
evidence.append({"path": turn["repo_path"], "marker": marker})

36
scripts/domain_case_loop.py
View File

@ -288,6 +288,9 @@ BUSINESS_EXPECTED_RESULT_MODES = {
"evidence_or_honest_boundary",
"ranking_or_limited_accounting_answer",
"same_inventory_margin_context_or_clarification",
"selected_object_purchase_provenance_or_honest_boundary",
"selected_object_purchase_date_or_honest_boundary",
"selected_object_purchase_documents_or_honest_boundary",
}
MCP_DISCOVERY_CHAIN_INTENT_ALIASES: dict[str, tuple[str, ...]] = {
@ -2607,6 +2610,20 @@ def business_expected_result_mode_matches(expected_result_mode: str, step_state:
detected_intent == "inventory_margin_ranking_for_nomenclature"
or capability_id == "inventory_inventory_margin_ranking_for_nomenclature"
)
selected_object_expected_routes = {
"selected_object_purchase_provenance_or_honest_boundary": (
"inventory_purchase_provenance_for_item",
"inventory_inventory_purchase_provenance_for_item",
),
"selected_object_purchase_date_or_honest_boundary": (
"inventory_purchase_provenance_for_item",
"inventory_inventory_purchase_provenance_for_item",
),
"selected_object_purchase_documents_or_honest_boundary": (
"inventory_purchase_documents_for_item",
"inventory_inventory_purchase_documents_for_item",
),
}
if expected_result_mode == "clarification_required":
return (
@ -2668,6 +2685,25 @@ def business_expected_result_mode_matches(expected_result_mode: str, step_state:
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
if expected_result_mode in selected_object_expected_routes:
expected_intent, expected_capability = selected_object_expected_routes[expected_result_mode]
route_matches = detected_intent == expected_intent or capability_id == expected_capability
factual_or_honest_boundary = (
truth_mode == "confirmed"
or answer_shape == "confirmed_factual"
or step_state.get("balance_confirmed") is True
or truth_mode in GUARDED_INSUFFICIENCY_TRUTH_MODES
or answer_shape in GUARDED_INSUFFICIENCY_ANSWER_SHAPES
or step_state.get("balance_confirmed") is False
)
return (
clean_business_review
and route_matches
and factual_or_honest_boundary
and bool(assistant_text)
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
return False

36
scripts/domain_truth_harness.py
View File

@ -428,6 +428,9 @@ BUSINESS_EXPECTED_RESULT_MODES = {
"evidence_or_honest_boundary",
"ranking_or_limited_accounting_answer",
"same_inventory_margin_context_or_clarification",
"selected_object_purchase_provenance_or_honest_boundary",
"selected_object_purchase_date_or_honest_boundary",
"selected_object_purchase_documents_or_honest_boundary",
}
@ -451,6 +454,20 @@ def business_expected_result_mode_matches(expected_result_mode: str, step_state:
detected_intent == "inventory_margin_ranking_for_nomenclature"
or capability_id == "inventory_inventory_margin_ranking_for_nomenclature"
)
selected_object_expected_routes = {
"selected_object_purchase_provenance_or_honest_boundary": (
"inventory_purchase_provenance_for_item",
"inventory_inventory_purchase_provenance_for_item",
),
"selected_object_purchase_date_or_honest_boundary": (
"inventory_purchase_provenance_for_item",
"inventory_inventory_purchase_provenance_for_item",
),
"selected_object_purchase_documents_or_honest_boundary": (
"inventory_purchase_documents_for_item",
"inventory_inventory_purchase_documents_for_item",
),
}
if expected_result_mode == "clarification_required":
return (
@ -497,6 +514,25 @@ def business_expected_result_mode_matches(expected_result_mode: str, step_state:
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
if expected_result_mode in selected_object_expected_routes:
expected_intent, expected_capability = selected_object_expected_routes[expected_result_mode]
route_matches = detected_intent == expected_intent or capability_id == expected_capability
factual_or_honest_boundary = (
truth_mode == "confirmed"
or answer_shape == "confirmed_factual"
or step_state.get("balance_confirmed") is True
or truth_mode in dcl.GUARDED_INSUFFICIENCY_TRUTH_MODES
or answer_shape in dcl.GUARDED_INSUFFICIENCY_ANSWER_SHAPES
or step_state.get("balance_confirmed") is False
)
return (
clean_business_review
and route_matches
and factual_or_honest_boundary
and bool(assistant_text)
and reply_type in {"partial_coverage", "factual", "factual_with_explanation"}
)
return False

47
scripts/save_agent_semantic_run.py
View File

@ -262,10 +262,57 @@ def validate_domain_pack_loop_dir(loop_dir: Path) -> dict[str, Any]:
}
def validate_domain_case_loop_pack_dir(run_dir: Path) -> dict[str, Any]:
run_dir = run_dir.resolve()
effective_runtime = require_effective_runtime_manifest(run_dir)
pack_state = load_json_object(run_dir / "pack_state.json", "Validated domain-case-loop pack_state.json")
repair_targets = load_json_object(run_dir / "repair_targets.json", "Validated domain-case-loop repair_targets.json")
severity_counts = repair_targets.get("severity_counts") if isinstance(repair_targets.get("severity_counts"), dict) else {}
problems: list[str] = []
assert_status(pack_state.get("execution_status"), "exact", "pack_state.execution_status", problems)
assert_status(pack_state.get("final_status"), "accepted", "pack_state.final_status", problems)
assert_status(pack_state.get("acceptance_status"), "accepted", "pack_state.acceptance_status", problems)
assert_status(repair_targets.get("final_status"), "accepted", "repair_targets.final_status", problems)
assert_status(repair_targets.get("acceptance_status"), "accepted", "repair_targets.acceptance_status", problems)
if int(repair_targets.get("target_count") or 0) != 0:
problems.append(f"repair_targets.target_count={repair_targets.get('target_count')}")
if int(severity_counts.get("P0") or 0) != 0 or int(severity_counts.get("P1") or 0) != 0:
problems.append(
f"repair_targets.severity_counts=P0:{severity_counts.get('P0') or 0},P1:{severity_counts.get('P1') or 0}"
)
if problems:
raise RuntimeError(
"Refusing to save AGENT autorun because the validated domain-case-loop pack is not clean: "
+ ", ".join(problems)
)
scenario_results = pack_state.get("scenario_results") if isinstance(pack_state.get("scenario_results"), list) else []
return {
"schema_version": VALIDATED_AGENT_SAVE_SCHEMA_VERSION,
"validation_status": "accepted_domain_case_loop_pack",
"validated_run_dir": repo_relative(run_dir),
"final_status": pack_state.get("final_status"),
"acceptance_status": pack_state.get("acceptance_status"),
"execution_status": pack_state.get("execution_status"),
"pack_id": pack_state.get("pack_id"),
"scenario_count": len(scenario_results),
"repair_target_count": repair_targets.get("target_count"),
"repair_target_severity_counts": repair_targets.get("severity_counts"),
"effective_runtime": build_effective_runtime_save_summary(effective_runtime, run_dir),
"saved_after_validated_replay": True,
}
def validate_accepted_run_dir(run_dir: Path) -> dict[str, Any]:
run_dir = run_dir.resolve()
if (run_dir / "loop_state.json").exists():
return validate_domain_pack_loop_dir(run_dir)
if (run_dir / "truth_review.json").exists():
return validate_truth_harness_run_dir(run_dir)
if (run_dir / "pack_state.json").exists() and (run_dir / "repair_targets.json").exists():
return validate_domain_case_loop_pack_dir(run_dir)
return validate_truth_harness_run_dir(run_dir)

146
scripts/test_agent_detector_runner.py
View File

@ -221,6 +221,152 @@ class AgentDetectorRunnerTests(unittest.TestCase):
self.assertEqual(statuses["forbidden_margin_terms"], "fail")
self.assertEqual(statuses["margin_domain_leak_accounting_route"], "fail")
def test_forbidden_regex_uses_current_assistant_answer_from_turn_json(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "run"
step_dir = artifact_dir / "steps" / "s01"
write_text(
step_dir / "output.md",
"# Assistant conversation export\n\n"
"## 1. user\n"
"Can you compute margin from payment_document evidence?\n\n"
"## 2. assistant\n"
"Clean current answer.\n",
)
write_json(
step_dir / "turn.json",
{
"user_message": {"text": "Can you compute margin from payment_document evidence?"},
"assistant_message": {"text": "Clean current answer.", "reply_type": "factual"},
"technical_debug_payload": {"detected_intent": "inventory_margin_ranking_for_nomenclature"},
},
)
registry_path = root / "detector_registry.json"
issue_catalog_path = root / "issue_catalog.json"
write_json(
registry_path,
{
"schema_version": "agent_detector_registry_v1",
"detectors": {
"forbidden_margin_terms": {
"kind": "answer_text_regex_forbidden",
"automation_level": "automatic",
"description": "No wrong-domain words in the current answer.",
"issue_codes": ["margin_domain_leak_accounting_route"],
"inputs": ["output.md"],
"check": {"forbidden_patterns": ["payment_document"]},
}
},
},
)
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
results = runner.build_detector_results(
artifact_dir,
detector_names=["forbidden_margin_terms"],
registry_path=registry_path,
issue_catalog_path=issue_catalog_path,
include_default_global=False,
)
self.assertEqual(results["summary"]["status"], "pass")
self.assertEqual(results["results"][0]["status"], "pass")
def test_trace_guard_uses_current_route_fields_not_full_turn_history(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "run"
step_dir = artifact_dir / "steps" / "s01"
write_json(
step_dir / "turn.json",
{
"user_message": {"text": "Ignore fixed_asset and answer selected item document."},
"assistant_message": {"text": "Inventory document answer.", "reply_type": "factual"},
"technical_debug_payload": {
"detected_intent": "inventory_purchase_documents_for_item",
"selected_recipe": "address_inventory_purchase_documents_for_item_v1",
"capability_id": "inventory_inventory_purchase_documents_for_item",
},
},
)
registry_path = root / "detector_registry.json"
issue_catalog_path = root / "issue_catalog.json"
write_json(
registry_path,
{
"schema_version": "agent_detector_registry_v1",
"detectors": {
"wrong_capability_family": {
"kind": "trace_value_guard",
"automation_level": "automatic",
"description": "No wrong current route family.",
"issue_codes": ["margin_domain_leak_accounting_route"],
"inputs": ["turn.json"],
"check": {"forbidden_trace_markers": ["fixed_asset"]},
}
},
},
)
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
results = runner.build_detector_results(
artifact_dir,
detector_names=["wrong_capability_family"],
registry_path=registry_path,
issue_catalog_path=issue_catalog_path,
include_default_global=False,
)
self.assertEqual(results["summary"]["status"], "pass")
self.assertEqual(results["results"][0]["status"], "pass")
def test_trace_guard_fails_when_current_route_field_has_forbidden_marker(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "run"
step_dir = artifact_dir / "steps" / "s01"
write_json(
step_dir / "turn.json",
{
"assistant_message": {"text": "Wrong route answer.", "reply_type": "factual"},
"technical_debug_payload": {
"detected_intent": "fixed_asset_amortization",
"selected_recipe": "address_fixed_asset_amortization_v1",
},
},
)
registry_path = root / "detector_registry.json"
issue_catalog_path = root / "issue_catalog.json"
write_json(
registry_path,
{
"schema_version": "agent_detector_registry_v1",
"detectors": {
"wrong_capability_family": {
"kind": "trace_value_guard",
"automation_level": "automatic",
"description": "No wrong current route family.",
"issue_codes": ["margin_domain_leak_accounting_route"],
"inputs": ["turn.json"],
"check": {"forbidden_trace_markers": ["fixed_asset"]},
}
},
},
)
write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
results = runner.build_detector_results(
artifact_dir,
detector_names=["wrong_capability_family"],
registry_path=registry_path,
issue_catalog_path=issue_catalog_path,
include_default_global=False,
)
self.assertEqual(results["summary"]["status"], "fail")
self.assertEqual(results["results"][0]["status"], "fail")
def test_stage_review_signal_detector_fails_when_issue_code_is_present(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)

49
scripts/test_domain_case_loop_step_state.py
View File

@ -715,6 +715,55 @@ class DomainCaseLoopStepStateTests(unittest.TestCase):
self.assertIn("wrong_result_mode", step_state["violated_invariants"])
def test_selected_object_purchase_document_mode_accepts_confirmed_factual_route(self) -> None:
step_state = dcl.build_scenario_step_state(
scenario_id="margin_selected_object_demo",
domain="margin_profitability",
step={
"step_id": "step_05_selected_item_documents",
"title": "Selected item document",
"depends_on": ["step_04_selected_item_purchase_date"],
"question_template": "Which document confirms this selected item?",
"expected_intents": ["inventory_purchase_documents_for_item"],
"expected_capability": "inventory_inventory_purchase_documents_for_item",
"expected_recipe": "address_inventory_purchase_documents_for_item_v1",
"expected_result_mode": "selected_object_purchase_documents_or_honest_boundary",
"required_answer_shape": "direct_answer_first",
},
step_index=5,
question_resolved="Which document confirms this selected item?",
analysis_context={},
turn_artifact={
"assistant_message": {
"reply_type": "factual",
"text": "Confirmation document for the selected item: Purchase document 0001.",
"message_id": "msg-1",
"trace_id": "trace-1",
},
"technical_debug_payload": {
"detected_mode": "address_query",
"detected_intent": "inventory_purchase_documents_for_item",
"selected_recipe": "address_inventory_purchase_documents_for_item_v1",
"capability_id": "inventory_inventory_purchase_documents_for_item",
"capability_route_mode": "exact",
"fallback_type": "none",
"mcp_call_status": "matched_non_empty",
"response_type": "FACTUAL_SUMMARY",
"result_mode": "confirmed_balance",
"truth_mode": "confirmed",
"answer_shape": "confirmed_factual",
"balance_confirmed": True,
"extracted_filters": {"item": "Selected item"},
},
"session_summary": {},
},
entries=[],
)
self.assertEqual(step_state["execution_status"], "exact")
self.assertNotIn("wrong_result_mode", step_state["violated_invariants"])
self.assertEqual(step_state["acceptance_status"], "validated")
def test_exact_confirmed_document_followup_sets_runtime_factual_validation(self) -> None:
step_state = dcl.build_scenario_step_state(
scenario_id="svk_pivot",

56
scripts/test_save_agent_semantic_run.py
View File

@ -79,6 +79,62 @@ class SaveAgentSemanticRunTests(unittest.TestCase):
self.assertEqual(metadata["effective_runtime"]["runner"], "domain_truth_harness.run-live")
self.assertEqual(metadata["effective_runtime"]["llm_model"], "test-model")
def write_domain_case_loop_pack(self, run_dir: Path, *, target_count: int = 0, p1_count: int = 0) -> None:
write_json(
run_dir / "pack_state.json",
{
"pack_id": "agent_pack",
"execution_status": "exact",
"acceptance_status": "accepted",
"final_status": "accepted",
"scenario_results": [{"scenario_id": "s01", "final_status": "accepted"}],
},
)
write_json(
run_dir / "repair_targets.json",
{
"target_count": target_count,
"acceptance_status": "accepted",
"final_status": "accepted",
"severity_counts": {"P0": 0, "P1": p1_count, "P2": 0},
},
)
write_json(
run_dir / runtime_manifest.EFFECTIVE_RUNTIME_FILE_NAME,
{
"schema_version": runtime_manifest.EFFECTIVE_RUNTIME_SCHEMA_VERSION,
"runner": "domain_case_loop.run-pack",
"git_sha": "test-sha",
"llm_provider": "local",
"llm_model": "test-model",
"temperature": 0.0,
"max_output_tokens": 2048,
"prompt_version": "normalizer_v2_0_2",
"prompt_source": "file",
"prompt_hash": "abc123",
"prompt_registry_status": "pass",
},
)
def test_validate_domain_case_loop_pack_accepts_clean_pack(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
run_dir = Path(tmp)
self.write_domain_case_loop_pack(run_dir)
metadata = saver.validate_accepted_run_dir(run_dir)
self.assertEqual(metadata["validation_status"], "accepted_domain_case_loop_pack")
self.assertEqual(metadata["effective_runtime"]["runner"], "domain_case_loop.run-pack")
self.assertEqual(metadata["repair_target_count"], 0)
def test_validate_domain_case_loop_pack_refuses_open_repair_targets(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
run_dir = Path(tmp)
self.write_domain_case_loop_pack(run_dir, target_count=1, p1_count=1)
with self.assertRaisesRegex(RuntimeError, "not clean"):
saver.validate_accepted_run_dir(run_dir)
def test_extract_questions_resolves_scenario_pack_bindings(self) -> None:
spec = {
"schema_version": "domain_scenario_pack_v1",