Укрепить агентный loop Phase107

scripts/agent_detector_runner.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import os
 import re
 from datetime import datetime, timezone
 from pathlib import Path
@@ -18,8 +19,35 @@ DEFAULT_LIMITED_NEXT_ACTION_EXTRA_PATTERNS = [
 ]
 
 
+def path_for_io(path: Path) -> str:
+    if os.name != "nt":
+        return str(path)
+    raw_value = str(path)
+    if raw_value.startswith("\\\\?\\"):
+        return raw_value
+    absolute_path = path if path.is_absolute() else Path.cwd() / path
+    absolute_value = str(absolute_path.absolute())
+    if absolute_value.startswith("\\\\?\\"):
+        return absolute_value
+    if absolute_value.startswith("\\\\"):
+        return "\\\\?\\UNC\\" + absolute_value.lstrip("\\")
+    return "\\\\?\\" + absolute_value
+
+
+def path_exists(path: Path) -> bool:
+    try:
+        return os.path.exists(path_for_io(path))
+    except OSError:
+        return False
+
+
+def read_text(path: Path) -> str:
+    with open(path_for_io(path), "r", encoding="utf-8") as handle:
+        return handle.read()
+
+
 def read_json(path: Path) -> Any:
-    return json.loads(path.read_text(encoding="utf-8"))
+    return json.loads(read_text(path))
 
 
 def read_json_object(path: Path) -> dict[str, Any]:
@@ -31,8 +59,9 @@ def read_json_object(path: Path) -> dict[str, Any]:
 
 
 def write_json(path: Path, payload: Any) -> None:
