P0-R1.1-R2-S0-R1 Deployment Source Correction

core/generation_runner.py

@@ -7,6 +7,7 @@ connections. Checkpoints are task-scoped, immutable-by-sequence JSON snapshots.
 """
 from __future__ import annotations
 
+import copy
 import hashlib
 import inspect
 import json
@@ -912,24 +913,121 @@ class GenerationRunner:
     def _repair_validate_study_fact_contract(
         mode: str, agent, quant_engine, text: str, data_metadata: dict, *, section_name: str
     ):
+        """S0-R1 compatibility gateway: inspect only and never mutate prose."""
         value = str(text or '').strip()
         diagnostics = []
         if str(mode or '') == 'asq':
-            value, diagnostics = agent.repair_qualitative_study_fact_contract(
-                value, data_metadata, section_name=section_name
-            )
-            agent.validate_qualitative_study_fact_contract(
+            diagnostics = agent.inspect_qualitative_study_fact_contract(
                 value, data_metadata, section_name=section_name
             )
         elif str(mode or '') == 'amj':
-            value, diagnostics = quant_engine.repair_quantitative_study_fact_contract(
-                value, data_metadata, section_name=section_name
-            )
-            quant_engine.validate_quantitative_study_fact_contract(
+            diagnostics = quant_engine.inspect_quantitative_study_fact_contract(
                 value, data_metadata, section_name=section_name
             )
         return value, list(diagnostics or [])
 
+    @staticmethod
+    def _study_fact_violations(mode: str, agent, quant_engine, text: str, data_metadata: dict, *, section_name: str):
+        if str(mode or '') == 'asq':
+            return list(agent.inspect_qualitative_study_fact_contract(text, data_metadata, section_name=section_name) or [])
+        if str(mode or '') == 'amj':
+            return list(quant_engine.inspect_quantitative_study_fact_contract(text, data_metadata, section_name=section_name) or [])
+        return []
+
+    @staticmethod
+    def _section_retry_feedback(violations, *, limit=8) -> str:
+        rows = []
+        for item in list(violations or [])[:max(1, int(limit))]:
+            if not isinstance(item, dict):
+                rows.append(str(item)[:500])
+                continue
+            kind = str(item.get('type') or 'validation_issue')
+            detail = str(item.get('repair_instruction') or item.get('reason') or item.get('preview') or '')
+            actor = str(item.get('actor') or '')
+            expected = item.get('expected')
+            observed = item.get('observed')
+            parts = [kind]
+            if actor:
+                parts.append(f'actor={actor}')
+            if observed not in (None, ''):
+                parts.append(f'observed={observed}')
+            if expected not in (None, ''):
+                parts.append(f'expected={expected}')
+            if detail:
+                parts.append(detail[:360])
+            rows.append(' | '.join(parts))
+        return '\n'.join(f'- {row}' for row in rows)
+
+    @staticmethod
+    def _snapshot_section_citation_state(agent, citation_plan: dict, unit_name: str) -> dict:
+        """Snapshot only Section-owned citation state so a rejected attempt cannot leak forward."""
+        return {
+            "used_citations": set(getattr(agent, "used_citations", set()) or set()),
+            "citation_audit_log": copy.deepcopy(list(getattr(agent, "citation_audit_log", []) or [])),
+            "citation_map": copy.deepcopy(dict(getattr(agent, "citation_map", {}) or {})),
+            "planned_items": copy.deepcopy(list(dict(citation_plan or {}).get(unit_name, []) or [])),
+        }
+
+    @staticmethod
+    def _restore_section_citation_state(agent, citation_plan: dict, unit_name: str, snapshot: dict) -> None:
+        """Restore citation ownership after an invalid Section attempt without rewinding API counters."""
+        snap = dict(snapshot or {})
+        lock = getattr(agent, "lock", None)
+        if lock is None:
+            agent.used_citations = set(snap.get("used_citations") or set())
+            agent.citation_audit_log = copy.deepcopy(list(snap.get("citation_audit_log") or []))
+            agent.citation_map = copy.deepcopy(dict(snap.get("citation_map") or {}))
+        else:
+            with lock:
+                agent.used_citations = set(snap.get("used_citations") or set())
+                agent.citation_audit_log = copy.deepcopy(list(snap.get("citation_audit_log") or []))
+                agent.citation_map = copy.deepcopy(dict(snap.get("citation_map") or {}))
+        citation_plan[unit_name] = copy.deepcopy(list(snap.get("planned_items") or []))
+        setattr(agent, "_last_citation_semantic_violations", [])
+
+    @classmethod
+    def _section_validation_bundle(
+        cls, mode: str, agent, quant_engine, text: str, data_metadata: dict, *,
+        section_id: str, section_name: str, evidence_contract: str = "", include_critic: bool = True,
+        citation_violations=None,
+    ) -> list[dict]:
+        """Collect typed S0-R1 validation feedback without mutating manuscript prose."""
+        value = str(text or "")
+        violations = []
+        if include_critic:
+            verdict = agent.critique_section(
+                value, section_id=section_id, section_name=section_name, mode=mode,
+                evidence_contract=evidence_contract,
+            )
+            if isinstance(verdict, dict) and verdict.get("pass") is False:
+                violations.append({
+                    "type": "top_journal_critic",
+                    "reason": str(verdict.get("critical_issue") or "")[:800],
+                    "repair_instruction": str(verdict.get("repair_instruction") or "")[:1200],
+                })
+        violations.extend(list(agent.inspect_authorial_ai_meta(value) or []))
+        violations.extend(cls._study_fact_violations(
+            mode, agent, quant_engine, value, data_metadata, section_name=section_name
+        ))
+        violations.extend(copy.deepcopy(list(citation_violations or [])))
+        return violations
+
+    def _checkpoint_section_retry_exhausted(
+        self, state: dict, *, section_id: str, section_name: str, violations, pct: int, attempts: int = 2
+    ) -> None:
+        typed = copy.deepcopy(list(violations or []))
+        state["section_retry_exhausted"] = {
+            "section_id": str(section_id or ""),
+            "section_name": str(section_name or ""),
+            "attempts": max(1, int(attempts or 1)),
+            "violations": typed[:12],
+            "semantic_integrity_corruption": False,
+        }
+        self._checkpoint(
+            "section_retry_exhausted", state, pct,
+            display_node=f"section_retry_exhausted:{section_name}",
+        )
+
     @staticmethod
     def _section_integrity_hash(value: str) -> str:
         exact = str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip()
@@ -957,6 +1055,16 @@ class GenerationRunner:
         rendered = str(replacement or "").strip()
         return (value[:match.start()] + rendered + "\n\n" + value[match.end():].lstrip("\n")).strip() + "\n\n"
 
+    @staticmethod
+    def _remove_top_level_section(manuscript: str, title: str) -> str:
+        value = str(manuscript or "")
+        pattern = re.compile(rf'(?ms)^#\s+{re.escape(str(title))}\s*$\n.*?(?=^#\s+[^#].*?$|\Z)')
+        matches = list(pattern.finditer(value))
+        if len(matches) != 1:
+            raise RuntimeError(f"top_level_section_remove_count_invalid:{title}:{len(matches)}")
+        match = matches[0]
+        return (value[:match.start()] + value[match.end():].lstrip("\n")).strip() + "\n\n"
+
     @classmethod
     def _validate_section_integrity_seal(cls, seal: dict, section_text: str, section_title: str) -> None:
         record = dict(seal or {})
@@ -1862,13 +1970,14 @@ class GenerationRunner:
                 old_tokens = cls._paragraph_token_set(old)
                 union = tokens | old_tokens
                 similarity = len(tokens & old_tokens) / max(1, len(union))
-                if similarity < 0.90:
+                normalized_same = re.sub(r"\W+", " ", value).strip().casefold() == re.sub(r"\W+", " ", old).strip().casefold()
+                if not normalized_same:
                     continue
                 old_citations = set(re.findall(r'\[CITE-\d+\]', old, flags=re.I))
                 old_numerics = set(re.findall(r'(?<![A-Za-z])[-+]?\d+(?:\.\d+)?%?(?![A-Za-z])', old))
                 # Never delete a paragraph carrying different evidence/facts.
                 if citations == old_citations and numerics == old_numerics:
-                    duplicate = (similarity, old[:180])
+                    duplicate = (1.0, old[:180])
                     break
             if duplicate:
                 diagnostics.append({"similarity": round(duplicate[0], 4), "removed_preview": value[:180]})
@@ -2113,23 +2222,33 @@ class GenerationRunner:
             working_keywords = list(working_front.get("keywords") or [])
             if not working_abstract or not working_keywords:
                 raise RuntimeError("working_front_matter_missing")
-            working_abstract, front_fact_rows = self._repair_validate_study_fact_contract(
+            _, front_fact_rows = self._repair_validate_study_fact_contract(
                 mode, agent, quant_engine, working_abstract, data_metadata,
                 section_name="Working Abstract",
             )
-            if not working_abstract:
-                raise RuntimeError("working_front_matter_empty_after_study_fact_contract")
             if front_fact_rows:
+                feedback = self._section_retry_feedback(front_fact_rows)
                 logging.warning(
-                    "Working Abstract study-fact repair | mode=%s count=%s types=%s",
-                    mode, len(front_fact_rows),
-                    sorted({str(row.get("type") or "") for row in front_fact_rows}),
+                    "Working Abstract fact validation retry | mode=%s count=%s types=%s",
+                    mode, len(front_fact_rows), sorted({str(row.get("type") or "") for row in front_fact_rows}),
                 )
+                working_front = agent.generate_working_front_matter(
+                    topic, keywords, research_logic_contract, synthesis_result,
+                    data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
+                )
+                working_abstract = str(working_front.get("abstract") or "").strip()
+                working_keywords = list(working_front.get("keywords") or [])
+                _, remaining_front_rows = self._repair_validate_study_fact_contract(
+                    mode, agent, quant_engine, working_abstract, data_metadata, section_name="Working Abstract",
+                )
+                if remaining_front_rows:
+                    raise RuntimeError(f"working_front_matter_study_fact_invalid_after_retry:{remaining_front_rows[:3]}")
                 quality_diagnostics.append({
                     "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
-                    "status":"deterministic_repair", "count":len(front_fact_rows),
-                    "examples":front_fact_rows[:3],
+                    "status":"section_retry_resolved", "count":len(front_fact_rows), "examples":front_fact_rows[:3],
                 })
+            if not working_abstract:
+                raise RuntimeError("working_front_matter_empty_after_validation")
             agent.global_memory["working_abstract"] = working_abstract
 
             self._progress(14, "outline_planning")
@@ -2359,36 +2478,48 @@ class GenerationRunner:
         working_abstract = str(state.get("working_abstract") or "").strip()
         working_keywords = list(state.get("working_keywords") or [])
         if working_abstract and not state.get("english_finalized"):
-            checked_working_abstract, resumed_front_rows = self._repair_validate_study_fact_contract(
-                mode, agent, quant_engine, working_abstract, data_metadata,
-                section_name="Working Abstract",
-            )
-            if not checked_working_abstract:
-                raise RuntimeError("working_front_matter_empty_after_study_fact_contract")
-            if checked_working_abstract != working_abstract or resumed_front_rows:
-                working_abstract = checked_working_abstract
+            _, resumed_front_rows = self._repair_validate_study_fact_contract(
+                mode, agent, quant_engine, working_abstract, data_metadata, section_name="Working Abstract",
+            )
+            if resumed_front_rows:
+                feedback = self._section_retry_feedback(resumed_front_rows)
+                working_front = agent.generate_working_front_matter(
+                    topic, keywords, research_logic_contract, synthesis_result,
+                    data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
+                )
+                candidate_abstract = str(working_front.get("abstract") or "").strip()
+                candidate_keywords = list(working_front.get("keywords") or [])
+                _, remaining = self._repair_validate_study_fact_contract(
+                    mode, agent, quant_engine, candidate_abstract, data_metadata, section_name="Working Abstract",
+                )
+                if remaining:
+                    raise RuntimeError(f"working_front_matter_study_fact_invalid_after_retry:{remaining[:3]}")
+                working_abstract = candidate_abstract
+                working_keywords = candidate_keywords or working_keywords
                 state["working_abstract"] = working_abstract
+                state["working_keywords"] = working_keywords
                 if str(state.get("full_draft_en") or "").strip():
                     state["full_draft_en"] = replace_abstract(str(state.get("full_draft_en") or ""), working_abstract)
+                    state["full_draft_en"] = self._replace_keywords(state["full_draft_en"], working_keywords)
                     full_draft_en = str(state["full_draft_en"] or "")
                 quality_diagnostics.append({
                     "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
-                    "status":"resume_deterministic_repair", "count":len(resumed_front_rows),
-                    "examples":resumed_front_rows[:3],
+                    "status":"resume_section_retry_resolved", "count":len(resumed_front_rows), "examples":resumed_front_rows[:3],
                 })
                 state["quality_diagnostics"] = quality_diagnostics
-                self._checkpoint("working_front_matter_fact_reconciled", state, 15)
+                self._checkpoint("working_front_matter_fact_retry_resolved", state, 15)
         if working_abstract:
             agent.global_memory["working_abstract"] = working_abstract
         if self._preview_target_text == '':
             self._preview_target_text = str(state.get('full_draft_translated') or '')
 
-        # P0-R1.1-R2 compatibility migration: old checkpoints predate Section
-        # Integrity Seals. Reconcile only already-completed English sections with
-        # deterministic local contracts, seal them, and invalidate target cache
-        # only when the exact English section changes. Never call the writer.
+        # S0-R1 compatibility migration: old completed English is never silently
+        # repaired. Unsealed sections are inspected only. Valid text receives a
+        # seal; invalid text is removed from the completed set so only that owning
+        # top-level section is regenerated by the normal Section Writer.
         seal_migration = False
         seal_text_changed = False
+        retry_section_keys = set()
         for section_key, section_title in self._unique_top_level_sections(adaptive_structure):
             if not self._section_is_complete(adaptive_structure, completed_sections, section_key):
                 continue
@@ -2401,32 +2532,55 @@ class GenerationRunner:
                 and existing_seal.get("sha256") == self._section_integrity_hash(section_text)
             ):
                 continue
-            migrated_text, enum_rows = agent.repair_and_validate_enumeration_contract(
-                section_text, section_name=f"Resume sealed section: {section_title}"
-            )
-            migrated_text, fact_rows = self._repair_validate_study_fact_contract(
-                mode, agent, quant_engine, migrated_text, data_metadata,
+            violations = self._study_fact_violations(
+                mode, agent, quant_engine, section_text, data_metadata,
                 section_name=f"Resume sealed section: {section_title}",
             )
-            if migrated_text.strip() != section_text.strip():
-                full_draft_en = self._replace_top_level_section(
-                    full_draft_en, section_title, migrated_text
+            try:
+                agent.validate_enumeration_contract(
+                    section_text, section_name=f"Resume sealed section: {section_title}"
                 )
+            except Exception as exc:
+                violations.append({
+                    "type":"enumeration_contract_invalid",
+                    "reason":str(exc)[:500], "preview":section_title,
+                })
+            ai_meta_rows = list(agent.inspect_authorial_ai_meta(section_text) or [])
+            violations.extend(ai_meta_rows)
+            if violations:
+                retry_section_keys.add(str(section_key))
                 translation_sections.pop(section_key, None)
-                seal_text_changed = True
+                section_integrity_seals.pop(section_key, None)
                 quality_diagnostics.append({
                     "scope":"english_section_seal_migration", "name":section_title,
-                    "status":"deterministic_repair", "count":len(enum_rows) + len(fact_rows),
-                    "examples":(list(enum_rows or []) + list(fact_rows or []))[:3],
+                    "status":"local_section_retry_required", "count":len(violations),
+                    "examples":violations[:3],
                 })
-                section_text = self._extract_top_level_section(full_draft_en, section_title)
+                continue
             section_integrity_seals[section_key] = self._build_section_integrity_seal(
                 section_key, section_title, section_text
             )
             seal_migration = True
+
+        if retry_section_keys:
+            # Remove only invalid unsealed top-level sections and their completed
+            # unit markers. Other completed English stays byte-stable.
+            for section_key, section_title in self._unique_top_level_sections(adaptive_structure):
+                if str(section_key) not in retry_section_keys:
+                    continue
+                if self._extract_top_level_section(full_draft_en, section_title):
+                    full_draft_en = self._remove_top_level_section(full_draft_en, section_title)
+                    seal_text_changed = True
+                relevant_names = {
+                    str(task.get('display_name') or '')
+                    for task in adaptive_structure
+                    if str(task.get('section_id') or task.get('section') or '') == str(section_key)
+                }
+                completed_sections = [name for name in completed_sections if name not in relevant_names]
+            seal_migration = True
+
         if seal_migration:
-            # Rebuild mutable manuscript memory only from sealed English text so
-            # rejected/raw claims cannot propagate into later sections.
+            # Rebuild mutable manuscript memory only from still-completed sealed English text.
             for memory_key in ("hypotheses", "variables", "model_summary", "method_summary"):
                 agent.global_memory[memory_key] = ""
             last_sealed = ""
@@ -2440,6 +2594,7 @@ class GenerationRunner:
             if last_sealed:
                 context_tail = " ".join(last_sealed.split()[-200:])
             state.update({
+                "completed_sections": completed_sections,
                 "full_draft_en": full_draft_en,
                 "translation_sections": translation_sections,
                 "section_integrity_seals": section_integrity_seals,
@@ -2519,23 +2674,28 @@ class GenerationRunner:
             working_keywords = list(working_front.get("keywords") or [])
             if not working_abstract or not working_keywords:
                 raise RuntimeError("working_front_matter_missing")
-            working_abstract, front_fact_rows = self._repair_validate_study_fact_contract(
-                mode, agent, quant_engine, working_abstract, data_metadata,
-                section_name="Working Abstract",
+            _, front_fact_rows = self._repair_validate_study_fact_contract(
+                mode, agent, quant_engine, working_abstract, data_metadata, section_name="Working Abstract",
             )
-            if not working_abstract:
-                raise RuntimeError("working_front_matter_empty_after_study_fact_contract")
             if front_fact_rows:
-                logging.warning(
-                    "Working Abstract resume study-fact repair | mode=%s count=%s types=%s",
-                    mode, len(front_fact_rows),
-                    sorted({str(row.get("type") or "") for row in front_fact_rows}),
-                )
+                feedback = self._section_retry_feedback(front_fact_rows)
+                working_front = agent.generate_working_front_matter(
+                    topic, keywords, research_logic_contract, synthesis_result,
+                    data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
+                )
+                working_abstract = str(working_front.get("abstract") or "").strip()
+                working_keywords = list(working_front.get("keywords") or [])
+                _, remaining = self._repair_validate_study_fact_contract(
+                    mode, agent, quant_engine, working_abstract, data_metadata, section_name="Working Abstract",
+                )
+                if remaining:
+                    raise RuntimeError(f"working_front_matter_study_fact_invalid_after_retry:{remaining[:3]}")
                 quality_diagnostics.append({
                     "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
-                    "status":"deterministic_repair", "count":len(front_fact_rows),
-                    "examples":front_fact_rows[:3],
+                    "status":"section_retry_resolved", "count":len(front_fact_rows), "examples":front_fact_rows[:3],
                 })
+            if not working_abstract:
+                raise RuntimeError("working_front_matter_empty_after_validation")
             agent.global_memory["working_abstract"] = working_abstract
             full_draft_en = replace_abstract(full_draft_en, working_abstract)
             full_draft_en = self._replace_keywords(full_draft_en, working_keywords)
@@ -2817,175 +2977,291 @@ Output only final academic prose. Do not report the word count."""
                 prefix += f"# {sec_title}\n\n"
             if sec.get("subsection_first") and subsection:
                 prefix += f"## {subsection}\n\n"
