Связать GUI-ревью с issue catalog и detector flywheel

This commit is contained in:
dctouch 2026-05-25 13:09:57 +03:00
parent 8ce007724a
commit f69b3e1994
8 changed files with 562 additions and 2 deletions

View File

@ -201,6 +201,76 @@
"check": {
"target_status": "needs_route_enablement"
}
},
"backend_error_response_signal": {
"kind": "stage_review_signal",
"automation_level": "automatic",
"description": "GUI/stage run review found a backend_error user-facing response.",
"issue_codes": ["backend_error_response"],
"inputs": ["run_review.json", "conversation_pairs.json"],
"check": {
"target_status": "backend_error_response"
}
},
"top_year_direct_answer_missing": {
"kind": "stage_review_signal",
"automation_level": "automatic",
"description": "GUI/stage run review found a top-year ranking answer without direct winning-year surface.",
"issue_codes": ["year_ranking_direct_answer_missing"],
"inputs": ["run_review.json", "conversation_pairs.json"],
"check": {
"target_status": "year_ranking_direct_answer_missing"
}
},
"vat_purchase_date_anchor_missing": {
"kind": "stage_review_signal",
"automation_level": "automatic",
"description": "GUI/stage run review found a VAT purchase-date question whose answer did not explain the purchase-date anchor.",
"issue_codes": ["purchase_date_vat_anchor_ambiguous"],
"inputs": ["run_review.json", "conversation_pairs.json"],
"check": {
"target_status": "purchase_date_vat_anchor_ambiguous"
}
},
"counterparty_value_flow_profit_mismatch": {
"kind": "stage_review_signal",
"automation_level": "automatic",
"description": "GUI/stage run review found a counterparty received/paid/net-flow question answered as company profit.",
"issue_codes": ["counterparty_value_flow_misrouted_to_company_profit"],
"inputs": ["run_review.json", "conversation_pairs.json"],
"check": {
"target_status": "counterparty_value_flow_misrouted_to_company_profit"
}
},
"answer_layering_noise_signal": {
"kind": "stage_review_signal",
"automation_level": "semi_automatic",
"description": "GUI/stage run review found answer layering noise before the business conclusion.",
"issue_codes": ["answer_layering_noise"],
"inputs": ["run_review.json", "conversation_pairs.json"],
"check": {
"target_status": "answer_layering_noise"
}
},
"business_answer_too_verbose_signal": {
"kind": "stage_review_signal",
"automation_level": "semi_automatic",
"description": "GUI/stage run review found a direct business answer that is too verbose for practical use.",
"issue_codes": ["business_answer_too_verbose"],
"inputs": ["run_review.json", "conversation_pairs.json"],
"check": {
"target_status": "business_answer_too_verbose"
}
},
"bank_counterparty_role_boundary_missing": {
"kind": "stage_review_signal",
"automation_level": "semi_automatic",
"description": "GUI/stage run review found a bank/financial counterparty presented as an ordinary business partner without a role boundary.",
"issue_codes": ["bank_counterparty_misclassified_as_business_partner"],
"inputs": ["run_review.json", "conversation_pairs.json"],
"check": {
"target_status": "bank_counterparty_misclassified_as_business_partner"
}
}
}
}

View File