-    path.parent.mkdir(parents=True, exist_ok=True)
-    path.write_text(json.dumps(payload, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    os.makedirs(path_for_io(path.parent), exist_ok=True)
+    with open(path_for_io(path), "w", encoding="utf-8") as handle:
+        handle.write(json.dumps(payload, ensure_ascii=False, indent=2) + "\n")
 
 
 def utc_now() -> str:
@@ -164,7 +193,7 @@ def select_detectors(
 
 def read_text_or_empty(path: Path) -> str:
     try:
-        return path.read_text(encoding="utf-8")
+        return read_text(path)
     except OSError:
         return ""
 
@@ -207,7 +236,7 @@ def first_pattern_search(patterns: list[re.Pattern[str]], text: str) -> re.Match
 
 
 def assistant_text_from_turn_path(path: Path | None) -> str:
-    if path is None or not path.exists():
+    if path is None or not path_exists(path):
         return ""
     payload = read_json_object(path)
     assistant_message = payload.get("assistant_message") if isinstance(payload.get("assistant_message"), dict) else {}
@@ -264,14 +293,21 @@ def output_turn_path(output_path: Path) -> Path | None:
     name = output_path.name
     if name == "output.md":
         candidate = output_path.with_name("turn.json")
-        return candidate if candidate.exists() else None
+        return candidate if path_exists(candidate) else None
     if name.endswith("_output.md"):
         prefix = name[: -len("_output.md")]
         candidate = output_path.with_name(f"{prefix}_turn.json")
-        return candidate if candidate.exists() else None
+        return candidate if path_exists(candidate) else None
     return None
 
 
+def output_step_state_path(output_path: Path) -> Path | None:
+    if output_path.name != "output.md":
+        return None
+    candidate = output_path.with_name("step_state.json")
+    return candidate if path_exists(candidate) else None
+
+
 def collect_output_artifacts(artifact_dir: Path) -> list[dict[str, Any]]:
     outputs: list[dict[str, Any]] = []
     seen: set[Path] = set()
@@ -292,6 +328,7 @@ def collect_output_artifacts(artifact_dir: Path) -> list[dict[str, Any]]:
                     "artifact_path": str(path.relative_to(artifact_dir)),
                     "text": read_text_or_empty(path),
                     "turn_path": turn_path,
+                    "step_state_path": output_step_state_path(path),
                 }
             )
     return outputs
@@ -497,6 +534,52 @@ def evaluate_limited_next_action(
     return build_result(detector_name, detector, status, message, evidence=failures)
 
 
+def evaluate_answer_text_shape(
+    detector_name: str,
+    detector: dict[str, Any],
+    outputs: list[dict[str, Any]],
+) -> dict[str, Any]:
+    if not outputs:
+        return build_result(detector_name, detector, "skipped", "no output.md-style artifacts matched detector scope")
+    failures: list[dict[str, Any]] = []
+    reviewed: list[dict[str, Any]] = []
+    unknown: list[dict[str, Any]] = []
+    for output in outputs:
+        step_state_path = output.get("step_state_path")
+        step_state = read_json_object(step_state_path) if isinstance(step_state_path, Path) else {}
+        review = step_state.get("business_first_review") if isinstance(step_state.get("business_first_review"), dict) else {}
+        direct_answer_first_ok = review.get("direct_answer_first_ok")
+        if direct_answer_first_ok is True:
+            reviewed.append({"path": output["repo_path"], "direct_answer_first_ok": True})
+        elif direct_answer_first_ok is False:
+            failures.append({"path": output["repo_path"], "direct_answer_first_ok": False})
+        else:
+            unknown.append({"path": output["repo_path"], "reason": "business_first_review_missing"})
+    if failures:
+        return build_result(
+            detector_name,
+            detector,
+            "fail",
+            "first-line direct answer check failed",
+            evidence=[*failures, *unknown],
+        )
+    if unknown:
+        return build_result(
+            detector_name,
+            detector,
+            "review",
+            "direct-answer shape requires business review",
+            evidence=[*reviewed, *unknown],
+        )
+    return build_result(
+        detector_name,
+        detector,
+        "pass",
+        "business-first step reviews confirm direct-answer-first shape",
+        evidence=reviewed,
+    )
+
+
 def evaluate_trace_guard(
     detector_name: str,
     detector: dict[str, Any],
@@ -686,7 +769,7 @@ def evaluate_detector(
     ):
         return evaluate_composite(detector_name, detector, results_by_name)
     if kind == "answer_text_shape":
-        return evaluate_manual_review(detector_name, detector, "direct-answer shape requires business review")
+        return evaluate_answer_text_shape(detector_name, detector, scoped_outputs)
     return build_result(detector_name, detector, "skipped", f"detector kind is not executable yet: {kind}")
 
 

scripts/domain_case_loop.py

@@ -317,6 +317,7 @@ DEFAULT_INVARIANT_SEVERITY: dict[str, str] = {
     "forbidden_recipe_selected": "P0",
     "focus_object_missing": "P0",
     "wrong_date_scope_state": "P0",
+    "out_of_window_date_in_answer": "P0",
     "direct_answer_missing": "P0",
     "top_level_noise_present": "P0",
     "business_direct_answer_missing": "P0",
@@ -930,6 +931,40 @@ def first_non_empty_lines(text: str, limit: int = 3) -> list[str]:
     return output
 
 
+def expected_single_year_from_date_scope(date_scope: Any) -> str | None:
+    if not isinstance(date_scope, dict):
+        return None
+    if str(date_scope.get("scope") or "").strip().lower() == "all_time":
+        return None
+    period_from = normalize_iso_date(date_scope.get("period_from"))
+    period_to = normalize_iso_date(date_scope.get("period_to"))
+    if not period_from or not period_to:
+        return None
+    from_year = period_from[:4]
+    to_year = period_to[:4]
+    if from_year != to_year:
+        return None
+    return from_year if re.fullmatch(r"(?:19|20)\d{2}", from_year) else None
+
+
+def dated_evidence_years_from_answer(text: Any) -> list[str]:
+    source = repair_text_mojibake(str(text or ""))
+    years: list[str] = []
+    for match in re.finditer(r"\b((?:19|20)\d{2})-\d{2}-\d{2}(?:T|\b)", source):
+        years.append(match.group(1))
+    for match in re.finditer(r"\b\d{1,2}\.\d{1,2}\.((?:19|20)\d{2})\b", source):
+        years.append(match.group(1))
+    return list(dict.fromkeys(years))
+
+
+def answer_has_out_of_window_dates_for_scope(text: Any, date_scope: Any) -> bool:
+    expected_year = expected_single_year_from_date_scope(date_scope)
+    if not expected_year:
+        return False
+    years = dated_evidence_years_from_answer(text)
+    return any(year != expected_year for year in years)
+
+
 def build_node_contract_index(raw_contract: dict[str, Any]) -> dict[str, dict[str, Any]]:
     scenario_tree = raw_contract.get("scenario_tree")
     if not isinstance(scenario_tree, dict):
@@ -2867,7 +2902,12 @@ def is_validated_clean_meta_chat_answer(
         return False
     semantic_tags = set(normalize_string_list(state.get("semantic_tags")))
     allowed_tags = {
+        "human_answer",
         "meta_smalltalk",
+        "mcp_discovery_gate_sanity",
+        "off_domain_living_chat",
+        "stale_replay_forbidden",
+        "context_boundary",
         "company_selected",
         "organization_authority",
         "meta_capability",
@@ -3112,6 +3152,10 @@ def validate_step_contract(step_state: dict[str, Any]) -> dict[str, Any]:
         if current_date_scope and current_date_scope != required_filters["as_of_date"]:
             violated_invariants.append("wrong_date_scope_state")
 
+    if answer_has_out_of_window_dates_for_scope(assistant_text, date_scope):
+        violated_invariants.append("out_of_window_date_in_answer")
+        warnings.append("out_of_window_date_in_answer")
+
     if should_require_direct_answer(state):
         if not actual_direct_answer or is_top_level_noise_line(actual_direct_answer):
             violated_invariants.append("direct_answer_missing")
@@ -3193,8 +3237,10 @@ def validate_step_contract(step_state: dict[str, Any]) -> dict[str, Any]:
     state["clarification_answer_validated"] = clarification_validated
     state["missing_axis_clarification_validated"] = missing_axis_clarification_validated
     state["clean_meta_chat_answer_validated"] = clean_meta_chat_validated
+    effective_execution_status = "exact" if clean_meta_chat_validated else execution_status
+    state["execution_status"] = effective_execution_status
     state["acceptance_status"] = acceptance_status_from_execution(
-        execution_status,
+        effective_execution_status,
         hard_fail,
         (
             bounded_validated

scripts/test_agent_detector_runner.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import os
 import sys
 import tempfile
 import unittest
@@ -287,6 +288,182 @@ class AgentDetectorRunnerTests(unittest.TestCase):
         self.assertEqual(len(evidence_paths), 1)
         self.assertIn("step_01_margin_root", evidence_paths[0])
 
+    def test_counterparty_value_flow_required_surface_scopes_to_current_net_step(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            artifact_dir = root / "run"
+            write_text(
+                artifact_dir
+                / "scenarios"
+                / "mixed_planner_counterparty_evidence_and_living_guard"
+                / "steps"
+                / "step_03_incoming_by_resolved_entity"
+                / "output.md",
+                "Входящие денежные поступления по контрагенту Группа СВК за 2020: 12 093 465 руб.",
+            )
+            write_text(
+                artifact_dir
+                / "scenarios"
+                / "mixed_planner_counterparty_evidence_and_living_guard"
+                / "steps"
+                / "step_05_net_after_payout"
+                / "output.md",
+                "По контрагенту Группа СВК за период 2020 получили 12 093 465 руб., заплатили 0 руб.; расчетное нетто в нашу сторону: 12 093 465 руб.",
+            )
+            registry_path = root / "detector_registry.json"
+            issue_catalog_path = root / "issue_catalog.json"
+            write_json(
+                registry_path,
+                {
+                    "schema_version": "agent_detector_registry_v1",
+                    "detectors": {
+                        "counterparty_value_flow_required_surface": {
+                            "kind": "answer_text_required_any",
+                            "automation_level": "automatic",
+                            "description": "Net value-flow answer must surface all money directions.",
+                            "issue_codes": ["counterparty_value_flow_misrouted_to_company_profit"],
+                            "inputs": ["output.md"],
+                            "check": {
+                                "artifact_path_include_patterns": ["(?i)step_05_net_after_payout"],
+                                "required_patterns_any": [
+                                    "(?is)(?=.*(СВК|Группа\\s+СВК))(?=.*(входящ|получил|получено|получили))(?=.*(исходящ|заплатил|заплачено|заплатили|ушло))(?=.*(нетто|сальдо|разниц|чистый\\s+денежный))"
+                                ],
+                            },
+                        }
+                    },
+                },
+            )
+            write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
+
+            results = runner.build_detector_results(
+                artifact_dir,
+                detector_names=["counterparty_value_flow_required_surface"],
+                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")
+        evidence_paths = [item["path"] for item in results["results"][0]["evidence"]]
+        self.assertEqual(len(evidence_paths), 1)
+        self.assertIn("step_05_net_after_payout", evidence_paths[0])
+
+    def test_answer_text_shape_uses_business_first_step_review(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            artifact_dir = root / "run"
+            step_dir = artifact_dir / "scenarios" / "mixed" / "steps" / "step_01"
+            write_text(step_dir / "output.md", "Direct business answer first.")
+            write_json(step_dir / "step_state.json", {"business_first_review": {"direct_answer_first_ok": True}})
+            registry_path = root / "detector_registry.json"
+            issue_catalog_path = root / "issue_catalog.json"
+            write_json(
+                registry_path,
+                {
+                    "schema_version": "agent_detector_registry_v1",
+                    "detectors": {
+                        "first_line_not_direct_answer": {
+                            "kind": "answer_text_shape",
+                            "automation_level": "semi_automatic",
+                            "description": "First line should be direct.",
+                            "issue_codes": ["business_direct_answer_missing"],
+                            "inputs": ["output.md"],
+                            "check": {"first_line_should_be": "business_answer_or_honest_boundary"},
+                        }
+                    },
+                },
+            )
+            write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
+
+            results = runner.build_detector_results(
+                artifact_dir,
+                detector_names=["first_line_not_direct_answer"],
+                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_answer_text_shape_fails_when_business_first_review_rejects_first_line(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            artifact_dir = root / "run"
+            step_dir = artifact_dir / "scenarios" / "mixed" / "steps" / "step_01"
+            write_text(step_dir / "output.md", "Let me inspect the route first.")
+            write_json(step_dir / "step_state.json", {"business_first_review": {"direct_answer_first_ok": False}})
+            registry_path = root / "detector_registry.json"
+            issue_catalog_path = root / "issue_catalog.json"
+            write_json(
+                registry_path,
+                {
+                    "schema_version": "agent_detector_registry_v1",
+                    "detectors": {
+                        "first_line_not_direct_answer": {
+                            "kind": "answer_text_shape",
+                            "automation_level": "semi_automatic",
+                            "description": "First line should be direct.",
+                            "issue_codes": ["business_direct_answer_missing"],
+                            "inputs": ["output.md"],
+                            "check": {"first_line_should_be": "business_answer_or_honest_boundary"},
+                        }
+                    },
+                },
+            )
+            write_json(issue_catalog_path, {"schema_version": "agent_issue_catalog_v1", "issues": {}})
+
+            results = runner.build_detector_results(
+                artifact_dir,
+                detector_names=["first_line_not_direct_answer"],
+                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_step_state_lookup_handles_long_artifact_paths(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            step_dir = (
+                root
+                / "run"
+                / "scenarios"
+                / ("mixed_planner_counterparty_evidence_and_living_guard_" + "s" * 48)
+                / "steps"
+                / ("step_04_payout_switch_by_resolved_entity_" + "x" * 80)
+            )
+            output_path = step_dir / "output.md"
+            step_state_path = step_dir / "step_state.json"
+            try:
+                os.makedirs(runner.path_for_io(step_dir), exist_ok=True)
+                with open(runner.path_for_io(output_path), "w", encoding="utf-8") as handle:
+                    handle.write("Direct business answer first.")
+                runner.write_json(step_state_path, {"business_first_review": {"direct_answer_first_ok": True}})
+
+                self.assertGreater(len(str(step_state_path.absolute())), 260)
+                self.assertEqual(runner.output_step_state_path(output_path), step_state_path)
+                self.assertEqual(
+                    runner.read_json_object(step_state_path).get("business_first_review"),
+                    {"direct_answer_first_ok": True},
+                )
+            finally:
+                for path in (output_path, step_state_path):
+                    try:
+                        os.remove(runner.path_for_io(path))
+                    except OSError:
+                        pass
+                current = step_dir
+                while current != root:
+                    try:
+                        os.rmdir(runner.path_for_io(current))
+                    except OSError:
+                        pass
+                    current = current.parent
+
     def test_composite_detector_fails_after_child_detector_fails(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)

scripts/test_domain_case_loop_step_state.py

@@ -153,6 +153,81 @@ class DomainCaseLoopStepStateTests(unittest.TestCase):
         self.assertEqual(dcl.derive_scenario_execution_status(step_outputs), "partial")
         self.assertEqual(dcl.derive_scenario_status(step_outputs), "accepted")
 
+    def test_clean_meta_smalltalk_is_exact_without_1c_capability(self) -> None:
+        step_state = dcl.build_scenario_step_state(
+            scenario_id="meta_chat_demo",
+            domain="agentic_loop",
+            step={
+                "step_id": "step_01",
+                "title": "Human smalltalk",
+                "depends_on": [],
+                "question_template": "привет, ты на связи?",
+                "semantic_tags": ["human_answer", "meta_smalltalk", "mcp_discovery_gate_sanity"],
+                "required_answer_shape": "direct_answer_first",
+            },
+            step_index=1,
+            question_resolved="привет, ты на связи?",
+            analysis_context={},
+            turn_artifact={
+                "assistant_message": {
+                    "reply_type": "factual_with_explanation",
+                    "text": "Привет! Да, я на связи. Готов помочь с анализом данных из 1С в режиме чтения.",
+                    "message_id": "msg-1",
+                    "trace_id": "trace-1",
+                },
+                "technical_debug_payload": {
+                    "detected_mode": "chat",
+                    "fallback_type": "none",
+                    "living_chat_response_source": "llm_chat",
+                },
+                "session_summary": {},
+            },
+            entries=[],
+        )
+
+        self.assertTrue(step_state["clean_meta_chat_answer_validated"])
+        self.assertEqual(step_state["execution_status"], "exact")
+        self.assertEqual(step_state["acceptance_status"], "validated")
+
+    def test_off_domain_living_chat_is_exact_without_1c_capability(self) -> None:
+        step_state = dcl.build_scenario_step_state(
+            scenario_id="off_domain_demo",
+            domain="agentic_loop",
+            step={
+                "step_id": "step_09",
+                "title": "Off-domain living chat",
+                "depends_on": [],
+                "question_template": "а чем капибара отличается от утки?",
+                "semantic_tags": ["off_domain_living_chat", "stale_replay_forbidden", "context_boundary"],
+                "required_answer_shape": "direct_answer_first",
+            },
+            step_index=9,
+            question_resolved="а чем капибара отличается от утки?",
+            analysis_context={},
+            turn_artifact={
+                "assistant_message": {
+                    "reply_type": "factual_with_explanation",
+                    "text": (
+                        "Капибара и утка отличаются принципиально: капибара - млекопитающее-грызун, "
+                        "а утка - птица. Поэтому у них разные тело, среда обитания и способ передвижения."
+                    ),
+                    "message_id": "msg-2",
+                    "trace_id": "trace-2",
+                },
+                "technical_debug_payload": {
+                    "detected_mode": "chat",
+                    "fallback_type": "none",
+                    "living_chat_response_source": "llm_chat",
+                },
+                "session_summary": {},
+            },
+            entries=[],
+        )
+
+        self.assertTrue(step_state["clean_meta_chat_answer_validated"])
+        self.assertEqual(step_state["execution_status"], "exact")
+        self.assertEqual(step_state["acceptance_status"], "validated")
+
     def test_today_scope_required_filter_and_direct_patterns_are_enforced(self) -> None:
         self.assertTrue(dcl.question_resets_temporal_scope("мы должны комуто денег на сегодня?"))
 
@@ -610,6 +685,32 @@ class DomainCaseLoopStepStateTests(unittest.TestCase):
         self.assertIsNone(step_state["date_scope"]["as_of_date"])
         self.assertEqual(step_state["date_scope"]["source"], "question_temporal_scope_reset")
 
+    def test_out_of_window_document_dates_reject_validated_step(self) -> None:
+        validated = dcl.validate_step_contract(
+            {
+                "execution_status": "exact",
+                "required_answer_shape": "direct_answer_first",
+                "reply_type": "factual",
+                "actual_direct_answer": "Контрагент: Группа СВК. Найдено документов: 19.",
+                "assistant_text": (
+                    "Контрагент: Группа СВК. Найдено документов: 19.\n"
+                    "1. 2021-11-10T12:00:07Z | Поступление на расчетный счет 00000000013 от 10.11.2021 12:00:07"
+                ),
+                "top_non_empty_lines": [
+                    "Контрагент: Группа СВК. Найдено документов: 19.",
+                    "1. 2021-11-10T12:00:07Z | Поступление на расчетный счет 00000000013 от 10.11.2021 12:00:07",
+                ],
+                "date_scope": {
+                    "period_from": "2020-01-01",
+                    "period_to": "2020-12-31",
+                },
+            }
+        )
+
+        self.assertEqual(validated["status"], "rejected")
+        self.assertIn("out_of_window_date_in_answer", validated["violated_invariants"])
+        self.assertIn("out_of_window_date_in_answer", validated["warnings"])
+
     def test_open_items_exact_negative_answer_validates_without_rows(self) -> None:
         step_state = dcl.build_scenario_step_state(
             scenario_id="open_items_negative_demo",