-            if mode == "asq" and section_id == "appendix":
-                # R6A-R4: Appendix is a deterministic reader-facing projection
-                # of the Evidence Ledger. It is never regenerated by the LLM.
-                en_raw = qual_engine.build_appendix_evidence_md(data_metadata)
-            else:
-                en_raw = self._stream_text(
-                    agent.stream_deepseek(
-                        sys_prompt, user_prompt, max_tokens=7000,
-                        model_role="writing", thinking=False, allow_fallback=False,
-                    ),
-                    preview_prefix=full_draft_en + prefix,
-                    preview_stage=display_name,
-                )
-                en_raw = re.sub(r"<think>.*?</think>", "", en_raw, flags=re.S).strip()
-                en_raw = re.sub(rf"(?im)^#\s*{re.escape(sec_title)}\s*\n+", "", en_raw).strip()
-                if subsection:
-                    en_raw = re.sub(rf"(?im)^##\s*{re.escape(subsection)}\s*\n+", "", en_raw).strip()
-                en_raw = re.sub(r"(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)", r"\1 \2", en_raw)
-                en_raw = re.sub(r"(?m)^#{4,}\s+(.*)", r"**\1**", en_raw)
-                en_raw = re.sub(r"(?m)^#\s+(.+)$", r"## \1", en_raw)
-                en_raw = agent.critique_and_repair_section(
-                    en_raw, section_id=section_id, section_name=display_name, mode=mode,
+            prior_retry_exhausted = state.get("section_retry_exhausted")
+            if (
+                isinstance(prior_retry_exhausted, dict)
+                and str(prior_retry_exhausted.get("section_name") or "") == display_name
+            ):
+                # S0-R1 deliberately does not invent an executor/network retry state.
+                # A content-invalid Section that exhausted its one local retry remains
+                # durable and fail-closed until a separately approved lifecycle change.
+                raise RuntimeError(f"section_retry_exhausted_requires_new_lifecycle:{display_name}")
+
+            deterministic_appendix = mode == "asq" and section_id == "appendix"
+            max_section_attempts = 1 if deterministic_appendix else 2
+            frozen_citation_state = self._snapshot_section_citation_state(
+                agent, citation_plan, display_name
+            )
+            attempt_feedback = ""
+            en_resolved = ""
+            survived = set()
+
+            for section_attempt in range(1, max_section_attempts + 1):
+                # Every attempt starts from exactly the same evidence/citation packet.
+                # Rejected attempts may not leak used tags, JIT references, or audit
+                # decisions into the retry or any later Section.
+                if section_attempt > 1:
+                    self._restore_section_citation_state(
+                        agent, citation_plan, display_name, frozen_citation_state
+                    )
+                planned_items = copy.deepcopy(list(citation_plan.get(display_name, []) or []))
+                citation_plan[display_name] = planned_items
+                attempt_diagnostics = []
+                candidate_user_prompt = user_prompt
+                if attempt_feedback:
+                    candidate_user_prompt += f"""
+
+<SECTION_RETRY_VALIDATION_FEEDBACK>
+The previous attempt for this same manuscript unit was rejected by validation. Rewrite ONLY this same unit once. Keep the identical frozen evidence packet, research logic, section role, and citation plan. Do not fetch, invent, or broaden evidence. Resolve every material issue below without changing the task or manuscript scope:
+{attempt_feedback}
+</SECTION_RETRY_VALIDATION_FEEDBACK>"""
+                    if agent._count_tokens(sys_prompt) + agent._count_tokens(candidate_user_prompt) > 60000:
+                        candidate_user_prompt = agent._truncate_context(
+                            candidate_user_prompt, 60000 - agent._count_tokens(sys_prompt)
+                        )
+
+                if deterministic_appendix:
+                    # R6A-R4: Appendix is a deterministic reader-facing projection
+                    # of the Evidence Ledger. It is never regenerated by the LLM.
+                    en_raw = qual_engine.build_appendix_evidence_md(data_metadata)
+                else:
+                    en_raw = self._stream_text(
+                        agent.stream_deepseek(
+                            sys_prompt, candidate_user_prompt, max_tokens=7000,
+                            model_role="writing", thinking=False, allow_fallback=False,
+                        ),
+                        preview_prefix=full_draft_en + prefix,
+                        preview_stage=(display_name if section_attempt == 1 else f"{display_name}:retry"),
+                    )
+                    en_raw = re.sub(r"<think>.*?</think>", "", en_raw, flags=re.S).strip()
+                    en_raw = re.sub(rf"(?im)^#\s*{re.escape(sec_title)}\s*\n+", "", en_raw).strip()
+                    if subsection:
+                        en_raw = re.sub(rf"(?im)^##\s*{re.escape(subsection)}\s*\n+", "", en_raw).strip()
+                    en_raw = re.sub(r"(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)", r"\1 \2", en_raw)
+                    en_raw = re.sub(r"(?m)^#{4,}\s+(.*)", r"**\1**", en_raw)
+                    en_raw = re.sub(r"(?m)^#\s+(.+)$", r"## \1", en_raw)
+                    if mode == "asq":
+                        # Model prose has no authority over quote presence/count.
+                        en_raw = self._strip_model_controlled_qualitative_quotes(en_raw, data_metadata)
+
+                # S0-R1 phase 1: feedback-only critic + pure semantic validators.
+                pre_validation_rows = self._section_validation_bundle(
+                    mode, agent, quant_engine, prefix + en_raw, data_metadata,
+                    section_id=section_id, section_name=display_name,
                     evidence_contract=evidence_packet,
+                    include_critic=not deterministic_appendix,
+                )
+                if pre_validation_rows:
+                    if section_attempt < max_section_attempts:
+                        attempt_feedback = self._section_retry_feedback(pre_validation_rows)
+                        logging.warning(
+                            "Section validation retry | unit=%s attempt=%s count=%s types=%s",
+                            display_name, section_attempt, len(pre_validation_rows),
+                            sorted({str(row.get("type") or "") for row in pre_validation_rows if isinstance(row, dict)}),
+                        )
+                        self._restore_section_citation_state(
+                            agent, citation_plan, display_name, frozen_citation_state
+                        )
+                        continue
+                    self._restore_section_citation_state(
+                        agent, citation_plan, display_name, frozen_citation_state
+                    )
+                    state.update({
+                        "citation_plan": citation_plan,
+                        "agent_state": agent.export_state(),
+                        "quality_diagnostics": quality_diagnostics,
+                    })
+                    self._checkpoint_section_retry_exhausted(
+                        state, section_id=section_id, section_name=display_name,
+                        violations=pre_validation_rows, pct=pct, attempts=max_section_attempts,
+                    )
+                    raise RuntimeError(
+                        f"section_content_invalid_after_retry:{display_name}:{pre_validation_rows[:3]}"
+                    )
+
+                allow_deterministic_citation_reinforcement = not (
+                    mode == "asq" and section_id in {"methods", "discussion"}
                 )
-                en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False, mode=mode)
-                if mode == "asq":
-                    # Model prose has no authority over quote presence/count.
-                    en_raw = self._strip_model_controlled_qualitative_quotes(en_raw, data_metadata)
-            allow_deterministic_citation_reinforcement = not (
-                mode == "asq" and section_id in {"methods", "discussion"}
-            )
-            if not is_quarantined and assigned_tag_list and allow_deterministic_citation_reinforcement:
-                en_raw = agent.reinforce_claim_specific_citation_coverage(
-                    en_raw, assigned_tag_list, section_name=display_name
-                )
-            self._progress(pct, f"citation_audit:{current_index}:{total_steps}")
-            used_before = set(agent.used_citations)
-            en_resolved = agent.audit_and_resolve_citations(
-                prefix + en_raw, section_name=display_name, quarantine=is_quarantined
-            )
-            used_after = set(agent.used_citations)
-            survived = used_after - used_before
-            if assigned_tag_list and not is_quarantined:
-                required_survival = min(
-                    len(assigned_tag_list),
-                    max(1, int(math.ceil(len(assigned_tag_list) * CITATION_SEMANTIC_SURVIVAL_RATIO))),
-                )
-                if len(survived) < required_survival:
-                    missing_items = [
-                        item for item in planned_items
-                        if str(item.get("tag") or "") not in used_after
-                    ]
-                    needed = max(0, required_survival - len(survived))
-                    recovery_items = agent.select_contextual_citation_recovery_items(
-                        missing_items, section_name=display_name, section_context=en_resolved,
-                        needed=needed, excluded_tags=used_after,
+                if not is_quarantined and assigned_tag_list and allow_deterministic_citation_reinforcement:
+                    en_raw = agent.reinforce_claim_specific_citation_coverage(
+                        en_raw, assigned_tag_list, section_name=display_name
+                    )
+                self._progress(pct, f"citation_audit:{current_index}:{total_steps}")
+                used_before = set(agent.used_citations)
+                en_resolved = agent.audit_and_resolve_citations(
+                    prefix + en_raw, section_name=display_name, quarantine=is_quarantined
+                )
+                # Capture immediately: citation recovery may later replace the
+                # agent's last-violation buffer. These violations belong to the
+                # writer attempt and require the one bounded same-Section retry.
+                citation_semantic_rows = copy.deepcopy(
+                    list(getattr(agent, "_last_citation_semantic_violations", []) or [])
+                )
+                if citation_semantic_rows:
+                    if section_attempt < max_section_attempts:
+                        attempt_feedback = self._section_retry_feedback(citation_semantic_rows)
+                        logging.warning(
+                            "Citation semantic Section retry | unit=%s attempt=%s count=%s types=%s",
+                            display_name, section_attempt, len(citation_semantic_rows),
+                            sorted({str(row.get("type") or "") for row in citation_semantic_rows if isinstance(row, dict)}),
+                        )
+                        self._restore_section_citation_state(
+                            agent, citation_plan, display_name, frozen_citation_state
+                        )
+                        continue
+                    self._restore_section_citation_state(
+                        agent, citation_plan, display_name, frozen_citation_state
+                    )
+                    state.update({
+                        "citation_plan": citation_plan,
+                        "agent_state": agent.export_state(),
+                        "quality_diagnostics": quality_diagnostics,
+                    })
+                    self._checkpoint_section_retry_exhausted(
+                        state, section_id=section_id, section_name=display_name,
+                        violations=citation_semantic_rows, pct=pct, attempts=max_section_attempts,
+                    )
+                    raise RuntimeError(
+                        f"section_citation_invalid_after_retry:{display_name}:{citation_semantic_rows[:3]}"
+                    )
+
+                used_after = set(agent.used_citations)
+                survived = used_after - used_before
+                if assigned_tag_list and not is_quarantined:
+                    required_survival = min(
+                        len(assigned_tag_list),
+                        max(1, int(math.ceil(len(assigned_tag_list) * CITATION_SEMANTIC_SURVIVAL_RATIO))),
                     )
-                    if len(recovery_items) < needed:
-                        recovery_items.extend(
-                            agent.fetch_contextual_jit_recovery_items(
-                                topic=topic, section_name=display_name, section_context=en_resolved,
-                                needed=needed - len(recovery_items),
-                                excluded_tags=used_after | {str(item.get("tag") or "") for item in recovery_items},
+                    if len(survived) < required_survival:
+                        missing_items = [
+                            item for item in planned_items
+                            if str(item.get("tag") or "") not in used_after
+                        ]
+                        needed = max(0, required_survival - len(survived))
+                        recovery_items = agent.select_contextual_citation_recovery_items(
+                            missing_items, section_name=display_name, section_context=en_resolved,
+                            needed=needed, excluded_tags=used_after,
+                        )
+                        # Evidence retrieval is permitted only on the first writer
+                        # attempt. Attempt 2 is bound to the original frozen packet.
+                        if section_attempt == 1 and len(recovery_items) < needed:
+                            recovery_items.extend(
+                                agent.fetch_contextual_jit_recovery_items(
+                                    topic=topic, section_name=display_name, section_context=en_resolved,
+                                    needed=needed - len(recovery_items),
+                                    excluded_tags=used_after | {str(item.get("tag") or "") for item in recovery_items},
+                                )
                             )
+                        planned_tags_for_unit = {str(item.get("tag") or "") for item in planned_items}
+                        for recovery_item in recovery_items[:needed]:
+                            tag = str(recovery_item.get("tag") or "")
+                            if tag and tag not in planned_tags_for_unit:
+                                planned_items.append(recovery_item)
+                                planned_tags_for_unit.add(tag)
+                        citation_plan[display_name] = planned_items
+                        # Recovery may attach evidence only to an already-written
+                        # claim. It never creates supplemental prose.
+                        self._progress(pct, f"citation_recovery:{current_index}:{total_steps}")
+                        recovered = agent.recover_citations_on_existing_claims(
+                            en_resolved,
+                            recovery_items[:needed],
+                            section_name=display_name,
+                            needed=needed,
                         )
-                    planned_tags_for_unit = {str(item.get("tag") or "") for item in planned_items}
-                    for recovery_item in recovery_items[:needed]:
-                        tag = str(recovery_item.get("tag") or "")
-                        if tag and tag not in planned_tags_for_unit:
-                            planned_items.append(recovery_item)
-                            planned_tags_for_unit.add(tag)
-                    citation_plan[display_name] = planned_items
-                    # R14-TQ2-PQ1-R1: recovery may attach evidence only to
-                    # an already-written claim. It is never authorized to create a
-                    # supplemental paragraph from a source merely to satisfy citation
-                    # survival. If no existing claim can be validated, leave prose
-                    # unchanged. Evidence coverage may later be reported as limited,
-                    # while citation correctness itself remains fail-closed.
-                    self._progress(pct, f"citation_recovery:{current_index}:{total_steps}")
-                    recovered = agent.recover_citations_on_existing_claims(
+                        if recovered != en_resolved:
+                            en_resolved = recovered
+                        used_after = set(agent.used_citations)
+                        survived = used_after - used_before
+
+                if mode == "asq" and section_id == "findings":
+                    en_resolved = self._render_planned_qualitative_quotes(
                         en_resolved,
-                        recovery_items[:needed],
-                        section_name=display_name,
-                        needed=needed,
+                        task=sec,
+                        data_metadata=data_metadata,
+                        quote_invocation_plan=quote_invocation_plan,
                     )
-                    if recovered != en_resolved:
-                        en_resolved = recovered
-                    used_after = set(agent.used_citations)
-                    survived = used_after - used_before
-            if mode == "asq" and section_id == "findings":
-                en_resolved = self._render_planned_qualitative_quotes(
-                    en_resolved,
-                    task=sec,
-                    data_metadata=data_metadata,
-                    quote_invocation_plan=quote_invocation_plan,
-                )
-
-            # P0-R1.1-R2: every mutating deterministic operation must complete
-            # before the Evidence Fact contract seals the English unit.
-            en_resolved, enumeration_rows = agent.repair_and_validate_enumeration_contract(
-                en_resolved, section_name=display_name
-            )
-            if enumeration_rows:
-                logging.warning(
-                    "Structural enumeration repaired | unit=%s count=%s details=%s",
-                    display_name, len(enumeration_rows), enumeration_rows[:3],
-                )
-                quality_diagnostics.append({
-                    "scope":"structural_enumeration", "name":display_name,
-                    "status":"deterministic_repair", "count":len(enumeration_rows),
-                    "examples":enumeration_rows[:3],
-                })
-            en_resolved, duplicate_rows = self._remove_severe_cross_unit_duplicates(en_resolved, full_draft_en)
-            if duplicate_rows:
-                logging.warning("Severe cross-unit redundancy removed | unit=%s count=%s max_similarity=%s", display_name, len(duplicate_rows), max(row["similarity"] for row in duplicate_rows))
-                quality_diagnostics.append({"scope":"manuscript_redundancy","name":display_name,"status":"bounded_duplicate_removed","count":len(duplicate_rows),"examples":duplicate_rows[:3]})
-            en_resolved, sentence_echo_rows = agent.remove_severe_cross_unit_sentence_echoes(
-                en_resolved, full_draft_en
-            )
-            if sentence_echo_rows:
-                logging.warning(
-                    "Severe cross-unit sentence echo removed | unit=%s count=%s max_similarity=%s",
-                    display_name, len(sentence_echo_rows),
-                    max(row["similarity"] for row in sentence_echo_rows),
-                )
-                quality_diagnostics.append({
-                    "scope":"manuscript_sentence_echo", "name":display_name,
-                    "status":"bounded_duplicate_removed", "count":len(sentence_echo_rows),
-                    "examples":sentence_echo_rows[:3],
-                })
 
-            # Evidence Fact repair/validation is intentionally LAST. Once this
-            # passes, no later operation may mutate this English unit.
-            if mode == "asq":
-                en_resolved, study_fact_rows = agent.repair_qualitative_study_fact_contract(
-                    en_resolved, data_metadata, section_name=display_name
+                # S0-R1 phase 2: only local/idempotent structural normalizers may
+                # mutate the candidate before the final pure validation bundle.
+                en_resolved, enumeration_rows = agent.repair_and_validate_enumeration_contract(
+                    en_resolved, section_name=display_name
                 )
-                if study_fact_rows:
+                if enumeration_rows:
                     logging.warning(
-                        "Qualitative Study Evidence Ledger repair | unit=%s count=%s types=%s",
-                        display_name, len(study_fact_rows),
-                        sorted({str(row.get("type") or "") for row in study_fact_rows}),
+                        "Structural enumeration repaired | unit=%s count=%s details=%s",
+                        display_name, len(enumeration_rows), enumeration_rows[:3],
                     )
-                    quality_diagnostics.append({
-                        "scope":"qualitative_study_fact_integrity", "name":display_name,
-                        "status":"deterministic_repair", "count":len(study_fact_rows),
-                        "examples":study_fact_rows[:3],
+                    attempt_diagnostics.append({
+                        "scope":"structural_enumeration", "name":display_name,
+                        "status":"deterministic_repair", "count":len(enumeration_rows),
+                        "examples":enumeration_rows[:3],
                     })
-                agent.validate_qualitative_study_fact_contract(
-                    en_resolved, data_metadata, section_name=display_name
-                )
-            elif mode == "amj":
-                en_resolved, study_fact_rows = quant_engine.repair_quantitative_study_fact_contract(
-                    en_resolved, data_metadata, section_name=display_name
+                en_resolved, duplicate_rows = self._remove_severe_cross_unit_duplicates(
+                    en_resolved, full_draft_en
                 )
-                if study_fact_rows:
+                if duplicate_rows:
                     logging.warning(
-                        "Quantitative Evidence Ledger repair | unit=%s count=%s types=%s",
-                        display_name, len(study_fact_rows),
-                        sorted({str(row.get("type") or "") for row in study_fact_rows}),
+                        "Exact cross-unit redundancy removed | unit=%s count=%s",
+                        display_name, len(duplicate_rows),
                     )
-                    quality_diagnostics.append({
-                        "scope":"quantitative_study_fact_integrity", "name":display_name,
-                        "status":"deterministic_repair", "count":len(study_fact_rows),
-                        "examples":study_fact_rows[:3],
+                    attempt_diagnostics.append({
+                        "scope":"manuscript_redundancy", "name":display_name,
+                        "status":"exact_duplicate_removed", "count":len(duplicate_rows),
+                        "examples":duplicate_rows[:3],
                     })
-                quant_engine.validate_quantitative_study_fact_contract(
-                    en_resolved, data_metadata, section_name=display_name
+                en_resolved, sentence_echo_rows = agent.remove_severe_cross_unit_sentence_echoes(
+                    en_resolved, full_draft_en
                 )
+                if sentence_echo_rows:
+                    logging.warning(
+                        "Exact cross-unit sentence echo removed | unit=%s count=%s",
+                        display_name, len(sentence_echo_rows),
+                    )
+                    attempt_diagnostics.append({
+                        "scope":"manuscript_sentence_echo", "name":display_name,
+                        "status":"exact_duplicate_removed", "count":len(sentence_echo_rows),
+                        "examples":sentence_echo_rows[:3],
+                    })
+
+                # Final Section validation is pure. Once it passes, the exact bytes
+                # are eligible for the Section Integrity Seal and may never be
+                # semantically rewritten by a later validator.
+                post_validation_rows = self._section_validation_bundle(
+                    mode, agent, quant_engine, en_resolved, data_metadata,
+                    section_id=section_id, section_name=display_name,
+                    evidence_contract=evidence_packet, include_critic=False,
+                )
+                if post_validation_rows:
+                    if section_attempt < max_section_attempts:
+                        attempt_feedback = self._section_retry_feedback(post_validation_rows)
+                        logging.warning(
+                            "Post-normalization Section retry | unit=%s attempt=%s count=%s types=%s",
+                            display_name, section_attempt, len(post_validation_rows),
+                            sorted({str(row.get("type") or "") for row in post_validation_rows if isinstance(row, dict)}),
+                        )
+                        self._restore_section_citation_state(
+                            agent, citation_plan, display_name, frozen_citation_state
+                        )
+                        continue
+                    self._restore_section_citation_state(
+                        agent, citation_plan, display_name, frozen_citation_state
+                    )
+                    state.update({
+                        "citation_plan": citation_plan,
+                        "agent_state": agent.export_state(),
+                        "quality_diagnostics": quality_diagnostics,
+                    })
+                    self._checkpoint_section_retry_exhausted(
+                        state, section_id=section_id, section_name=display_name,
+                        violations=post_validation_rows, pct=pct, attempts=max_section_attempts,
+                    )
+                    raise RuntimeError(
+                        f"section_content_invalid_after_retry:{display_name}:{post_validation_rows[:3]}"
+                    )
+
+                quality_diagnostics.extend(attempt_diagnostics)
+                state.pop("section_retry_exhausted", None)
+                break
 
             # Recompute citation survival/claim demand from the exact text that
             # will be sealed, not from an earlier pre-cleanup intermediate.
@@ -3216,23 +3492,52 @@ Output only final academic prose. Do not report the word count."""
             final_keywords = list(final_front.get("keywords") or [])
             if not final_abstract or not final_keywords:
                 raise RuntimeError("final_front_matter_missing")
-            final_abstract, final_front_fact_rows = self._repair_validate_study_fact_contract(
+            final_front_fact_rows = self._study_fact_violations(
                 mode, agent, quant_engine, final_abstract, data_metadata,
                 section_name="Final Abstract",
             )
-            if not final_abstract:
-                raise RuntimeError("final_abstract_empty_after_study_fact_contract")
             if final_front_fact_rows:
+                feedback = self._section_retry_feedback(final_front_fact_rows)
                 logging.warning(
-                    "Final Abstract study-fact repair | mode=%s count=%s types=%s",
+                    "Final Abstract pure-validation retry | mode=%s count=%s types=%s",
                     mode, len(final_front_fact_rows),
-                    sorted({str(row.get("type") or "") for row in final_front_fact_rows}),
-                )
+                    sorted({str(row.get("type") or "") for row in final_front_fact_rows if isinstance(row, dict)}),
+                )
+                final_front = agent.generate_final_front_matter(
+                    topic, keywords, body_only, data_evidence=user_data_md, mode=mode,
+                    validation_feedback=feedback,
+                )
+                final_abstract = str(final_front.get("abstract") or "").strip()
+                final_keywords = list(final_front.get("keywords") or [])
+                if not final_abstract or not final_keywords:
+                    raise RuntimeError("final_front_matter_missing_after_retry")
+                remaining_final_fact_rows = self._study_fact_violations(
+                    mode, agent, quant_engine, final_abstract, data_metadata,
+                    section_name="Final Abstract",
+                )
+                if remaining_final_fact_rows:
+                    state["final_abstract_retry_exhausted"] = {
+                        "attempts": 2,
+                        "violations": copy.deepcopy(remaining_final_fact_rows[:12]),
+                        "semantic_integrity_corruption": False,
+                    }
+                    state.update({
+                        "agent_state": agent.export_state(),
+                        "quality_diagnostics": quality_diagnostics,
+                    })
+                    self._checkpoint(
+                        "final_abstract_retry_exhausted", state, 92,
+                        display_node="final_abstract_retry_exhausted",
+                    )
+                    raise RuntimeError(
+                        f"final_abstract_invalid_after_retry:{remaining_final_fact_rows[:3]}"
+                    )
                 quality_diagnostics.append({
                     "scope":"final_abstract_study_fact_integrity", "name":"Final Abstract",
-                    "status":"deterministic_repair", "count":len(final_front_fact_rows),
+                    "status":"front_matter_retry_resolved", "count":len(final_front_fact_rows),
                     "examples":final_front_fact_rows[:3],
                 })
+            state.pop("final_abstract_retry_exhausted", None)
             full_draft_en = replace_abstract(full_draft_en, final_abstract)
             full_draft_en = self._replace_keywords(full_draft_en, final_keywords)
             body_only, _reference_tail = agent.validate_reference_tail(

core/pipeline_agent.py

@@ -1468,7 +1468,7 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         raise DeepSeekRequestError(f'{phase} front matter validation failed: {last_error}')
 
     def generate_working_front_matter(
-        self, topic, keywords, research_logic_contract, synthesis_result, data_evidence='', mode='amj'
+        self, topic, keywords, research_logic_contract, synthesis_result, data_evidence='', mode='amj', validation_feedback=''
     ):
         """Create a provisional visible abstract/keyword anchor without prewriting results."""
         evidence = self._truncate_context(str(data_evidence or ''), 3500)
@@ -1488,7 +1488,10 @@ USER KEYWORD SEEDS: {keywords}
 </LITERATURE_SYNTHESIS>
 <FROZEN_RESEARCH_EVIDENCE>
 {evidence}
-</FROZEN_RESEARCH_EVIDENCE>"""
+</FROZEN_RESEARCH_EVIDENCE>
+<LOCAL_VALIDATION_FEEDBACK>
+{str(validation_feedback or '')[:2500]}
+</LOCAL_VALIDATION_FEEDBACK>"""
         def validate_working(parsed):
             abstract = self._validate_working_abstract(parsed.get('abstract'))
             working_keywords = self._front_matter_keywords(parsed.get('keywords'))
@@ -1499,7 +1502,7 @@ USER KEYWORD SEEDS: {keywords}
             prompt, phase='Working', max_tokens=1100, validator=validate_working,
         )
 
-    def generate_final_front_matter(self, topic, keywords, manuscript, data_evidence='', mode='amj'):
+    def generate_final_front_matter(self, topic, keywords, manuscript, data_evidence='', mode='amj', validation_feedback=''):
         """Generate final Abstract + Keywords only from the frozen completed body/evidence."""
         evidence = self._truncate_context(str(data_evidence or ''), 5000)
         body = self._abstract_manuscript_context(manuscript, 18000)
@@ -1514,11 +1517,17 @@ USER KEYWORD SEEDS: {keywords}
 </FROZEN_RESEARCH_EVIDENCE>
 <COMPLETED_FROZEN_MANUSCRIPT>
 {body}
-</COMPLETED_FROZEN_MANUSCRIPT>"""
+</COMPLETED_FROZEN_MANUSCRIPT>
+<LOCAL_VALIDATION_FEEDBACK>
+{str(validation_feedback or '')[:2500]}
+</LOCAL_VALIDATION_FEEDBACK>"""
         def validate_final(parsed):
             abstract = re.sub(r'(?im)^#+\s*abstract\s*', '', str(parsed.get('abstract') or '')).strip()
             if self._front_matter_has_citation(abstract):
                 raise DeepSeekRequestError('Final Abstract must not contain citations.')
+            meta_findings = self.inspect_authorial_ai_meta(abstract)
+            if meta_findings:
+                raise DeepSeekRequestError('Final Abstract contains high-confidence authorial AI-writing meta-commentary.')
             final_keywords = self._front_matter_keywords(parsed.get('keywords'))
             return {'abstract': abstract, 'keywords': final_keywords}
 
@@ -1527,9 +1536,8 @@ USER KEYWORD SEEDS: {keywords}
             prompt, phase='Final', max_tokens=1200, validator=validate_final,
         )
         abstract = structured['abstract']
-        abstract = self.polish_academic_section(abstract, 'Abstract', abstract=True, mode=mode)
         if self._front_matter_has_citation(abstract):
-            raise DeepSeekRequestError('Final Abstract revision introduced a citation.')
+            raise DeepSeekRequestError('Final Abstract contains a citation after structured validation.')
         max_words = max(80, int(APP_CONFIG.get('manuscript_quality', {}).get('abstract_max_words', 200)))
         if not abstract:
             raise DeepSeekRequestError('Final Abstract is empty after validation.')
@@ -3667,6 +3675,115 @@ INPUT:
         )
         return bool(strong.search(value))
 
+    @classmethod
+    def inspect_qualitative_study_fact_contract(cls, text, metadata, section_name=""):
+        """Pure ASQ Study Evidence inspection. Never mutates manuscript prose.
+
+        S0-R1 deliberately prefers false negatives to destructive false positives.
+        Actor/source violations are reported only when the sentence contains an
+        explicit grammatical relation (for example ``interviews with farmers`` or
+        ``farmers were interviewed``). Mere proximity is not evidence ownership.
+        External literature and participant quotations are excluded.
+        """
+        value = str(text or "")
+        facts = dict((metadata or {}).get("study_facts") or {})
+        relations = [row for row in (facts.get("data_source_roles") or []) if isinstance(row, dict)]
+        allowed_by_actor = {}
+        for row in relations:
+            role = re.sub(r"\s+", " ", str(row.get("actor_role") or "")).strip()
+            family = cls._qual_source_family(row.get("data_source"))
+            if not role or not family:
+                continue
+            for variant in cls._qual_actor_variants(role):
+                allowed_by_actor.setdefault(variant, set()).add(family)
+
+        procedure_lists = {
+            "trustworthiness": [str(x or "").strip().casefold() for x in (facts.get("trustworthiness_procedures") or []) if str(x or "").strip()],
+            "ethics": [str(x or "").strip().casefold() for x in (facts.get("ethics_procedures") or []) if str(x or "").strip()],
+            "reflexivity": [str(x or "").strip().casefold() for x in (facts.get("reflexivity_procedures") or []) if str(x or "").strip()],
+        }
+        procedure_patterns = {
+            "trustworthiness": r"\bmember\s+check(?:ing|s)?\b|\bpeer\s+debrief(?:ing|ed)?\b|\baudit\s+trail\b|\bexternal\s+audit\b",
+            "ethics": r"\binformed\s+consent\b|\bethics?\s+approval\b|\binstitutional\s+review\s+board\b|\bIRB\b|\bapproval\s+(?:number|code)\b",
+            "reflexivity": r"\bfield\s+journal\b|\breflexive\s+journal\b|\bpositionality\s+(?:statement|memo)\b|\breflexive\s+memo(?:s)?\b",
+        }
+        source_patterns = {
+            "interview": r"(?:semi[- ]structured\s+)?interviews?|interviewed",
+            "focus_group": r"focus\s+groups?",
+            "observation": r"(?:field\s+)?observations?|observed",
+            "log": r"(?:usage|interaction|system)\s+logs?|log\s+data",
+            "document": r"documents?|archival\s+(?:records?|materials?)|records?",
+        }
+        source_union = "(?:" + "|".join(source_patterns.values()) + ")"
+
+        def procedure_key(raw):
+            observed = str(raw or "").casefold()
+            if "member check" in observed: return "member check"
+            if "peer debrief" in observed: return "peer debrief"
+            if "audit trail" in observed: return "audit trail"
+            if "external audit" in observed: return "external audit"
+            if "informed consent" in observed: return "informed consent"
+            if "ethic" in observed and "approval" in observed: return "ethics approval"
+            if "review board" in observed or observed.strip() == "irb": return "review board"
+            if "approval" in observed: return "approval"
+            if "field journal" in observed: return "field journal"
+            if "reflexive journal" in observed: return "reflexive journal"
+            if "positionality" in observed: return "positionality"
+            return "reflexive memo"
+
+        violations = []
+        seen = set()
+        for _start, _end, sentence in cls._sentence_spans(value):
+            core = str(sentence or "")
+            if not core.strip() or not cls._is_study_owned_qualitative_claim(core, section_name):
+                continue
+
+            # Check every explicit procedure mention; do not stop after the first.
+            for category, pattern in procedure_patterns.items():
+                for match in re.finditer(pattern, core, flags=re.I):
+                    key = procedure_key(match.group(0))
+                    authorized = any(key in item for item in procedure_lists[category])
+                    if authorized:
+                        continue
+                    sig = ("procedure", category, key, match.start())
+                    if sig in seen:
+                        continue
+                    seen.add(sig)
+                    violations.append({
+                        "type":"unsupported_study_procedure", "category":category,
+                        "procedure":key, "preview":core.strip()[:180],
+                    })
+
+            # Explicit actor/source ownership only. These patterns intentionally
+            # do not use nearest-distance heuristics.
+            for actor_variant, allowed_families in allowed_by_actor.items():
+                actor = re.escape(actor_variant)
+                patterns = [
+                    rf"(?P<source>{source_union})\s+(?:with|of|from|among|involving)\s+(?:the\s+)?(?P<actor>{actor})(?![A-Za-z])",
+                    rf"(?P<source>interviewed|observed)\s+(?:the\s+)?(?P<actor>{actor})(?![A-Za-z])",
+                    rf"(?P<actor>(?<![A-Za-z]){actor}(?![A-Za-z]))\s+(?:were|was|are|is)\s+(?P<source>interviewed|observed)\b",
+                    rf"(?P<actor>(?<![A-Za-z]){actor}(?![A-Za-z]))\s+(?:participated|took\s+part)\s+in\s+(?P<source>{source_union})",
+                    rf"(?P<actor>(?<![A-Za-z]){actor}(?![A-Za-z]))\s+(?:provided|submitted|generated)\s+(?P<source>{source_union})",
+                    rf"(?P<source>interview|focus\s+group|observation)\s+participants?[^.!?\n]{{0,120}}(?P<actor>(?<![A-Za-z]){actor}(?![A-Za-z]))",
+                ]
+                for pattern in patterns:
+                    for match in re.finditer(pattern, core, flags=re.I):
+                        raw_source = match.group("source")
+                        family = cls._qual_source_family(raw_source)
+                        if not family or family in allowed_families:
+                            continue
+                        sig = ("relation", actor_variant, family, match.start(), match.end())
+                        if sig in seen:
+                            continue
+                        seen.add(sig)
+                        violations.append({
+                            "type":"actor_data_source_relation_mismatch",
+                            "actor":actor_variant, "observed":family,
+                            "allowed":sorted(allowed_families),
+                            "preview":core.strip()[:180],
+                        })
+        return violations
+
     @classmethod
     def repair_qualitative_study_fact_contract(cls, text, metadata, section_name=""):
         """Repair only high-confidence *study-owned* qualitative fact violations.
@@ -3842,9 +3959,9 @@ INPUT:
 
     @classmethod
     def validate_qualitative_study_fact_contract(cls, text, metadata, section_name=""):
-        repaired, diagnostics = cls.repair_qualitative_study_fact_contract(text, metadata, section_name=section_name)
-        if repaired.strip() != str(text or "").strip():
-            raise RuntimeError(f"qualitative_study_fact_contract_unrepaired:{section_name}:{diagnostics[:3]}")
+        violations = cls.inspect_qualitative_study_fact_contract(text, metadata, section_name=section_name)
+        if violations:
+            raise RuntimeError(f"qualitative_study_fact_contract_invalid:{section_name}:{violations[:3]}")
         return True
 
     @classmethod
@@ -3954,8 +4071,8 @@ INPUT:
                 union = tokens | old_tokens
                 similarity = len(tokens & old_tokens) / max(1, len(union))
                 normalized_same = re.sub(r"\W+", " ", core).strip().casefold() == re.sub(r"\W+", " ", old).strip().casefold()
-                if normalized_same or similarity >= 0.96:
-                    duplicate = (1.0 if normalized_same else similarity, old[:180])
+                if normalized_same:
+                    duplicate = (1.0, old[:180])
                     break
             if duplicate:
                 replacements.append((start, end, ""))
@@ -3970,19 +4087,36 @@ INPUT:
             value = re.sub(r" {2,}", " ", value)
         return value.strip(), diagnostics
 
+    @classmethod
+    def _citation_tag_only_rewrite(cls, original, selected_tag=None):
+        """Change citation markup only; preserve academic claim wording."""
+        source = str(original or "")
+        leading = source[: len(source) - len(source.lstrip())]
+        trailing = source[len(source.rstrip()):]
+        core = source.strip()
+        core = re.sub(r"\s*[\[(]?CITE-\d+[\])]?\s*", " ", core, flags=re.I)
+        core = re.sub(r"[ \t]{2,}", " ", core).strip()
+        if selected_tag:
+            tag = str(selected_tag).strip()
+            punctuation = core[-1] if core and core[-1] in '.!?。！？' else ''
+            if punctuation:
+                core = core[:-1].rstrip()
+            core = f"{core} {tag}{punctuation}".strip()
+        return f"{leading}{core}{trailing}"
+
     def semantic_audit_claim_specific_citations(
         self,
         text,
         section_name="",
     ):
-        """Audit cited sentences against retrieved evidence at sentence scope.
+        """Audit citation support while granting the audit TAG-ONLY mutation authority.
 
-        The audit may split a sentence when different sources support different
-        claims, or keep one source when several sources support the same claim.
-        It may not invent facts, numbers, quotations, or citation tags. A cited
-        claim that cannot be supported is removed from the manuscript and logged;
-        one unsupported sentence must not discard an otherwise valid paper.
+        S0-R1 forbids citation QA from rewriting or deleting academic claims.
+        Unsupported or split-required claims are returned as typed section
+        violations for the owning Section Writer retry; supported decisions may
+        only select/remove citation tags.
         """
+        self._last_citation_semantic_violations = []
         if not text:
             return text
 
@@ -4158,83 +4292,62 @@ Return exactly:
                 status = str(item.get("status", "")).strip().lower()
                 revised = str(item.get("revised_text", "")).strip()
                 if status not in {
-                    "ok",
-                    "split",
-                    "uncited_common_knowledge",
-                    "unsupported",
+                    "ok", "split", "uncited_common_knowledge", "unsupported",
                 }:
                     raise DeepSeekRequestError(
                         f"Invalid citation audit status for {record['id']}: {status!r}"
                     )
-                if status == "unsupported":
-                    reason = str(item.get("reason", ""))[:500]
-                    unsupported.append(
-                        {
-                            "id": record["id"],
-                            "sentence": record["original"][:500],
-                            "reason": reason,
-                        }
-                    )
-                    # An unsupported externally verifiable claim cannot remain in
-                    # the manuscript with an arbitrary citation. Remove only this
-                    # sentence and retain an auditable record of the decision.
-                    replacements[record["id"]] = ""
-                    self._append_citation_audit(
-                        section_name,
-                        "remove_unsupported_claim",
-                        record["original"],
-                        "",
-                        reason or "No candidate evidence directly supported the claim.",
-                    )
-                    continue
-                if not revised:
-                    raise DeepSeekRequestError(
-                        f"Citation semantic audit returned empty text for {record['id']}"
-                    )
                 revised_tags = {
                     f"[{tag.upper()}]"
                     for tag in re.findall(r"CITE-\d+", revised, flags=re.IGNORECASE)
                 }
-                if status in {"ok", "split"} and not revised_tags:
-                    raise DeepSeekRequestError(
-                        f"Citation semantic audit removed all citation tags for {record['id']} "
-                        f"while status={status}."
-                    )
                 disallowed = revised_tags - set(record["allowed_tags"])
                 if disallowed:
                     raise DeepSeekRequestError(
                         f"Citation semantic audit invented tag(s) for {record['id']}: "
                         f"{sorted(disallowed)}"
                     )
-                if self._numeric_tokens(record["original"]) != self._numeric_tokens(revised):
-                    raise DeepSeekRequestError(
-                        f"Citation semantic audit changed numerical content for {record['id']}"
+                reason = str(item.get("reason", ""))[:500]
+                if status == "unsupported":
+                    violation = {
+                        "type":"citation_unsupported_claim", "id":record["id"],
+                        "sentence":record["original"][:500], "reason":reason,
+                    }
+                    unsupported.append(violation)
+                    self._last_citation_semantic_violations.append(violation)
+                    replacements[record["id"]] = self._citation_tag_only_rewrite(record["original"], None)
+                    self._append_citation_audit(
+                        section_name, "flag_unsupported_claim_for_section_retry",
+                        record["original"], replacements[record["id"]],
+                        reason or "No candidate evidence directly supported the claim.",
                     )
-                missing_quotes = [
-                    quote for quote in self._quoted_passages(record["original"])
-                    if quote not in revised
-                ]
-                if missing_quotes:
-                    raise DeepSeekRequestError(
-                        f"Citation semantic audit changed quoted evidence for {record['id']}"
+                    continue
+                if status == "split":
+                    violation = {
+                        "type":"citation_claim_split_required", "id":record["id"],
+                        "sentence":record["original"][:500], "reason":reason,
+                    }
+                    self._last_citation_semantic_violations.append(violation)
+                    replacements[record["id"]] = self._citation_tag_only_rewrite(record["original"], None)
+                    self._append_citation_audit(
+                        section_name, "flag_claim_split_for_section_retry",
+                        record["original"], replacements[record["id"]], reason,
                     )
-                sentence_violations = self.find_multi_citation_violations(revised)
-                if sentence_violations:
+                    continue
+                if status == "uncited_common_knowledge":
+                    replacements[record["id"]] = self._citation_tag_only_rewrite(record["original"], None)
+                    continue
+                if len(revised_tags) != 1:
                     raise DeepSeekRequestError(
-                        f"Citation semantic audit left multiple citations for {record['id']}: "
-                        f"{sentence_violations[:2]}"
+                        f"Citation semantic audit must select exactly one citation tag for {record['id']} while status=ok."
                     )
-                replacements[record["id"]] = self._preserve_sentence_boundary_whitespace(
-                    record["original"],
-                    revised,
-                )
+                selected = next(iter(revised_tags))
+                replacements[record["id"]] = self._citation_tag_only_rewrite(record["original"], selected)
 
         if unsupported:
             logging.warning(
-                "Citation audit removed unsupported claims: section=%s count=%s examples=%s",
-                section_name,
-                len(unsupported),
-                json.dumps(unsupported[:3], ensure_ascii=False),
+                "Citation audit flagged unsupported claims for section retry: section=%s count=%s examples=%s",
+                section_name, len(unsupported), json.dumps(unsupported[:3], ensure_ascii=False),
             )
 
         rebuilt = []
@@ -4247,14 +4360,6 @@ Return exactly:
         audited = "".join(rebuilt)
         audited = re.sub(r"[ \t]+\n", "\n", audited)
         audited = re.sub(r"\n{3,}", "\n\n", audited).strip()
-        if unsupported:
-            repaired = self._repair_local_discourse_after_citation_removal(text, audited, section_name)
-            if repaired != audited:
-                self._append_citation_audit(
-                    section_name, "repair_local_discourse_framing", audited, repaired,
-                    "Renumbered surviving ordinal/enumeration framing after unsupported claim removal; no external claim added.",
-                )
-                audited = repaired
         violations = self.find_multi_citation_violations(audited)
         if violations:
             raise DeepSeekRequestError(
@@ -4290,13 +4395,9 @@ Return exactly:
         return coverage, overlap
 
     def _local_citation_recovery(self, text, section_name='', retain_lexical=True):
-        """Conservative deterministic recovery for technical audit failures.
-
-        This routine never keeps the first citation merely because it appeared
-        first. A source is retained only when lexical evidence is sufficiently
-        strong; otherwise the externally verifiable sentence is removed and the
-        decision is written to the citation audit log.
-        """
+        """Conservative TAG-ONLY recovery for citation audit technical failures."""
+        if not hasattr(self, '_last_citation_semantic_violations'):
+            self._last_citation_semantic_violations = []
         rebuilt = []
         cursor = 0
         for start, end, sentence in self._sentence_spans(text):
@@ -4306,79 +4407,38 @@ Return exactly:
             unique_tags = list(dict.fromkeys(tags))
             valid = [tag for tag in unique_tags if tag in self.citation_map]
             revised = sentence
-
             if unique_tags and not valid:
-                revised = re.sub(r'\s*[\[(]?CITE-\d+[\])]?\s*', ' ', sentence, flags=re.I)
-                revised = re.sub(r'\s+', ' ', revised).strip()
-                self._append_citation_audit(
-                    section_name,
-                    'remove_unknown_tag',
-                    original,
-                    revised,
-                    'No valid citation tag was available.',
-                )
+                revised = self._citation_tag_only_rewrite(original, None)
+                self._append_citation_audit(section_name, 'remove_unknown_tag', original, revised, 'No valid citation tag was available.')
             elif len(valid) == 1:
                 selected_tag = valid[0]
                 coverage, overlap = self._citation_support_score(sentence, selected_tag)
                 if retain_lexical and overlap >= 2 and coverage >= 0.12:
-                    self._append_citation_audit(
-                        section_name,
-                        'retain_lexically_supported_source',
-                        original,
-                        sentence,
-                        f'Lexical support overlap={overlap}, coverage={coverage:.3f}.',
-                    )
+                    revised = self._citation_tag_only_rewrite(original, selected_tag)
                 else:
-                    revised = ''
-                    self._append_citation_audit(
-                        section_name,
-                        'remove_unverified_claim',
-                        original,
-                        '',
-                        f'The single candidate did not meet the conservative support threshold; overlap={overlap}, coverage={coverage:.3f}.',
-                    )
+                    revised = self._citation_tag_only_rewrite(original, None)
+                    self._last_citation_semantic_violations.append({
+                        'type':'citation_support_unverified', 'sentence':str(original).strip()[:500],
+                        'reason':f'No verified citation support; overlap={overlap}, coverage={coverage:.3f}.',
+                    })
+                    self._append_citation_audit(section_name, 'flag_unverified_claim_for_section_retry', original, revised, f'overlap={overlap}, coverage={coverage:.3f}')
             elif len(valid) > 1:
                 scored = sorted(
-                    (
-                        (*self._citation_support_score(sentence, tag), -self._citation_tag_number(tag), tag)
-                        for tag in valid
-                    ),
+                    ((*self._citation_support_score(sentence, tag), -self._citation_tag_number(tag), tag) for tag in valid),
                     reverse=True,
                 )
                 coverage, overlap, _, selected_tag = scored[0]
                 if retain_lexical and overlap >= 2 and coverage >= 0.12:
-                    revised = re.sub(
-                        r'\s*[\[(]?CITE-\d+(?:\s*[,;]\s*CITE-\d+)*[\])]?\s*',
-                        ' ',
-                        sentence,
-                        flags=re.I,
-                    )
-                    revised = re.sub(r'\s+', ' ', revised).strip()
-                    punctuation = revised[-1] if revised and revised[-1] in '.!?。！？' else ''
-                    if punctuation:
-                        revised = revised[:-1].rstrip()
-                    revised = f"{revised} {selected_tag}{punctuation}"
-                    self._append_citation_audit(
-                        section_name,
-                        'select_evidence_supported_source',
-                        original,
-                        revised,
-                        f'Lexical support overlap={overlap}, coverage={coverage:.3f}.',
-                    )
+                    revised = self._citation_tag_only_rewrite(original, selected_tag)
+                    self._append_citation_audit(section_name, 'select_evidence_supported_source', original, revised, f'overlap={overlap}, coverage={coverage:.3f}')
                 else:
-                    revised = ''
-                    self._append_citation_audit(
-                        section_name,
-                        'remove_unverified_claim',
-                        original,
-                        '',
-                        f'No candidate met the conservative support threshold; best overlap={overlap}, coverage={coverage:.3f}.',
-                    )
-            rebuilt.append(
-                self._preserve_sentence_boundary_whitespace(original, revised)
-                if revised
-                else ""
-            )
+                    revised = self._citation_tag_only_rewrite(original, None)
+                    self._last_citation_semantic_violations.append({
+                        'type':'citation_support_unverified', 'sentence':str(original).strip()[:500],
+                        'reason':f'No verified candidate support; best overlap={overlap}, coverage={coverage:.3f}.',
+                    })
+                    self._append_citation_audit(section_name, 'flag_unverified_claim_for_section_retry', original, revised, f'best overlap={overlap}, coverage={coverage:.3f}')
+            rebuilt.append(revised)
             cursor = end
         rebuilt.append(str(text)[cursor:])
         recovered = ''.join(rebuilt)
@@ -4461,6 +4521,7 @@ Return exactly:
         return revised
 
     def audit_and_resolve_citations(self, text, section_name='', quarantine=False):
+        self._last_citation_semantic_violations = []
         value = str(text or '')
         if quarantine:
             return self.resolve_citations(value, quarantine=True)
@@ -4505,6 +4566,21 @@ Return exactly:
             )
         return self.resolve_citations(value, quarantine=False)
 
+    @staticmethod
+    def inspect_authorial_ai_meta(text):
+        """Return typed high-confidence authorial AI-meta findings without editing text."""
+        findings = find_authorial_ai_meta_spans(str(text or ""))
+        output = []
+        for finding in findings:
+            output.append({
+                "type":"authorial_ai_meta_trace",
+                "code":str(finding.get("code") or finding.get("type") or "AI_META"),
+                "preview":str(finding.get("text") or finding.get("match") or "")[:180],
+                "start":int(finding.get("start") or 0),
+                "end":int(finding.get("end") or 0),
+            })
+        return output
+
     @staticmethod
     def _remove_ai_meta_traces(text):
         """Bounded deterministic repair for high-confidence authorial meta only.
@@ -4638,13 +4714,13 @@ TEXT:
             return value
         return revised.strip()
 
-    def critique_and_repair_section(self, text, *, section_id, section_name, mode, evidence_contract=''):
-        """Independent top-journal critic with bounded local repair."""
+    def critique_section(self, text, *, section_id, section_name, mode, evidence_contract=''):
+        """Independent top-journal critic that returns feedback only; never rewrites prose."""
         value = str(text or '').strip()
         mode = 'asq' if str(mode).lower() == 'asq' else 'amj'
         critical_sections = ({'introduction', 'theory', 'discussion'} if mode == 'amj' else {'introduction', 'theoretical_framing', 'findings', 'discussion'})
         if str(section_id or '') not in critical_sections or len(value.split()) < 120:
-            return value
+            return {'pass': True, 'critical_issue': '', 'repair_instruction': ''}
         dimensions = APP_CONFIG.get('top_journal_critic_profiles', {}).get(mode, [])
         profile = APP_CONFIG.get('section_reasoning_profiles', {}).get(mode, {}).get(str(section_id or ''), '')
         prompt = f"""Assess ONLY this manuscript unit as an independent top-journal critic. Return valid JSON only:
@@ -4660,35 +4736,30 @@ TEXT:
         )
         parsed = self._extract_json_from_text(raw)
         if not isinstance(parsed, dict) or parsed.get('pass') is not False:
-            return value
-        instruction = re.sub(r'\s+', ' ', str(parsed.get('repair_instruction') or parsed.get('critical_issue') or '')).strip()
+            return {'pass': True, 'critical_issue': '', 'repair_instruction': ''}
+        issue = re.sub(r'\s+', ' ', str(parsed.get('critical_issue') or '')).strip()
+        instruction = re.sub(r'\s+', ' ', str(parsed.get('repair_instruction') or issue)).strip()
         if not instruction:
-            return value
-        repair_prompt = f"""Repair only the material issue identified by an independent reviewer.
-Do not change or remove any number, citation tag, quotation/QUOTE tag, hypothesis label, variable abbreviation, heading, or empirical conclusion direction. Do not add a new statistic, source, quotation, setting fact, sample fact, or method. Preserve the approximate length.
-MODE: {mode}
-SECTION: {section_name}
-REVIEWER INSTRUCTION: {instruction}
-EVIDENCE CONTRACT: {evidence_contract[:4500]}
-TEXT:
-{value}
-Output only the repaired section."""
-        revised = self.call_deepseek(
-            'You are a meticulous top-journal academic editor.', repair_prompt,
-            max_tokens=max(1600, min(8000, self._count_tokens(value) + 1000)), temp=0.1,
-            model_role='writing', thinking=False, allow_fallback=False,
-        ).strip()
-        invariants = [
-            (self._numeric_tokens(value), self._numeric_tokens(revised)),
-            (Counter(re.findall(r'CITE-\d+', value, re.I)), Counter(re.findall(r'CITE-\d+', revised, re.I))),
-            (Counter(re.findall(r'QUOTE-\d+', value, re.I)), Counter(re.findall(r'QUOTE-\d+', revised, re.I))),
-            (Counter(re.findall(r'\bH\d+[A-Za-z]?\b', value)), Counter(re.findall(r'\bH\d+[A-Za-z]?\b', revised))),
-        ]
-        if any(left != right for left, right in invariants):
-            logging.warning('Top-journal local repair rejected because protected content changed: %s', section_name)
-            return value
-        logging.info('Top-journal local repair accepted: section=%s issue=%s', section_name, instruction[:180])
-        return self._remove_ai_meta_traces(revised)
+            return {'pass': True, 'critical_issue': '', 'repair_instruction': ''}
+        return {'pass': False, 'critical_issue': issue[:800], 'repair_instruction': instruction[:1200]}
+
+    def critique_and_repair_section(self, text, *, section_id, section_name, mode, evidence_contract=''):
+        """Legacy compatibility wrapper. Active S0-R1 path uses critique_section().
+
+        The wrapper intentionally preserves manuscript bytes and only records the
+        critic verdict. It no longer owns a second writing-model rewrite.
+        """
+        value = str(text or '').strip()
+        verdict = self.critique_section(
+            value, section_id=section_id, section_name=section_name, mode=mode,
+            evidence_contract=evidence_contract,
+        )
+        if verdict.get('pass') is False:
+            logging.warning(
+                'Top-journal critic material issue (no direct rewrite): section=%s issue=%s',
+                section_name, str(verdict.get('repair_instruction') or verdict.get('critical_issue') or '')[:180],
+            )
+        return value
 
     def _abstract_manuscript_context(self, manuscript, max_tokens=18000):
         value = str(manuscript or '').strip()

core/quantitative_engine.py

@@ -328,7 +328,13 @@ Keywords: {keywords}"""
                 if char == ".":
                     prev = value[index - 1] if index > 0 else ""
                     nxt = value[index + 1] if index + 1 < len(value) else ""
-                    decimal = prev.isdigit() and nxt.isdigit()
+                    # Keep both ordinary decimals (0.31) and leading-decimal
+                    # statistical notation (p < .05) inside one sentence.
+                    leading_decimal = bool(
+                        nxt.isdigit()
+                        and re.search(r"(?:[<>=≤≥]|\b(?:p|r|β))\s*$", value[max(0, index - 12):index], flags=re.I)
+                    )
+                    decimal = (prev.isdigit() and nxt.isdigit()) or leading_decimal
                     prefix = value[max(0, index - 12): index + 1].lower()
                     abbreviation = any(prefix.endswith(item) for item in abbreviations)
                     boundary = not decimal and not abbreviation
@@ -487,6 +493,136 @@ Keywords: {keywords}"""
             ] if strict else [],
         }
 
+    @classmethod
+    def inspect_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
+        """Pure AMJ quantitative Evidence Ledger inspection; never rewrites prose."""
+        value = str(text or "")
+        meta = dict(metadata or {})
+        ledger = dict(meta.get("quantitative_evidence_ledger") or {})
+        if not ledger:
+            ledger = cls.build_quantitative_evidence_ledger(meta)
+        sample_size = ledger.get("sample_size")
+        scale_points = ledger.get("scale_points")
+        strict = bool(ledger.get("strict"))
+        constructs = dict(ledger.get("constructs") or {})
+        results = dict(ledger.get("association_results") or {})
+        uploaded = dict(ledger.get("uploaded_metric_anchors") or {})
+        known_constructs = [str(v) for v in constructs.values() if str(v)]
+        construct_folded = [c.casefold() for c in known_constructs]
+        forbidden = [str(x) for x in ledger.get("forbidden_unowned_methods") or []]
+        role_patterns = {
+            "predictor": r"(?:predictor|independent\s+variable|predictor\s+construct)",
+            "mediator": r"(?:mediator|mediating\s+variable|mediating\s+construct)",
+            "outcome": r"(?:outcome|dependent\s+variable|outcome\s+construct)",
+        }
+        causal_patterns = (
+            r"(?i)\bhad\s+a\s+causal\s+effect\s+on\b",
+            r"(?i)\bexerted\s+a\s+causal\s+effect\s+on\b",
+            r"(?i)\bcausal\s+effect\s+on\b",
+            r"(?i)\b(?:an|a|the)\s+causal\s+effect\b",
+            r"(?i)\bcausal\s+effects\b",
+            r"(?i)\bcausal\s+effect\b",
+            r"(?i)\bcausal\s+mediation\b",
+        )
+        violations = []
+
+        def add_metric_violations(sentence, pattern, expected, *, dtype="float", label="metric"):
+            if expected is None:
+                return
+            regex = re.compile(pattern, re.I)
+            for match in regex.finditer(sentence):
+                observed = match.group(2)
+                try:
+                    same = int(float(observed)) == int(expected) if dtype == "int" else abs(float(observed) - float(expected)) < 0.0006
+                except Exception:
+                    same = False
+                if not same:
+                    expected_text = str(int(expected)) if dtype == "int" else f"{float(expected):.3f}".rstrip("0").rstrip(".")
+                    violations.append({
+                        "type":"quantitative_metric_mismatch", "metric":label,
+                        "observed":observed, "expected":expected_text,
+                        "preview":str(sentence).strip()[:180],
+                    })
+
+        for _start, _end, sentence in cls._sentence_spans(value):
+            ownership = cls._quantitative_sentence_ownership(sentence, section_name, known_constructs)
+            if ownership not in {"study", "result"}:
+                continue
+            if sample_size:
+                for match in re.finditer(r"(?i)(\b(?:sample(?:\s+size)?(?:\s+of)?|N)\s*(?:=|:)?\s*)(\d{2,6})\b", sentence):
+                    observed = int(match.group(2))
+                    if observed != int(sample_size):
+                        violations.append({"type":"sample_size_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]})
+                for match in re.finditer(r"(?i)\b(\d{2,6})\s+(respondents|participants|observations|cases)\b", sentence):
+                    observed = int(match.group(1))
+                    if observed != int(sample_size):
+                        violations.append({"type":"sample_count_mismatch","observed":observed,"expected":int(sample_size),"preview":sentence.strip()[:180]})
+            if scale_points:
+                for match in re.finditer(r"(?i)\b([2-9]|1[0-1])[- ]point\s+(Likert(?:-type)?\s+scale)", sentence):
+                    observed = int(match.group(1))
+                    if observed != int(scale_points):
+                        violations.append({"type":"scale_points_mismatch","observed":observed,"expected":int(scale_points),"preview":sentence.strip()[:180]})
+            for role, role_pattern in role_patterns.items():
+                expected = str(constructs.get(role) or "")
+                if not expected:
+                    continue
+                for candidate in known_constructs:
+                    if candidate.casefold() == expected.casefold():
+                        continue
+                    pattern = re.compile(rf"(?i)({role_pattern}\s*(?:was|is|:|=)?\s*){re.escape(candidate)}")
+                    if pattern.search(sentence):
+                        violations.append({"type":"construct_role_mismatch","role":role,"observed":candidate,"expected":expected,"preview":sentence.strip()[:180]})
+            if strict:
+                for pattern in causal_patterns:
+                    if re.search(pattern, sentence):
+                        violations.append({"type":"causal_language_violation","preview":sentence.strip()[:180]})
+                        break
+                if (any(c in sentence.casefold() for c in construct_folded) if construct_folded else False) and re.search(r"(?i)\bp\s*(?:=|<|>|≤|≥)\s*(?:0?\.\d+|\d+(?:\.\d+)?)", sentence):
+                    violations.append({"type":"unfrozen_p_value","preview":sentence.strip()[:180]})
+                if ownership == "study":
+                    bad = next((term for term in forbidden if re.search(rf"(?i)(?<![A-Za-z]){re.escape(term)}(?![A-Za-z])", sentence)), "")
+                    if bad:
+                        violations.append({"type":"unsupported_quantitative_method","method":bad,"preview":sentence.strip()[:180]})
+
+            vif_expected = results.get("predictor_mediator_vif") if strict else uploaded.get("VIF")
+            boot_expected = results.get("bootstrap_iterations") if strict else uploaded.get("bootstrap_iterations")
+            add_metric_violations(sentence, r"(\bVIF\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", vif_expected, label="VIF")
+            add_metric_violations(sentence, r"(\bbootstrap\s+(?:resamples?|iterations?)\s*(?:=|:)?\s*)(\d{2,6})", boot_expected, dtype="int", label="bootstrap_iterations")
+            if strict and results:
+                for role, key in (("mediator", "r2_mediator"), ("outcome", "r2_outcome")):
+                    construct = str(constructs.get(role) or "")
+                    if construct and results.get(key) is not None:
+                        add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)?\s*{re.escape(construct)}\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", results.get(key), label=key)
+                lower = sentence.casefold()
+                selected = None
+                specs = [
+                    ("indirect", results.get("indirect"), [constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")]),
+                    ("direct", results.get("direct"), [constructs.get("predictor"), constructs.get("outcome")]),
+                    ("a", results.get("a"), [constructs.get("predictor"), constructs.get("mediator")]),
+                    ("b", results.get("b"), [constructs.get("mediator"), constructs.get("outcome")]),
+                ]
+                if "indirect" in lower and all(str(x or "").casefold() in lower for x in specs[0][2] if x):
+                    selected = specs[0]
+                elif f"controlling {str(constructs.get('mediator') or '').casefold()}" in lower and str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
+                    selected = specs[1]
+                elif f"controlling {str(constructs.get('predictor') or '').casefold()}" in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower:
+                    selected = specs[3]
+                elif str(constructs.get("predictor") or "").casefold() in lower and str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() not in lower:
+                    selected = specs[2]
+                elif str(constructs.get("mediator") or "").casefold() in lower and str(constructs.get("outcome") or "").casefold() in lower and str(constructs.get("predictor") or "").casefold() not in lower:
+                    selected = specs[3]
+                if selected and selected[1] is not None:
+                    add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", selected[1], label=selected[0])
+            elif not strict:
+                for label, expected in dict(uploaded.get("r2_by_label") or {}).items():
+                    if label and label in sentence.casefold():
+                        add_metric_violations(sentence, rf"(R(?:²|\^2)\s*(?:for|of)\s*{re.escape(label)}\s*(?:=|:)\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_R2:{label}")
+                for path, expected in dict(uploaded.get("beta_by_path") or {}).items():
+                    left, _, right = path.partition(" -> ")
+                    if left and right and left in sentence.casefold() and right in sentence.casefold():
+                        add_metric_violations(sentence, r"((?:β|standardized\s+estimate)\s*(?:=|:)?\s*)([-+]?\d+(?:\.\d+)?)", expected, label=f"uploaded_beta:{path}")
+        return violations
+
     @classmethod
     def repair_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
         """Repair deterministic AMJ drift only in study/result-owned prose."""
@@ -656,9 +792,9 @@ Keywords: {keywords}"""
 
     @classmethod
     def validate_quantitative_study_fact_contract(cls, text, metadata, section_name=""):
-        repaired, diagnostics = cls.repair_quantitative_study_fact_contract(text, metadata, section_name=section_name)
-        if repaired.strip() != str(text or "").strip():
-            raise RuntimeError(f"quantitative_study_fact_contract_unrepaired:{section_name}:{diagnostics[:3]}")
+        violations = cls.inspect_quantitative_study_fact_contract(text, metadata, section_name=section_name)
+        if violations:
+            raise RuntimeError(f"quantitative_study_fact_contract_invalid:{section_name}:{violations[:3]}")
         return True
 
     def resolve_empirical_data(