Привязать detector gate к accepted agent loop

docs/orchestration/domain_scenario_loop_repo_adapter.md

@@ -195,6 +195,8 @@ When the coder result is `patched`, the next `ingest-gui-run` is treated as post
 
 The stage closing gate enforces that rule even when the inner pack loop reports `accepted`: `loop_accepted_gate` preserves the raw loop verdict, but stage-level `accepted_gate` stays `false` with `stage_closing_gate.status = blocked_pending_repair_validation` until the latest patched repair has a matching successful validation run.
 
+Inside the autonomous domain pack loop, acceptance is detector-aware as well: `accepted_gate` requires analyst acceptance, deterministic gate success, and a clean detector-results gate. If `detector_results.json` reports `fail` or `review`, the loop must not close as accepted; it should surface the failed/review detector names in `business_audit.json`, `loop_state.json`, and the Lead Codex handoff so the next repair pass starts from the concrete user-facing signal instead of a hidden green status.
+
 ## Placeholder contract
 
 Scenario questions can reference earlier step outputs with placeholders such as:

scripts/agent_detector_runner.py

@@ -238,7 +238,11 @@ def collect_output_artifacts(artifact_dir: Path) -> list[dict[str, Any]]:
     for path in sorted(artifact_dir.rglob("*.md")):
         if path in seen:
             continue