@ -257,6 +257,212 @@
"route_candidate_pack",
"accepted_smoke_pack"
]
},
"backend_error_response": {
"severity": "P0",
"business_meaning": "Пользовательский ответ сорвался во внутреннюю runtime/LLM ошибку вместо полезного бизнес-ответа или честной границы.",
"root_layers": ["runtime_error", "business_utility"],
"detectors": ["backend_error_response_signal"],
"allowed_patch_targets": [
"llm_normalizer/backend/src/services/assistantLivingChatRuntimeAdapter.ts",
"llm_normalizer/backend/src/services/assistantLivingChatHandlerRuntimeAdapter.ts",
"llm_normalizer/backend/src/services/assistantService.ts",
"scripts/review_assistant_stage1_run.py"
],
"forbidden_patch_targets": [
"fake success fallback",
"business routing rewrites",
"MCP protocol rewrite"
],
"rerun_matrix": [
"failed_gui_turn",
"same_saved_session_replay",
"big_gui_pack",
"accepted_smoke_pack"
],
"acceptance": {
"must_have": [
"no backend_error reply_type",
"user-facing answer or honest boundary",
"trace remains diagnosable"
],
"must_not_have": [
"internal error text",
"empty assistant answer",
"masked fake business fact"
]
}
},
"year_ranking_direct_answer_missing": {
"severity": "P0",
"business_meaning": "Вопрос про самый доходный/лучший год не получает прямой годовой ranking-ответ в первой строке.",
"root_layers": ["answer_shape_mismatch", "business_utility", "ranking_answer_surface"],
"detectors": ["top_year_direct_answer_missing"],
"allowed_patch_targets": [
"llm_normalizer/backend/src/services/assistantMcpDiscoveryResponseCandidate.ts",
"llm_normalizer/backend/src/services/assistantMcpDiscoveryRuntimeBridge.ts",
"llm_normalizer/backend/src/services/assistantMcpDiscoveryTurnInputAdapter.ts",
"scripts/review_assistant_stage1_run.py"
],
"forbidden_patch_targets": [
"fake revenue totals",
"global business overview rewrite",
"silent route masking"
],
"rerun_matrix": [
"failed_top_year_turn",
"business_overview_counterparty_pack",
"big_gui_pack",
"accepted_smoke_pack"
],
"acceptance": {
"must_have": [
"direct winning year in first answer line",
"basis of ranking",
"amount or honest limitation",
"profit-vs-cashflow limitation if needed"
],
"must_not_have": [
"generic business overview first",
"route ids",
"unqualified чистая прибыль claim"
]
}
},
"purchase_date_vat_anchor_ambiguous": {
"severity": "P1",
"business_meaning": "Ответ по НДС на дату покупки считает налоговый период, но не объясняет, как дата покупки стала якорем периода.",
"root_layers": ["domain_anchor_gap", "field_mapping_gap", "answer_surface"],
"detectors": ["vat_purchase_date_anchor_missing"],
"allowed_patch_targets": [
"llm_normalizer/backend/src/services/address_runtime/composeStage.ts",
"llm_normalizer/backend/src/services/addressFilterExtractor.ts",
"llm_normalizer/backend/src/services/assistantTransitionPolicy.ts",
"scripts/review_assistant_stage1_run.py"
],
"forbidden_patch_targets": [
"fake purchase date",
"VAT route rewrite without evidence",
"MCP protocol rewrite"
],
"rerun_matrix": [
"failed_vat_purchase_date_turn",
"vat_tax_period_pack",
"selected_object_followup_pack",
"accepted_smoke_pack"
],
"acceptance": {
"must_have": [
"purchase date anchor explanation",
"tax period window",
"sales/purchase book basis",
"honest unknown if exact purchase date is inferred"
],
"must_not_have": [
"unexplained period jump",
"fake document source",
"forecast presented as confirmed declaration"
]
}
},
"counterparty_value_flow_misrouted_to_company_profit": {
"severity": "P0",
"business_meaning": "Контрагентский вопрос про получили/заплатили/нетто отвечен company-level прибылью или 90/91/99 вместо контрагентского денежного потока.",
"root_layers": ["followup_action_resolution_gap", "answer_shape_mismatch", "domain_scope_leak"],
"detectors": ["counterparty_value_flow_profit_mismatch"],
"allowed_patch_targets": [
"llm_normalizer/backend/src/services/assistantMcpDiscoveryResponseCandidate.ts",
"llm_normalizer/backend/src/services/assistantMcpDiscoveryRuntimeBridge.ts",
"llm_normalizer/backend/src/services/assistantMcpDiscoveryTurnInputAdapter.ts",
"llm_normalizer/backend/src/services/assistantTransitionPolicy.ts",
"scripts/review_assistant_stage1_run.py"
],
"forbidden_patch_targets": [
"fake counterparty totals",
"company profit fallback",
"global orchestration rewrite"
],
"rerun_matrix": [
"failed_counterparty_value_flow_turn",
"business_overview_counterparty_pack",
"cross_scope_followup_pack",
"accepted_smoke_pack"
],
"acceptance": {
"must_have": [
"counterparty scope",
"received amount or honest unknown",
"paid amount or honest unknown",
"net amount or honest unknown",
"company-profit boundary"
],
"must_not_have": [
"90/91/99 as direct counterparty answer",
"company profit as counterparty net",
"wrong selected entity"
]
}
},
"answer_layering_noise": {
"severity": "P1",
"business_meaning": "Ответ открывается лишним слоем рассуждений/рамок вместо короткого прямого вывода.",
"root_layers": ["answer_surface", "business_utility"],
"detectors": ["answer_layering_noise_signal"],
"allowed_patch_targets": [
"llm_normalizer/backend/src/services/address_runtime/composeStage.ts",
"llm_normalizer/backend/src/services/assistantMcpDiscoveryResponseCandidate.ts",
"scripts/review_assistant_stage1_run.py"
],
"forbidden_patch_targets": [
"fake evidence",
"route masking"
],
"rerun_matrix": [
"failed_gui_turn",
"direct_answer_surface_pack",
"accepted_smoke_pack"
]
},
"business_answer_too_verbose": {
"severity": "P1",
"business_meaning": "Ответ слишком длинный для прямого бизнес-вопроса и снижает практическую полезность.",
"root_layers": ["answer_surface", "business_utility"],
"detectors": ["business_answer_too_verbose_signal"],
"allowed_patch_targets": [
"llm_normalizer/backend/src/services/address_runtime/composeStage.ts",
"llm_normalizer/backend/src/services/assistantMcpDiscoveryResponseCandidate.ts",
"scripts/review_assistant_stage1_run.py"
],
"forbidden_patch_targets": [
"evidence deletion from artifacts",
"business fact removal"
],
"rerun_matrix": [
"failed_gui_turn",
"direct_answer_surface_pack",
"accepted_smoke_pack"
]
},
"bank_counterparty_misclassified_as_business_partner": {
"severity": "P1",
"business_meaning": "Банк/финансовая организация ошибочно подана как обычный клиент/поставщик без роли, назначения платежа и договорного контекста.",
"root_layers": ["business_semantic_role_gap", "domain_scope"],
"detectors": ["bank_counterparty_role_boundary_missing"],
"allowed_patch_targets": [
"llm_normalizer/backend/src/services/address_runtime/composeStage.ts",
"llm_normalizer/backend/src/services/assistantMcpDiscoveryResponseCandidate.ts",
"scripts/review_assistant_stage1_run.py"
],
"forbidden_patch_targets": [
"hard-coded bank blacklist as final truth",
"fake operation purpose",
"global routing rewrite"
],
"rerun_matrix": [
"failed_bank_counterparty_turn",
"counterparty_value_pack",
"accepted_smoke_pack"
]
}
}
}

View File

@ -433,14 +433,55 @@ def evaluate_stage_review_signal(
for path in sorted(artifact_dir.rglob("*.json")):
if path.name not in {"run_review.json", "repair_targets.json", "semantic_repair_targets.json"}:
continue
text = read_text_or_empty(path)
if target_status in text:
payload = read_json_object(path)
if stage_review_payload_has_signal(payload, target_status):
evidence.append({"path": repo_relative(path), "target_status": target_status})
status = "fail" if evidence else "pass"
message = "stage review signal found" if evidence else "stage review signal not found"
return build_result(detector_name, detector, status, message, evidence=evidence)
def record_has_stage_review_signal(record: dict[str, Any], target_status: str) -> bool:
if str(record.get("issue_code") or "").strip() == target_status:
return True
if target_status in normalize_string_list(record.get("issue_codes")):
return True
for field_name in ("target_status", "status", "repair_status"):
if str(record.get(field_name) or "").strip() == target_status:
return True
return False
def records_have_stage_review_signal(records: Any, target_status: str) -> bool:
if not isinstance(records, list):
return False
return any(isinstance(item, dict) and record_has_stage_review_signal(item, target_status) for item in records)
def stage_review_payload_has_signal(payload: Any, target_status: str) -> bool:
if isinstance(payload, list):
return records_have_stage_review_signal(payload, target_status)
if not isinstance(payload, dict):
return False
summary = payload.get("summary") if isinstance(payload.get("summary"), dict) else {}
issue_counts = summary.get("issue_counts") if isinstance(summary.get("issue_counts"), dict) else {}
try:
issue_count = int(issue_counts.get(target_status) or 0)
except (TypeError, ValueError):
issue_count = 0
if issue_count > 0:
return True
if records_have_stage_review_signal(payload.get("findings"), target_status):
return True
if records_have_stage_review_signal(payload.get("repair_targets"), target_status):
return True
if records_have_stage_review_signal(payload.get("primary_repair_targets"), target_status):
return True
if records_have_stage_review_signal(payload.get("targets"), target_status):
return True
return record_has_stage_review_signal(payload, target_status)
def child_status(detector_name: str, results_by_name: dict[str, dict[str, Any]]) -> str:
child = results_by_name.get(detector_name)
return str(child.get("status") if child else "skipped")

View File

@ -31,6 +31,15 @@ AUTO_CODER_ALLOWED_ISSUE_CODES = {
"business_next_step_missing",
"technical_garbage_in_answer",
}
GUI_REVIEW_REQUIRED_ISSUE_CODES = {
"answer_layering_noise",
"backend_error_response",
"bank_counterparty_misclassified_as_business_partner",
"business_answer_too_verbose",
"counterparty_value_flow_misrouted_to_company_profit",
"purchase_date_vat_anchor_ambiguous",
"year_ranking_direct_answer_missing",
}
REQUIRED_ISSUE_FIELDS = {
"severity",
"business_meaning",
@ -269,6 +278,7 @@ def check_issue_catalog(path: Path) -> tuple[dict[str, Any], list[str], list[str
"schema_version": payload.get("schema_version"),
"issue_count": len(issues),
"auto_coder_allowed_issue_codes": sorted(AUTO_CODER_ALLOWED_ISSUE_CODES),
"gui_review_required_issue_codes": sorted(GUI_REVIEW_REQUIRED_ISSUE_CODES),
}
if payload.get("schema_version") != "agent_issue_catalog_v1":
failures.append("issue_catalog_schema_version_mismatch")
@ -301,6 +311,9 @@ def check_issue_catalog(path: Path) -> tuple[dict[str, Any], list[str], list[str
missing_allowed = sorted(AUTO_CODER_ALLOWED_ISSUE_CODES.difference(issues.keys()))
for issue_code in missing_allowed:
failures.append(f"auto_coder_allowlisted_issue_missing_from_catalog:{issue_code}")
missing_gui_review = sorted(GUI_REVIEW_REQUIRED_ISSUE_CODES.difference(issues.keys()))
for issue_code in missing_gui_review:
failures.append(f"gui_review_issue_missing_from_catalog:{issue_code}")
return summary, failures, warnings

View File

@ -17,8 +17,10 @@ REPO_ROOT = Path(__file__).resolve().parents[1]
DEFAULT_SESSIONS_DIR = REPO_ROOT / "llm_normalizer" / "data" / "assistant_sessions"
DEFAULT_REPORTS_DIR = REPO_ROOT / "llm_normalizer" / "reports"
DEFAULT_OUTPUT_ROOT = REPO_ROOT / "artifacts" / "domain_runs" / "gui_run_reviews"
ISSUE_CATALOG_PATH = REPO_ROOT / "docs" / "orchestration" / "issue_catalog.json"
RUN_REVIEW_SCHEMA_VERSION = "assistant_stage1_run_review_v1"
QUESTION_QUALITY_SCHEMA_VERSION = "assistant_stage1_question_quality_v1"
DETECTOR_CANDIDATES_SCHEMA_VERSION = "agent_detector_candidates_v1"
DOMAIN_MARKERS: dict[str, tuple[str, ...]] = {
"vat": ("ндс", "налог", "вычет", "счет-фактур"),
@ -677,6 +679,69 @@ def build_repair_targets(findings: list[dict[str, Any]]) -> list[dict[str, Any]]
)
def load_issue_catalog_detector_map(path: Path = ISSUE_CATALOG_PATH) -> dict[str, list[str]]:
try:
payload = load_json(path)
except (OSError, json.JSONDecodeError):
return {}
issues = payload.get("issues") if isinstance(payload, dict) and isinstance(payload.get("issues"), dict) else {}
result: dict[str, list[str]] = {}
for issue_code, issue in issues.items():
if not isinstance(issue, dict):
continue
detectors = dcl.normalize_string_list(issue.get("detectors"))
if detectors:
result[str(issue_code)] = detectors
return result
def build_detector_candidates(
review: dict[str, Any],
*,
issue_catalog_path: Path = ISSUE_CATALOG_PATH,
) -> dict[str, Any]:
detector_map = load_issue_catalog_detector_map(issue_catalog_path)
candidates: list[dict[str, Any]] = []
seen: set[tuple[str, str, str, str]] = set()
findings = review.get("findings") if isinstance(review.get("findings"), list) else []
for finding in findings:
if not isinstance(finding, dict):
continue
step_id = str(finding.get("step_id") or "").strip()
turn_index = str(finding.get("turn_index") or "").strip()
evidence_paths = ["run_review.json", "conversation_pairs.json"]
for issue_code in dcl.normalize_string_list(finding.get("issue_codes")):
detectors = detector_map.get(issue_code) or [f"{issue_code}_signal"]
for detector in detectors:
key = (issue_code, detector, step_id, turn_index)
if key in seen:
continue
seen.add(key)
candidates.append(
{
"issue_code": issue_code,
"detector": detector,
"severity": (
finding.get("issue_severities", {}).get(issue_code)
if isinstance(finding.get("issue_severities"), dict)
else finding.get("severity")
),
"step_id": step_id or None,
"turn_index": finding.get("turn_index"),
"question": finding.get("question"),
"evidence_paths": evidence_paths,
}
)
return {
"schema_version": DETECTOR_CANDIDATES_SCHEMA_VERSION,
"created_at": now_iso(),
"source_run_id": review.get("run_id"),
"issue_catalog_path": repo_relative(issue_catalog_path),
"candidate_count": len(candidates),
"candidates": candidates,
}
def build_run_review(
*,
run_id: str,
@ -845,6 +910,7 @@ def save_run_review(review: dict[str, Any], output_dir: Path) -> None:
write_json(output_dir / "conversation_pairs.json", review.get("conversation_pairs") or [])
write_json(output_dir / "question_quality_review.json", review.get("question_quality_review") or {})
write_json(output_dir / "repair_targets.json", review.get("repair_targets") or [])
write_json(output_dir / "detector_candidates.json", build_detector_candidates(review))
def build_parser() -> argparse.ArgumentParser:

View File

@ -221,6 +221,121 @@ class AgentDetectorRunnerTests(unittest.TestCase):
self.assertEqual(statuses["forbidden_margin_terms"], "fail")
self.assertEqual(statuses["margin_domain_leak_accounting_route"], "fail")
def test_stage_review_signal_detector_fails_when_issue_code_is_present(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "review"
write_json(
artifact_dir / "run_review.json",
{
"summary": {"overall_business_status": "fail"},
"findings": [{"issue_codes": ["backend_error_response"]}],
},
)
registry_path = root / "detector_registry.json"
issue_catalog_path = root / "issue_catalog.json"
write_json(
registry_path,
{
"schema_version": "agent_detector_registry_v1",
"detectors": {
"backend_error_response_signal": {
"kind": "stage_review_signal",
"automation_level": "automatic",
"description": "Backend error issue in GUI review.",
"issue_codes": ["backend_error_response"],
"inputs": ["run_review.json"],
"check": {"target_status": "backend_error_response"},
}
},
},
)
write_json(
issue_catalog_path,
{
"schema_version": "agent_issue_catalog_v1",
"issues": {
"backend_error_response": {
"detectors": ["backend_error_response_signal"],
}
},
},
)
results = runner.build_detector_results(
artifact_dir,
issue_codes=["backend_error_response"],
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]["detector"], "backend_error_response_signal")
self.assertEqual(results["results"][0]["status"], "fail")
def test_stage_review_signal_detector_ignores_possible_invariant_names(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
artifact_dir = root / "review"
write_json(
artifact_dir / "run_review.json",
{
"summary": {"overall_business_status": "pass", "issue_counts": {}},
"findings": [],
"repair_targets": [],
"step_states": [
{
"business_first_review": {
"invariant_severity": {
"backend_error_response": "P0"
}
}
}
],
},
)
registry_path = root / "detector_registry.json"
issue_catalog_path = root / "issue_catalog.json"
write_json(
registry_path,
{
"schema_version": "agent_detector_registry_v1",
"detectors": {
"backend_error_response_signal": {
"kind": "stage_review_signal",
"automation_level": "automatic",
"description": "Backend error issue in GUI review.",
"issue_codes": ["backend_error_response"],
"inputs": ["run_review.json"],
"check": {"target_status": "backend_error_response"},
}
},
},
)
write_json(
issue_catalog_path,
{
"schema_version": "agent_issue_catalog_v1",
"issues": {
"backend_error_response": {
"detectors": ["backend_error_response_signal"],
}
},
},
)
results = runner.build_detector_results(
artifact_dir,
issue_codes=["backend_error_response"],
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")
if __name__ == "__main__":
unittest.main()

View File

@ -23,6 +23,14 @@ class AgentReliabilityContractHealthcheckTests(unittest.TestCase):
"agent_margin_profitability_reliability_20260524",
result["domain_scenario_packs"]["pack_ids"],
)
self.assertIn(
"backend_error_response",
result["issue_catalog"]["gui_review_required_issue_codes"],
)
self.assertIn(
"year_ranking_direct_answer_missing",
result["issue_catalog"]["gui_review_required_issue_codes"],
)
def test_issue_catalog_healthcheck_blocks_auto_coder_issue_without_contract(self) -> None:
with tempfile.TemporaryDirectory() as tmp:

View File

@ -239,10 +239,51 @@ class AssistantStage1RunReviewTests(unittest.TestCase):
self.assertTrue((output_dir / "run_review.json").exists())
self.assertTrue((output_dir / "run_review.md").exists())
self.assertTrue((output_dir / "detector_candidates.json").exists())
detector_candidates = json.loads((output_dir / "detector_candidates.json").read_text(encoding="utf-8"))
self.assertEqual(detector_candidates["schema_version"], "agent_detector_candidates_v1")
self.assertEqual(detector_candidates["candidate_count"], 0)
markdown = (output_dir / "run_review.md").read_text(encoding="utf-8")
self.assertIn("overall_business_status", markdown)
self.assertIn("Question Quality", markdown)
def test_detector_candidates_use_issue_catalog_detectors(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)
catalog_path = root / "issue_catalog.json"
write_json(
catalog_path,
{
"schema_version": "agent_issue_catalog_v1",
"issues": {
"backend_error_response": {
"detectors": ["backend_error_response_signal"]
}
},
},
)
review = {
"run_id": "assistant-stage1-detector-candidates",
"findings": [
{
"step_id": "turn_001",
"turn_index": 1,
"question": "приветик - че как там дела",
"severity": "P0",
"issue_severities": {"backend_error_response": "P0"},
"issue_codes": ["backend_error_response"],
}
],
}
candidates = reviewer.build_detector_candidates(review, issue_catalog_path=catalog_path)
self.assertEqual(candidates["candidate_count"], 1)
candidate = candidates["candidates"][0]
self.assertEqual(candidate["issue_code"], "backend_error_response")
self.assertEqual(candidate["detector"], "backend_error_response_signal")
self.assertEqual(candidate["evidence_paths"], ["run_review.json", "conversation_pairs.json"])
def test_review_flags_mcp_leak_and_bank_role_misclassification(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
root = Path(tmp)