-        if path.name == "output.md" or path.name in {"scenario_output.md"} or path.name.endswith("_output.md"):
+        # scenario_output.md is an aggregate technical export and often contains
+        # debug payloads that are not part of the user-facing assistant answer.
+        if path.name == "scenario_output.md":
+            continue
+        if path.name == "output.md" or path.name.endswith("_output.md"):
             seen.add(path)
             turn_path = output_turn_path(path)
             outputs.append(

scripts/domain_case_loop.py

@@ -6010,6 +6010,56 @@ def detector_evidence_paths_for_target(target: dict[str, Any]) -> list[str]:
     return [str(step_state.with_name("output.md")), str(step_state.with_name("turn.json"))]
 
 
+ACCEPTED_PACK_SMOKE_DETECTORS = [
+    {
+        "issue_code": "technical_garbage_in_answer",
+        "detector": "runtime_tokens_in_user_answer",
+        "severity": "P1",
+    },
+    {
+        "issue_code": "technical_garbage_in_answer",
+        "detector": "capability_ids_in_user_answer",
+        "severity": "P1",
+    },
+    {
+        "issue_code": "business_direct_answer_missing",
+        "detector": "top_level_scaffold_before_answer",
+        "severity": "P1",
+    },
+    {
+        "issue_code": "business_next_step_missing",
+        "detector": "limited_answer_without_next_action",
+        "severity": "P2",
+    },
+]
+
+
+def should_add_accepted_pack_smoke_detectors(
+    repair_targets: dict[str, Any],
+    existing_candidates: list[dict[str, Any]],
+) -> bool:
+    if existing_candidates:
+        return False
+    if int(repair_targets.get("target_count") or 0) != 0:
+        return False
+    accepted_statuses = {
+        str(repair_targets.get("acceptance_status") or "").strip(),
+        str(repair_targets.get("final_status") or "").strip(),
+    }
+    return "accepted" in accepted_statuses
+
+
+def build_accepted_pack_smoke_detector_candidates() -> list[dict[str, Any]]:
+    return [
+        {
+            **candidate,
+            "sample_target_id": "accepted_pack_smoke",
+            "evidence_paths": [],
+        }
+        for candidate in ACCEPTED_PACK_SMOKE_DETECTORS
+    ]
+
+
 def build_detector_candidates(repair_targets: dict[str, Any], catalog: dict[str, Any] | None = None) -> dict[str, Any]:
     source = catalog if isinstance(catalog, dict) else load_issue_catalog()
     issues = source.get("issues") if isinstance(source.get("issues"), dict) else {}
@@ -6038,6 +6088,8 @@ def build_detector_candidates(repair_targets: dict[str, Any], catalog: dict[str,
                     "evidence_paths": evidence_paths,
                 }
             )
+    if should_add_accepted_pack_smoke_detectors(repair_targets, candidates):
+        candidates.extend(build_accepted_pack_smoke_detector_candidates())
     return {
         "schema_version": "detector_candidates_v1",
         "candidate_count": len(candidates),
@@ -6087,6 +6139,19 @@ def summarize_detector_results(detector_results: dict[str, Any] | None, *, limit
     }
 
 
+def evaluate_detector_results_gate(detector_results_summary: dict[str, Any]) -> tuple[bool, str]:
+    status = str(detector_results_summary.get("status") or "not_run").strip()
+    if status == "fail":
+        failed = normalize_string_list(detector_results_summary.get("failed_detectors"))
+        suffix = ",".join(failed) if failed else str(detector_results_summary.get("fail") or 0)
+        return False, f"detector_results_failed:{suffix}"
+    if status == "review":
+        review = normalize_string_list(detector_results_summary.get("review_detectors"))
+        suffix = ",".join(review) if review else str(detector_results_summary.get("review") or 0)
+        return False, f"detector_results_review_required:{suffix}"
+    return True, f"detector_results_{status or 'not_run'}"
+
+
 def build_blocking_issue_contract(target: dict[str, Any], catalog: dict[str, Any]) -> dict[str, Any]:
     issue_code = str(target.get("issue_code") or target.get("problem_type") or "other").strip()
     entry = issue_catalog_entry(issue_code, catalog)
@@ -6141,6 +6206,8 @@ def build_business_audit_contract(
         if isinstance(target, dict) and str(target.get("severity") or "").upper() in {"P0", "P1"}
     ]
     rerun_matrix = collect_rerun_matrix(repair_targets)
+    detector_results_summary = summarize_detector_results(detector_results)
+    detector_results_gate_ok, detector_results_gate_reason = evaluate_detector_results_gate(detector_results_summary)
     result = {
         "schema_version": "business_audit_contract_v1",
         "created_at": datetime.now(timezone.utc).replace(microsecond=0).isoformat(),
@@ -6152,6 +6219,8 @@ def build_business_audit_contract(
         "accepted_gate": accepted_gate,
         "deterministic_gate_ok": deterministic_gate_ok,
         "deterministic_gate_reason": deterministic_gate_reason,
+        "detector_results_gate_ok": detector_results_gate_ok,
+        "detector_results_gate_reason": detector_results_gate_reason,
         "human_meaning": {
             "user_intent_summary": analyst_verdict.get("user_intent_summary"),
             "expected_direct_answer": analyst_verdict.get("expected_direct_answer"),
@@ -6173,7 +6242,7 @@ def build_business_audit_contract(
             "severity_counts": repair_targets.get("severity_counts") or {},
             "priority_foci": _limited_dict_items(repair_targets.get("priority_foci"), limit=8),
         },
-        "detector_results_summary": summarize_detector_results(detector_results),
+        "detector_results_summary": detector_results_summary,
         "rerun_matrix": rerun_matrix,
         "artifact_refs": {
             "business_audit_md": repo_relative(business_audit_markdown_path),
@@ -6320,6 +6389,8 @@ def build_lead_coder_handoff(
         for item in repair_items
         if isinstance(item, dict) and str(item.get("target_source") or "") == "route_candidate_enablement"
     ]
+    detector_results_summary = summarize_detector_results(detector_results)
+    detector_results_gate_ok, detector_results_gate_reason = evaluate_detector_results_gate(detector_results_summary)
     candidate_files = [repo_relative(path) for path in build_coder_snapshot_paths(repair_targets)]
     artifact_refs = {
         "pack_dir": repo_relative(pack_dir),
@@ -6356,13 +6427,15 @@ def build_lead_coder_handoff(
         "accepted_gate": accepted_gate,
         "deterministic_gate_ok": deterministic_gate_ok,
         "deterministic_gate_reason": deterministic_gate_reason,
+        "detector_results_gate_ok": detector_results_gate_ok,
+        "detector_results_gate_reason": detector_results_gate_reason,
         "requires_user_decision": requires_user_decision,
         "user_decision_type": user_decision_type,
         "user_decision_prompt": user_decision_prompt,
         "artifact_refs": artifact_refs,
         "issue_codes": issue_codes,
         "rerun_matrix": rerun_matrix,
-        "detector_results_summary": summarize_detector_results(detector_results),
+        "detector_results_summary": detector_results_summary,
         "human_meaning": {
             "user_intent_summary": analyst_verdict.get("user_intent_summary"),
             "expected_direct_answer": analyst_verdict.get("expected_direct_answer"),
@@ -6407,6 +6480,8 @@ def build_lead_coder_handoff_markdown(handoff: dict[str, Any]) -> str:
         f"- loop_decision: `{handoff.get('loop_decision')}`",
         f"- deterministic_gate_ok: `{handoff.get('deterministic_gate_ok')}`",
         f"- deterministic_gate_reason: `{handoff.get('deterministic_gate_reason') or 'n/a'}`",
+        f"- detector_results_gate_ok: `{handoff.get('detector_results_gate_ok')}`",
+        f"- detector_results_gate_reason: `{handoff.get('detector_results_gate_reason') or 'n/a'}`",
         "",
         "## Read First",
         f"- business_audit: `{artifact_refs.get('business_audit')}`",
@@ -6898,7 +6973,6 @@ def handle_run_pack_loop(args: argparse.Namespace) -> int:
                 target_score=target_score,
             ),
         )
-        accepted_gate = analyst_accepted_gate and deterministic_gate_ok
         issue_catalog_snapshot = build_issue_catalog_snapshot(repair_targets)
         rerun_matrix_contract = {
             "schema_version": "rerun_matrix_v1",
@@ -6914,6 +6988,8 @@ def handle_run_pack_loop(args: argparse.Namespace) -> int:
         )
         write_json(detector_results_path, detector_results)
         detector_results_summary = summarize_detector_results(detector_results)
+        detector_results_gate_ok, detector_results_gate_reason = evaluate_detector_results_gate(detector_results_summary)
+        accepted_gate = analyst_accepted_gate and deterministic_gate_ok and detector_results_gate_ok
         business_audit_contract = build_business_audit_contract(
             analyst_verdict=analyst_verdict,
             repair_targets=repair_targets,
@@ -6961,6 +7037,8 @@ def handle_run_pack_loop(args: argparse.Namespace) -> int:
             "accepted_gate": accepted_gate,
             "deterministic_gate_ok": deterministic_gate_ok,
             "deterministic_gate_reason": deterministic_gate_reason,
+            "detector_results_gate_ok": detector_results_gate_ok,
+            "detector_results_gate_reason": detector_results_gate_reason,
             "requires_user_decision": requires_user_decision,
             "user_decision_type": user_decision_type,
             "user_decision_prompt": user_decision_prompt,

scripts/test_agent_detector_runner.py

@@ -273,6 +273,22 @@ class AgentDetectorRunnerTests(unittest.TestCase):
         self.assertEqual(results["summary"]["status"], "pass")
         self.assertEqual(results["results"][0]["status"], "pass")
 
+    def test_answer_text_outputs_ignore_aggregate_scenario_export(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            artifact_dir = root / "run"
+            write_text(
+                artifact_dir / "scenarios" / "inventory" / "scenario_output.md",
+                "# Technical export\n\ncapability_id: address.inventory.debug\nruntime_debug: true\n",
+            )
+            step_output = artifact_dir / "scenarios" / "inventory" / "steps" / "s01" / "output.md"
+            write_text(step_output, "Clean user-facing answer.")
+
+            outputs = runner.collect_output_artifacts(artifact_dir)
+
+        artifact_paths = [item["artifact_path"].replace("\\", "/") for item in outputs]
+        self.assertEqual(artifact_paths, ["scenarios/inventory/steps/s01/output.md"])
+
     def test_trace_guard_uses_current_route_fields_not_full_turn_history(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)

scripts/test_domain_case_loop_lead_handoff.py

@@ -104,6 +104,8 @@ class DomainCaseLoopLeadHandoffTests(unittest.TestCase):
         self.assertIn("detector_results", saved["artifact_refs"])
         self.assertEqual(saved["detector_results_summary"]["status"], "fail")
         self.assertEqual(saved["detector_results_summary"]["failed_detectors"], ["first_line_not_direct_answer"])
+        self.assertFalse(saved["detector_results_gate_ok"])
+        self.assertEqual(saved["detector_results_gate_reason"], "detector_results_failed:first_line_not_direct_answer")
         self.assertIn("business_direct_answer_missing", saved["issue_codes"])
         self.assertIn("failed_scenario", saved["rerun_matrix"])
         self.assertTrue(latest_handoff_exists)
@@ -163,6 +165,67 @@ class DomainCaseLoopLeadHandoffTests(unittest.TestCase):
         self.assertIn("detector_results_json", contract["artifact_refs"])
         self.assertEqual(contract["detector_results_summary"]["status"], "review")
         self.assertEqual(contract["detector_results_summary"]["review_detectors"], ["missing_revenue_cogs_margin_fields"])
+        self.assertFalse(contract["detector_results_gate_ok"])
+        self.assertEqual(
+            contract["detector_results_gate_reason"],
+            "detector_results_review_required:missing_revenue_cogs_margin_fields",
+        )
+
+    def test_accepted_pack_without_repair_targets_gets_smoke_detector_candidates(self) -> None:
+        candidates = dcl.build_detector_candidates(
+            {
+                "target_count": 0,
+                "acceptance_status": "accepted",
+                "final_status": "accepted",
+                "targets": [],
+            }
+        )
+
+        detector_names = [item["detector"] for item in candidates["candidates"]]
+
+        self.assertEqual(candidates["candidate_count"], 4)
+        self.assertIn("runtime_tokens_in_user_answer", detector_names)
+        self.assertIn("capability_ids_in_user_answer", detector_names)
+        self.assertIn("top_level_scaffold_before_answer", detector_names)
+        self.assertIn("limited_answer_without_next_action", detector_names)
+
+    def test_targeted_repair_candidates_do_not_add_broad_smoke_detectors(self) -> None:
+        candidates = dcl.build_detector_candidates(
+            {
+                "target_count": 1,
+                "acceptance_status": "partial",
+                "final_status": "partial",
+                "targets": [
+                    {
+                        "target_id": "pack:s01",
+                        "issue_code": "business_direct_answer_missing",
+                        "severity": "P0",
+                        "evidence_paths": ["artifacts/domain_runs/pack/steps/s01/output.md"],
+                    }
+                ],
+            }
+        )
+
+        detector_names = [item["detector"] for item in candidates["candidates"]]
+
+        self.assertIn("first_line_not_direct_answer", detector_names)
+        self.assertNotIn("runtime_tokens_in_user_answer", detector_names)
+
+    def test_detector_results_gate_allows_pass_and_blocks_fail_or_review(self) -> None:
+        pass_ok, pass_reason = dcl.evaluate_detector_results_gate({"status": "pass"})
+        fail_ok, fail_reason = dcl.evaluate_detector_results_gate(
+            {"status": "fail", "failed_detectors": ["runtime_tokens_in_user_answer"]}
+        )
+        review_ok, review_reason = dcl.evaluate_detector_results_gate(
+            {"status": "review", "review_detectors": ["required_contract_fields_missing"]}
+        )
+
+        self.assertTrue(pass_ok)
+        self.assertEqual(pass_reason, "detector_results_pass")
+        self.assertFalse(fail_ok)
+        self.assertEqual(fail_reason, "detector_results_failed:runtime_tokens_in_user_answer")
+        self.assertFalse(review_ok)
+        self.assertEqual(review_reason, "detector_results_review_required:required_contract_fields_missing")
 
     def test_auto_coder_gate_blocks_non_allowlisted_issue_codes(self) -> None:
         repair_targets = {