S0-R1-C5.5 Pre-Production Contract Freeze Failure-Surface Closure

core/generation_runner.py

@@ -49,6 +49,28 @@ TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v8'
 FINAL_CITATION_LEDGER_VERSION = 'resolved-final-ledger-v1'
 ENGLISH_SECTION_SEAL_VERSION = 'english-section-integrity-seal-v1'
 
+# S0-R1-C5.5: one shared private-vocabulary registry is consumed by both
+# Writer-facing sanitization and final reader-facing validation.  Keeping a
+# single registry prevents the historical drift where a phrase was allowed
+# into the Writer prompt/output but rejected later by the purity Gate.
+AUTHOR_FACING_PRIVATE_VOCABULARY = (
+    (r"\bfrozen\s+research\s+evidence\s+packet\b", "study evidence"),
+    (r"\bfrozen\s+evidence\s+packet\b", "study evidence"),
+    (r"\bevidence\s+packet\b", "study evidence"),
+    (r"\b(?:study\s+)?evidence\s+ledger\b", "study evidence record"),
+    (r"\bfrozen\s+(?:research\s+)?evidence\b", "study evidence"),
+    (r"\bfrozen\s+study\s+evidence\b", "study evidence"),
+    (r"\bfrozen\s+(?:research\s+|study\s+)?(facts?|numbers?|design|setting|sample|measures?|results?|themes?|quotations?|procedures?)\b", r"documented \1"),
+    (r"\b(?:study|result)-owned\s+facts?\b", "study-specific facts"),
+    (r"\bfrozen\s+task-level\s+concepts?\b", "canonical research concepts"),
+    (r"\bfrozen\s+theme\s+summary\b", "theme summary"),
+    (r"\bfrozen\s+thematic\s+structure\b", "thematic structure"),
+    (r"\bfrozen\s+statistical\s+evidence\b", "study statistical evidence"),
+    (r"\bfrozen\s+measurement\s+design\b", "documented measurement design"),
+    (r"\bfrozen\s+association\s+model\b", "specified association model"),
+    (r"\bfrozen\s+bootstrap\s+and\s+collinearity\s+checks\b", "documented bootstrap and collinearity checks"),
+)
+
 
 def _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")):
     lines = str(text or "").split("\n")
@@ -966,6 +988,107 @@ class GenerationRunner:
             rows.append(' | '.join(parts))
         return '\n'.join(f'- {row}' for row in rows)
 
+    @staticmethod
+    def _section_violation_severity(row, *, section_name: str = "", mode: str = "", exhausted: bool = False) -> str:
+        """Map one validation issue to REPAIRABLE/FATAL/ADVISORY.
+
+        Before the one local retry, all substantive correctness issues remain
+        repairable.  After retry exhaustion only issues that can change study
+        facts, core inference, hypothesis identity, core evidence support, or
+        citation integrity may terminate the manuscript.  Minor reader-facing
+        style/background defects become advisory notes instead of destroying an
+        otherwise complete paper.
+        """
+        item = dict(row or {}) if isinstance(row, dict) else {"type": "unknown_validation_issue"}
+        kind = str(item.get("type") or "unknown_validation_issue")
+        section = str(section_name or "")
+        if not exhausted:
+            if kind == "top_journal_critic":
+                return "ADVISORY"
+            return "REPAIRABLE"
+
+        always_fatal = {
+            "causal_language_violation",
+            "sample_size_mismatch", "sample_count_mismatch", "scale_points_mismatch",
+            "construct_role_mismatch", "quantitative_metric_mismatch", "unfrozen_p_value",
+            "unsupported_quantitative_method", "actor_data_source_relation_mismatch",
+            "unsupported_study_procedure", "enumeration_contract_invalid",
+            "hypothesis_contract_missing_id", "hypothesis_contract_ordinal_alias",
+            "hypothesis_contract_pairing_mismatch", "hypothesis_contract_unknown_id",
+            "citation_outside_frozen_section_whitelist", "authorial_ai_meta_trace",
+        }
+        if kind in always_fatal:
+            return "FATAL"
+
+        if kind in {"author_facing_internal_workflow_language", "top_journal_critic"}:
+            return "ADVISORY"
+
+        if kind in {"citation_claim_split_required", "citation_unsupported_claim", "citation_support_unverified"}:
+            core_citation_section = bool(re.search(
+                r"(?i)(?:Theoretical Mechanisms|Hypothesis Development|Methodology|Results?|Findings?)",
+                section,
+            ))
+            return "FATAL" if core_citation_section else "ADVISORY"
+
+        # Unknown validation types remain fail-closed.  This keeps C5.5 from
+        # silently downgrading a future integrity contract merely because it is
+        # new to the severity registry.
+        return "FATAL"
+
+    @classmethod
+    def _partition_section_violations(cls, rows, *, section_name: str = "", mode: str = "", exhausted: bool = False):
+        blocking, advisory = [], []
+        for row in list(rows or []):
+            severity = cls._section_violation_severity(
+                row, section_name=section_name, mode=mode, exhausted=exhausted
+            )
+            if severity == "ADVISORY":
+                advisory.append(row)
+            else:
+                blocking.append(row)
+        return blocking, advisory
+
+    @staticmethod
+    def _record_section_advisories(quality_notes: list, rows, *, section_name: str = "") -> None:
+        """Append reader-facing, nonfatal notes without exposing control vocabulary."""
+        for row in list(rows or []):
+            if not isinstance(row, dict):
+                continue
+            kind = str(row.get("type") or "quality_advisory")
+            if kind == "citation_claim_split_required":
+                detail = "A non-core literature sentence may combine more than one externally verifiable proposition."
+                action = "Before formal submission, consider splitting that sentence and checking the citation attached to each proposition."
+            elif kind in {"citation_unsupported_claim", "citation_support_unverified"}:
+                detail = "A non-core literature statement could not be fully verified against the available retrieved evidence."
+                action = "Review the statement and its supporting literature before formal submission; the manuscript was completed because the issue does not alter the study's core results or design."
+            elif kind == "author_facing_internal_workflow_language":
+                detail = "A minor reader-facing wording issue remained after automatic revision."
+                action = "Review the indicated passage for ordinary academic phrasing before formal submission."
+            elif kind == "authorial_ai_meta_trace":
+                detail = "A minor author-facing wording pattern may benefit from manual stylistic review."
+                action = "Review the indicated passage and remove any wording that does not read as ordinary scholarly prose."
+            elif kind == "top_journal_critic":
+                detail = str(row.get("reason") or "An optional journal-style refinement was identified.")[:500]
+                action = "This is a non-blocking style suggestion; revise only if it improves conciseness, accuracy, or rigor."
+            else:
+                detail = "A nonfatal quality issue was recorded for author review."
+                action = "Review the indicated section before formal submission if additional polishing is desired."
+            note = {
+                "type": "section_quality_advisory",
+                "severity": "ADVISORY",
+                "section": str(section_name or "Manuscript"),
+                "issue_type": kind,
+                "detail": detail,
+                "user_action": action,
+            }
+            signature = (note["section"], note["issue_type"], note["detail"])
+            if any(
+                (str(existing.get("section") or ""), str(existing.get("issue_type") or ""), str(existing.get("detail") or "")) == signature
+                for existing in quality_notes if isinstance(existing, dict)
+            ):
+                continue
+            quality_notes.append(note)
+
     @staticmethod
     def _citation_violation_tags(rows) -> set:
         """Extract only citation tags owned by the rejected claim sentence.
@@ -986,24 +1109,10 @@ class GenerationRunner:
 
     @staticmethod
     def _author_facing_internal_workflow_violations(text: str) -> list[dict]:
-        """Detect SmartPaper control vocabulary that must stay out of author prose.
-
-        The check is intentionally phrase-scoped.  A generic word such as
-        ``frozen`` can be legitimate academic content, but phrases naming the
-        private evidence/state machinery are never reader-facing manuscript
-        language.
-        """
+        """Detect reader-facing leakage using the same registry as sanitization."""
         value = str(text or "")
-        patterns = [
-            r"\bfrozen\s+(?:research\s+)?evidence\b",
-            r"\bfrozen\s+evidence\s+packet\b",
-            r"\bevidence\s+packet\b",
-            r"\bfrozen\s+(?:theme\s+summary|thematic\s+structure|statistical\s+evidence|measurement\s+design|association\s+model|bootstrap\s+and\s+collinearity\s+checks|task-level\s+concepts?)\b",
-            r"\b(?:study|result)-owned\s+fact\b",
-            r"\bevidence\s+ledger\b",
-        ]
         rows = []
-        for pattern in patterns:
+        for pattern, _replacement in AUTHOR_FACING_PRIVATE_VOCABULARY:
             for match in re.finditer(pattern, value, flags=re.I):
                 start = max(0, match.start() - 80)
                 end = min(len(value), match.end() + 120)
@@ -1011,10 +1120,9 @@ class GenerationRunner:
                 rows.append({
                     "type": "author_facing_internal_workflow_language",
                     "preview": preview[:320],
-                    "reason": "Private SmartPaper evidence/state vocabulary surfaced in reader-facing manuscript prose.",
+                    "reason": "Private generation-control vocabulary surfaced in reader-facing manuscript prose.",
                     "repair_instruction": (
-                        "Rewrite the sentence as ordinary academic prose describing the study, evidence, method, or limitation directly. "
-                        "Do not mention frozen evidence, evidence packets, ledgers, study-owned/result-owned facts, or other orchestration labels."
+                        "Rewrite the sentence as ordinary academic prose describing the study, evidence, method, or limitation directly."
                     ),
                 })
                 if len(rows) >= 6:
@@ -1023,34 +1131,13 @@ class GenerationRunner:
 
     @staticmethod
     def _writer_visible_evidence_context(text: str) -> str:
-        """Remove private orchestration labels before evidence reaches a Writer.
-
-        Internal evidence/state vocabulary remains unchanged in server-side metadata
-        and validators.  This view changes only the control labels presented to the
-        prose model, so AMJ and ASQ receive ordinary scholarly study descriptions
-        instead of phrases that the author-facing purity Gate later has to reject.
-        """
+        """Return a reader-safe Writer view using the shared private registry."""
         value = str(text or "")
-        substitutions = (
-            (r"(?im)^(#{1,6}\s+)Frozen\s+", r"\1"),
-            (r"(?im)^\|\s*Field\s*\|\s*Frozen value\s*\|$", "| Field | Study value |"),
-            (r"\bfrozen\s+research\s+evidence\s+packet\b", "study evidence"),
-            (r"\bfrozen\s+evidence\s+packet\b", "study evidence"),
-            (r"\bevidence\s+packet\b", "study evidence"),
-            (r"\b(?:study\s+)?evidence\s+ledger\b", "study evidence record"),
-            (r"\bfrozen\s+(?:research\s+)?evidence\b", "study evidence"),
-            (r"\bfrozen\s+study\s+evidence\b", "study evidence"),
-            (r"\bfrozen\s+(?:research\s+|study\s+)?(facts?|numbers?|design|setting|sample|measures?|results?|themes?|quotations?|procedures?)\b", r"documented \1"),
-            (r"\b(?:study|result)-owned\s+facts?\b", "study-specific facts"),
-            (r"\bfrozen\s+task-level\s+concepts?\b", "canonical research concepts"),
-            (r"\bfrozen\s+theme\s+summary\b", "theme summary"),
-            (r"\bfrozen\s+thematic\s+structure\b", "thematic structure"),
-            (r"\bfrozen\s+statistical\s+evidence\b", "study statistical evidence"),
-        )
-        for pattern, replacement in substitutions:
+        # Formatting-only labels are sanitized before the phrase registry.
+        value = re.sub(r"(?im)^(#{1,6}\s+)Frozen\s+", r"\1", value)
+        value = re.sub(r"(?im)^\|\s*Field\s*\|\s*Frozen value\s*\|$", "| Field | Study value |", value)
+        for pattern, replacement in AUTHOR_FACING_PRIVATE_VOCABULARY:
             value = re.sub(pattern, replacement, value, flags=re.I)
-        # Evidence renderers sometimes append an Executor-only sentence after a
-        # reader-relevant instruction.  Drop only that private suffix.
         value = re.sub(r"(?im);\s*the Executor\b[^\n]*", ".", value)
         value = re.sub(r"(?im)^\s*the Executor\b[^\n]*$", "", value)
         return value
@@ -2392,6 +2479,7 @@ class GenerationRunner:
         else:
             state = self.checkpoints.load_confirmed() or {}
         quality_diagnostics = list(state.get("quality_diagnostics") or [])
+        quality_notes = list(state.get("quality_notes") or [])
         self._evidence_coverage_events = list(state.get("evidence_coverage_events") or [])
         if str(state.get("evidence_status") or "") == "limited":
             self._preview_evidence_status = "limited"
@@ -3355,6 +3443,13 @@ class GenerationRunner:
                     "\nCURRENT-STUDY PROVENANCE RULE: Do not attach external literature citations to this study's own research question, H1-Hn statement, sample/design fact, analytic choice, result, limitation specific to the available study evidence, or author positioning. "
                     "If prior literature provides a rationale, state that external rationale in a separate sentence and cite only that sentence."
                 )
+            if section_id not in {"results", "findings", "appendix"}:
+                citation_rule += (
+                    "\nATOMIC LITERATURE CLAIM CONTRACT: Write one externally verifiable proposition per sentence. "
+                    "Do not combine two literature-dependent findings, mechanisms, definitions, or outcome claims under one citation. "
+                    "When two propositions require different sources, write two sentences and attach each sentence only to its own assigned source. "
+                    "A current-study fact or H1-Hn/RQ statement must remain citation-free unless a separate prior-literature rationale is stated in its own sentence."
+                )
             if bool(state.get("evidence_risk_detected")) and section_id not in {"results", "findings", "appendix"}:
                 citation_rule += (
                     "\nEVIDENCE-LIMITED CLAIM-DENSITY CONTRACT: Direct high-relevance literature coverage is limited. Treat assigned claim-evidence items as the ceiling for source-dependent empirical, definitional, prevalence, and field-gap claims in this unit. "
@@ -3465,6 +3560,15 @@ class GenerationRunner:
             prompt_context_tail = context_tail[-2500:]
             if mode == "asq":
                 prompt_context_tail = self._redact_qualitative_quotes(prompt_context_tail, data_metadata)
+            if bool(state.get("evidence_risk_detected")):
+                writer_synthesis_context = (
+                    "EVIDENCE-LIMITED SYNTHESIS BOUNDARY: direct claim-specific literature coverage is limited. "
+                    "The claim-evidence plan is the only source-dependent manuscript authority for externally verifiable literature statements in this unit. "
+                    "Do not convert broad synthesis themes, general model knowledge, or unassigned references into manuscript claims. "
+                    "When no supported external claim is assigned, prefer concise study positioning, transparent limitations, and cautious conceptual reasoning without invented citation support."
+                )
+            else:
+                writer_synthesis_context = synthesis_result
             user_prompt = f"""Write the assigned English manuscript unit.
 Section: {sec_title}
 Subsection: {subsection or 'None'}
@@ -3489,7 +3593,7 @@ AUTHOR-FACING PURITY: Write only reader-facing scholarly prose. Do not mention i
 </WORKING_ABSTRACT_ANCHOR>
 ANCHOR PRECEDENCE: The STUDY_EVIDENCE and PROVISIONAL_RESEARCH_LOGIC override the Working Abstract if any tension appears. The Working Abstract is continuity guidance only and must never force a result.
 {memory_block}
-Literature synthesis: {synthesis_result}
+Literature synthesis: {writer_synthesis_context}
 <STUDY_EVIDENCE>
 {writer_scoped_data_md}
 </STUDY_EVIDENCE>
@@ -3601,6 +3705,11 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     if mode == "asq":
                         # Model prose has no authority over quote presence/count.
                         en_raw = self._strip_model_controlled_qualitative_quotes(en_raw, data_metadata)
+                    # C5.5 uses the same private-vocabulary registry for both
+                    # Writer-facing evidence and reader-facing output.  This
+                    # deterministic phrasing cleanup cannot change study facts,
+                    # citations, hypotheses, numbers, or quote ownership.
+                    en_raw = self._writer_visible_evidence_context(en_raw)
 
                 # S0-R1-C4 phase-1 barrier: all writer-owned semantic validators
                 # inspect the same attempt before the single same-Section retry budget
@@ -3621,8 +3730,11 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     set(attempt_assigned_tag_list) - set(retry_excluded_tags)
                     if not is_quarantined else set()
                 )
+                attempt_section_text = agent.strip_ineligible_current_study_citations(
+                    prefix + en_raw, display_name
+                )
                 en_resolved = agent.audit_and_resolve_citations(
-                    prefix + en_raw, section_name=display_name, section_id=section_id,
+                    attempt_section_text, section_name=display_name, section_id=section_id,
                     quarantine=is_quarantined, allowed_tags=allowed_section_citation_tags
                 )
                 # Capture immediately: later recovery may replace the agent's last-
@@ -3632,51 +3744,67 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     list(getattr(agent, "_last_citation_semantic_violations", []) or [])
                 )
                 phase1_validation_rows = self._section_validation_bundle(
-                    mode, agent, quant_engine, prefix + en_raw, data_metadata,
+                    mode, agent, quant_engine, attempt_section_text, data_metadata,
                     section_id=section_id, section_name=display_name,
                     evidence_contract=evidence_packet,
                     include_critic=not deterministic_appendix,
                     citation_violations=citation_semantic_rows, hypothesis_ledger=hypothesis_ledger,
                 )
-                non_citation_count = max(
-                    0, len(phase1_validation_rows) - len(citation_semantic_rows)
-                )
                 if phase1_validation_rows:
                     retry_excluded_tags.update(self._citation_violation_tags(citation_semantic_rows))
-                    if section_attempt < max_section_attempts:
-                        attempt_feedback = self._section_retry_feedback(phase1_validation_rows)
+                    exhausted = section_attempt >= max_section_attempts
+                    blocking_rows, advisory_rows = self._partition_section_violations(
+                        phase1_validation_rows, section_name=display_name, mode=mode, exhausted=exhausted
+                    )
+                    if blocking_rows and not exhausted:
+                        attempt_feedback = self._section_retry_feedback(blocking_rows)
                         logging.warning(
                             "Unified Phase-1 Section retry | unit=%s attempt=%s count=%s types=%s",
-                            display_name, section_attempt, len(phase1_validation_rows),
+                            display_name, section_attempt, len(blocking_rows),
                             sorted({
                                 str(row.get("type") or "")
-                                for row in phase1_validation_rows if isinstance(row, dict)
+                                for row in blocking_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=phase1_validation_rows, pct=pct, attempts=max_section_attempts,
-                    )
-                    terminal_kind = (
-                        "section_citation_invalid_after_retry"
-                        if citation_semantic_rows and non_citation_count == 0
-                        else "section_content_invalid_after_retry"
-                    )
-                    raise RuntimeError(
-                        f"{terminal_kind}:{display_name}:{phase1_validation_rows[:3]}"
-                    )
+                    if advisory_rows:
+                        self._record_section_advisories(
+                            quality_notes, advisory_rows, section_name=display_name
+                        )
+                        logging.warning(
+                            "Nonfatal section quality advisory | unit=%s count=%s types=%s",
+                            display_name, len(advisory_rows),
+                            sorted({str(row.get("type") or "") for row in advisory_rows if isinstance(row, dict)}),
+                        )
+                    if blocking_rows:
+                        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,
+                            "quality_notes": quality_notes,
+                        })
+                        self._checkpoint_section_retry_exhausted(
+                            state, section_id=section_id, section_name=display_name,
+                            violations=blocking_rows, pct=pct, attempts=max_section_attempts,
+                        )
+                        blocking_citation_rows = [
+                            row for row in blocking_rows
+                            if isinstance(row, dict) and str(row.get("type") or "").startswith("citation_")
+                        ]
+                        terminal_kind = (
+                            "section_citation_invalid_after_retry"
+                            if blocking_citation_rows and len(blocking_citation_rows) == len(blocking_rows)
+                            else "section_content_invalid_after_retry"
+                        )
+                        raise RuntimeError(
+                            f"{terminal_kind}:{display_name}:{blocking_rows[:3]}"
+                        )
 
                 used_after = set(agent.used_citations)
                 survived = used_after - used_before
@@ -3802,32 +3930,47 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     hypothesis_ledger=hypothesis_ledger,
                 )
                 if post_validation_rows:
-                    if section_attempt < max_section_attempts:
-                        attempt_feedback = self._section_retry_feedback(post_validation_rows)
+                    exhausted = section_attempt >= max_section_attempts
+                    blocking_rows, advisory_rows = self._partition_section_violations(
+                        post_validation_rows, section_name=display_name, mode=mode, exhausted=exhausted
+                    )
+                    if blocking_rows and not exhausted:
+                        attempt_feedback = self._section_retry_feedback(blocking_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)}),
+                            display_name, section_attempt, len(blocking_rows),
+                            sorted({str(row.get("type") or "") for row in blocking_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]}"
-                    )
+                    if advisory_rows:
+                        self._record_section_advisories(
+                            quality_notes, advisory_rows, section_name=display_name
+                        )
+                        logging.warning(
+                            "Nonfatal post-normalization quality advisory | unit=%s count=%s types=%s",
+                            display_name, len(advisory_rows),
+                            sorted({str(row.get("type") or "") for row in advisory_rows if isinstance(row, dict)}),
+                        )
+                    if blocking_rows:
+                        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,
+                            "quality_notes": quality_notes,
+                        })
+                        self._checkpoint_section_retry_exhausted(
+                            state, section_id=section_id, section_name=display_name,
+                            violations=blocking_rows, pct=pct, attempts=max_section_attempts,
+                        )
+                        raise RuntimeError(
+                            f"section_content_invalid_after_retry:{display_name}:{blocking_rows[:3]}"
+                        )
 
                 quality_diagnostics.extend(attempt_diagnostics)
                 assigned_tag_list = list(attempt_assigned_tag_list)
@@ -3942,6 +4085,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     "context_tail": context_tail,
                     "accumulated_deviation": accumulated_deviation,
                     "quality_diagnostics": quality_diagnostics,
+                    "quality_notes": quality_notes,
                     "citation_plan": citation_plan,
                     "citation_minimums": citation_minimums,
                     "citation_reserve_targets": citation_reserve_targets,
@@ -4388,6 +4532,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             citation_audit=agent.citation_audit_log,
             literature_metrics=agent.get_source_metrics(),
             quality_diagnostics=quality_diagnostics,
+            quality_notes=quality_notes,
             evidence_coverage=state.get("evidence_coverage_summary") or {
                 "status": str(state.get("evidence_status") or "adequate"),
                 "early_risk_detected": bool(state.get("evidence_risk_detected")),
@@ -4408,6 +4553,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 "package_validated": True,
                 "evidence_status": str(state.get("evidence_status") or "adequate"),
                 "evidence_coverage_summary": dict(state.get("evidence_coverage_summary") or {}),
+                "quality_notes": quality_notes,
                 "agent_state": agent.export_state(),
             }
         )

core/pipeline_agent.py

@@ -3304,6 +3304,39 @@ INPUT:
     def _citation_recovery_eligible_claim(cls, sentence, section_name=''):
         return cls._citation_claim_provenance(sentence, section_name) == 'EXTERNAL_LITERATURE_CLAIM'
 
+    def strip_ineligible_current_study_citations(self, text, section_name=''):
+        """Remove external citation tags from current-study/provenance-owned prose.
+
+        C5.5 moves provenance enforcement in front of semantic citation audit.
+        This is TAG-ONLY mutation: study facts, H1-Hn statements, research
+        questions, result-owned claims, and structural sentences keep their
+        prose unchanged while any model-added [CITE-XX] tag is removed.
+        """
+        value = str(text or '')
+        rebuilt = []
+        cursor = 0
+        for start, end, sentence in self._sentence_spans(value):
+            rebuilt.append(value[cursor:start])
+            original = str(sentence or '')
+            if re.search(r'CITE-\d+', original, re.I):
+                provenance = self._citation_claim_provenance(original, section_name)
+                if provenance != 'EXTERNAL_LITERATURE_CLAIM':
+                    revised = self._citation_tag_only_rewrite(original, None)
+                    self._append_citation_audit(
+                        section_name,
+                        'remove_nonliterature_citation_by_provenance',
+                        original, revised, provenance,
+                    )
+                    rebuilt.append(revised)
+                    cursor = end
+                    continue
+            rebuilt.append(original)
+            cursor = end
+        rebuilt.append(value[cursor:])
+        cleaned = ''.join(rebuilt)
+        cleaned = re.sub(r'[ \t]+\n', '\n', cleaned)
+        return re.sub(r'\n{3,}', '\n\n', cleaned).strip()
+
     @classmethod
     def count_external_literature_claims(cls, text, section_name=''):
         return sum(
@@ -4883,7 +4916,7 @@ Return exactly:
         self, text, section_name='', section_id='', quarantine=False, allowed_tags=None
     ):
         self._last_citation_semantic_violations = []
-        value = str(text or '')
+        value = self.strip_ineligible_current_study_citations(str(text or ''), section_name)
         if quarantine:
             return self.resolve_citations(value, quarantine=True)
         unknown = sorted({f"[{tag.upper()}]" for tag in re.findall(r'CITE-\d+', value, re.I)} - set(self.citation_map))

core/production_guard.py

@@ -572,6 +572,45 @@ def build_word_count_quality_report(diagnostics: list[dict[str, Any]] | None) ->
     ])
     return "\n".join(lines).strip() + "\n"
 
+def build_quality_notes(notes: list[dict[str, Any]] | None) -> str:
+    """Build a concise reader-facing record of nonfatal quality advisories.
+
+    C5.5 intentionally separates these notes from the manuscript itself.  The
+    paper remains clean scholarly prose; only issues that do not alter core
+    study facts, inference, hypothesis identity, translation meaning, or file
+    integrity may appear here.
+    """
+    items = [dict(item) for item in (notes or []) if isinstance(item, dict)]
+    lines = [
+        "SmartPaper Quality Notes",
+        "",
+        "Writing standard: Conciseness, Accuracy, and Rigor (CAR).",
+        "Policy: The manuscript was completed because the items below were judged nonfatal and did not alter the study's core content, main findings, research design, hypothesis identity, or file integrity.",
+        "These notes are for author review before formal submission and are intentionally kept outside the manuscript body.",
+        "",
+    ]
+    advisory = [item for item in items if str(item.get("severity") or "").upper() == "ADVISORY"]
+    if not advisory:
+        lines.append("No nonfatal quality advisory was recorded.")
+    else:
+        lines.append("Items for author review:")
+        for index, item in enumerate(advisory, 1):
+            section = str(item.get("section") or "Manuscript")
+            issue = str(item.get("issue_type") or "quality_advisory")
+            detail = str(item.get("detail") or "A nonfatal quality issue was recorded.").strip()
+            action = str(item.get("user_action") or "Review before formal submission if additional polishing is desired.").strip()
+            lines.append(f"{index}. Section: {section}")
+            lines.append(f"   Issue: {issue}")
+            lines.append(f"   Note: {detail}")
+            lines.append(f"   Suggested check: {action}")
+    lines.extend([
+        "",
+        "Important boundary:",
+        "Fatal integrity problems are not listed here because they must stop generation rather than be downgraded to advisory status.",
+    ])
+    return "\n".join(lines).strip() + "\n"
+
+
 def build_evidence_coverage_report(summary: dict[str, Any] | None) -> str:
     """Build a neutral reader-facing literature evidence coverage report.
 
@@ -644,6 +683,7 @@ def build_validated_package(
     citation_audit: list[dict[str, Any]],
     literature_metrics: dict[str, Any],
     quality_diagnostics: list[dict[str, Any]] | None = None,
+    quality_notes: list[dict[str, Any]] | None = None,
     evidence_coverage: dict[str, Any] | None = None,
 ) -> PackageResult:
     validate_markdown_manuscript(english_markdown, language_name="English")
@@ -680,6 +720,7 @@ def build_validated_package(
         "CITATION_AUDIT.json": json.dumps(citation_audit or [], ensure_ascii=False, indent=2).encode("utf-8"),
         "LITERATURE_METRICS.json": json.dumps(literature_metrics or {}, ensure_ascii=False, indent=2).encode("utf-8"),
         "WORD_COUNT_QUALITY_REPORT.txt": build_word_count_quality_report(quality_diagnostics).encode("utf-8"),
+        "SmartPaper_Quality_Notes.txt": build_quality_notes(quality_notes).encode("utf-8"),
         "EVIDENCE_COVERAGE_REPORT.txt": build_evidence_coverage_report(evidence_coverage).encode("utf-8"),
     }
     manifest = _manifest(entries)

core/quantitative_engine.py

@@ -602,40 +602,23 @@ Keywords: {keywords}"""
 
     @classmethod
     def _classify_quantitative_causal_clause(cls, clause):
-        """Classify study/result prose as causal assertion, explicit limitation, or non-causal.
+        """Classify one clause by causal *assertion scope*, not token presence.
 
-        The contract is semantic at the clause level rather than a blacklist of the
-        token ``causal``.  Explicit statements that restrict interpretation to
-        association/covariation are permitted, while affirmative causal verbs and
-        causal-identification claims remain fail-closed.
+        C5.5 freezes a single semantic contract for Writer/Validator/Retry. A
+        causal phrase inside an explicit limitation (for example, ``cannot
+        establish that X causes Y``) is not an affirmative claim. Conversely,
+        an actual causal assertion remains fail-closed even when another part
+        of the sentence contains a disclaimer.
         """
         value = re.sub(r"\s+", " ", str(clause or "")).strip().casefold()
         if not value:
             return "NON_CAUSAL"
 
-        causal_nouns = r"(?:causal(?:ity|\s+(?:effect|effects|mediation|mechanism|pathway|relationship|relationships|identification))|cause(?:s|d)?|causation)"
-        limitation_patterns = (
-            rf"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't|fails? to|failed to|is not able to|are not able to|was not able to|were not able to|unable to|insufficient to)\s+(?:identify|establish|demonstrate|show|infer|support|imply|claim|justify|warrant|permit|allow|prove)\b[^.;]{{0,80}}\b{causal_nouns}\b",
-            rf"\b(?:is|are|was|were)\s+insufficient\s+to\s+(?:identify|establish|demonstrate|show|infer|support|prove)\b[^.;]{{0,80}}\b{causal_nouns}\b",
-            rf"\b(?:is|are|was|were)\s+not\s+(?:designed|intended)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|prove)\b[^.;]{{0,80}}\b{causal_nouns}\b",
-            rf"\bwithout\s+(?:identifying|establishing|demonstrating|showing|inferring|supporting|proving)\b[^.;]{{0,80}}\b{causal_nouns}\b",
-            rf"\b(?:no|without)\s+(?:evidence|basis|support|identification|proof)\b[^.;]{{0,60}}\b{causal_nouns}\b",
-            rf"\b(?:not|never)\s+(?:evidence|proof)\s+(?:of|for)\b[^.;]{{0,50}}\b{causal_nouns}\b",
-            rf"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+(?:interpreted|read|treated|understood)\s+as\b[^.;]{{0,60}}\b{causal_nouns}\b",
-            rf"\b(?:is|are|was|were)\s+(?:restricted|limited|confined)\s+to\b[^.;]{{0,120}}\b(?:not|rather than|instead of)\s+(?:to\s+)?\b{causal_nouns}\b",
-            rf"\b(?:restricted|limited|confined)\s+to\b[^.;]{{0,120}}\b(?:not|rather than|instead of)\s+(?:to\s+)?\b{causal_nouns}\b",
-            rf"\b(?:association|associations|covariation|covariance|correlation|correlations|indirect association|descriptive pattern|descriptive patterns)\b[^.;]{{0,80}}\b(?:not(?:\s+as)?|rather than|instead of)\s+(?:an?\s+|the\s+)?{causal_nouns}\b",
-            rf"\b{causal_nouns}\b\s+(?:is|are|was|were)\s+(?:not|never)\s+(?:identified|established|demonstrated|shown|inferred|supported|claimed|justified|warranted|proven)\b",
-            rf"\b{causal_nouns}\b\s+(?:cannot|can't|could not|couldn't)\s+be\s+(?:identified|established|demonstrated|shown|inferred|supported|claimed|justified|warranted|proven)\b",
-            rf"\b(?:inability|lack\s+of\s+(?:an?\s+)?ability)\s+to\s+(?:identify|establish|demonstrate|show|infer|support|justify|warrant|prove)\b[^.;]{{0,80}}\b{causal_nouns}\b",
-            rf"\b(?:not|no)\s+(?:an?\s+|the\s+)?{causal_nouns}\b",
-            rf"\b(?:cannot|can't|could not|couldn't)\s+be\s+used\s+to\s+(?:infer|establish|demonstrate|show|prove)\s+(?:an?\s+|the\s+)?{causal_nouns}\b",
-            rf"\bwithout\s+(?:proving|establishing|demonstrating|showing|identifying)\s+(?:an?\s+|the\s+)?{causal_nouns}\b",
+        causal_term = (
+            r"(?:causal(?:ity|\s+(?:effect|effects|mediation|mechanism|pathway|relationship|relationships|identification))|"
+            r"causation|cause(?:s|d|ing)?)"
         )
-        if any(re.search(pattern, value, flags=re.I) for pattern in limitation_patterns):
-            return "EXPLICIT_CAUSAL_LIMITATION"
-
-        affirmative_patterns = (
+        direct_patterns = (
             r"\b(?:causes?|caused|causing)\b",
             r"\b(?:leads?|led)\s+to\b",
             r"\b(?:drives?|drove|driven)\b",
@@ -643,12 +626,91 @@ Keywords: {keywords}"""
             r"\b(?:produces?|produced)\b",
             r"\b(?:results?|resulted)\s+in\b",
             r"\b(?:transmits?|transmitted)\s+(?:an?\s+|the\s+)?causal\s+effect\b",
+            r"\bcausally\s+(?:transmits?|mediates?|affects?|influences?)\b",
             r"\b(?:has|have|had|exerts?|exerted)\s+(?:an?\s+|the\s+)?causal\s+effect\b",
-            r"\bcausal\s+(?:effect|effects|mediation|mechanism|pathway|relationship|relationships|identification)\b",
-            r"\b(?:demonstrates?|demonstrated|shows?|showed|establishes?|established|identifies?|identified|proves?|proved)\s+(?:a\s+|the\s+)?causal(?:ity|\s+(?:effect|effects|mechanism|pathway|relationship|relationships|mediation))\b",
         )
-        if any(re.search(pattern, value, flags=re.I) for pattern in affirmative_patterns):
-            return "AFFIRMATIVE_CAUSAL"
+
+        def direct_match_is_disclaimed(match):
+            prefix = value[max(0, match.start() - 220):match.start()]
+            if re.search(
+                r"(?:\b(?:does|do|did|can|could|would|should|must|may|might)\s+not|"
+                r"\bcannot|\bcan't|\bcouldn't|\bnever)\s+(?:\w+\s+){0,2}$",
+                prefix,
+                flags=re.I,
+            ):
+                return True
+            if re.search(
+                r"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't|"
+                r"fails? to|failed to|unable to|insufficient to|not able to)\s+"
+                r"(?:identify|establish|demonstrate|show|infer|support|imply|mean|claim|justify|warrant|prove)"
+                r"\b[^.;]{0,180}?(?:\bthat\b|\bwhether\b)[^.;]{0,120}$",
+                prefix,
+                flags=re.I,
+            ):
+                return True
+            if re.search(
+                r"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+"
+                r"(?:interpreted|read|treated|understood)\s+(?:as|to mean)\b[^.;]{0,140}$",
+                prefix,
+                flags=re.I,
+            ):
+                return True
+            if re.search(
+                r"\bwithout\s+(?:identifying|establishing|demonstrating|showing|inferring|supporting|proving)"
+                r"\b[^.;]{0,180}?(?:\bthat\b|\bwhether\b)[^.;]{0,120}$",
+                prefix,
+                flags=re.I,
+            ):
+                return True
+            return False
+
+        saw_disclaimed_direct = False
+        for pattern in direct_patterns:
+            for match in re.finditer(pattern, value, flags=re.I):
+                if direct_match_is_disclaimed(match):
+                    saw_disclaimed_direct = True
+                    continue
+                return "AFFIRMATIVE_CAUSAL"
+
+        affirmative_noun_patterns = (
+            rf"\b(?:is|are|was|were|represent(?:s|ed)?|constitut(?:es|ed))\s+(?:an?\s+|the\s+)?{causal_term}\b",
+            rf"\b(?:identify|identifies|identified|establish|establishes|established|demonstrate|demonstrates|demonstrated|show|shows|showed|prove|proves|proved|provide|provides|provided)\b[^.;]{{0,80}}\b{causal_term}\b",
+            rf"\b(?:evidence|proof)\s+(?:of|for)\s+(?:an?\s+|the\s+)?{causal_term}\b",
+            rf"\b{causal_term}\b[^.;]{{0,80}}\b(?:is|are|was|were)\s+(?!not\b|never\b)",
+        )
+        for pattern in affirmative_noun_patterns:
+            for match in re.finditer(pattern, value, flags=re.I):
+                prefix = value[max(0, match.start() - 80):match.start()]
+                if re.search(
+                    r"\b(?:no|not|without|cannot|can't|insufficient|unable|does not|doesn't|do not|don't|did not|didn't|not able)\b",
+                    prefix,
+                    flags=re.I,
+                ):
+                    continue
+                return "AFFIRMATIVE_CAUSAL"
+
+        limitation_patterns = (
+            rf"\b(?:cannot|can't|could not|couldn't|does not|doesn't|do not|don't|did not|didn't|fails? to|failed to|"
+            rf"is not able to|are not able to|was not able to|were not able to|unable to|insufficient to|not able to)\s+"
+            rf"(?:identify|establish|demonstrate|show|infer|support|imply|mean|claim|justify|warrant|permit|allow|prove)\b[^.;]*\b{causal_term}\b",
+            rf"\b(?:is|are|was|were)\s+not\s+(?:designed|intended)\s+to\s+"
+            rf"(?:identify|establish|demonstrate|show|infer|support|prove)\b[^.;]*\b{causal_term}\b",
+            rf"\bwithout\s+(?:identifying|establishing|demonstrating|showing|inferring|supporting|proving)\b[^.;]*\b{causal_term}\b",
+            rf"\b(?:no|without)\s+(?:evidence|basis|support|identification|proof)\b[^.;]*\b{causal_term}\b",
+            rf"\b(?:is|are|was|were)\s+not\s+(?:evidence|proof)\s+(?:of|for)\s+(?:an?\s+|the\s+)?{causal_term}\b",
+            rf"\b(?:cannot|can't|could not|couldn't)\s+be\s+used\s+to\s+(?:infer|establish|demonstrate|show|prove)\b[^.;]*\b{causal_term}\b",
+            rf"\b(?:should not|shouldn't|must not|cannot|can't|could not|couldn't)\s+be\s+"
+            rf"(?:interpreted|read|treated|understood)\s+(?:as|to mean)\b[^.;]*\b{causal_term}\b",
+            rf"\b(?:is|are|was|were)?\s*(?:restricted|limited|confined)\s+to\b[^.;]*"
+            rf"\b(?:not|rather than|instead of)\s+(?:to\s+)?(?:an?\s+|the\s+)?{causal_term}\b",
+            rf"\b(?:association|associations|covariation|covariance|correlation|correlations|indirect association|descriptive pattern|descriptive patterns)\b[^.;]*"
+            rf"\b(?:not(?:\s+as)?|rather than|instead of)\s+(?:an?\s+|the\s+)?{causal_term}\b",
+            rf"\b{causal_term}\b\s+(?:is|are|was|were)\s+(?:not|never)\s+"
+            rf"(?:identified|established|demonstrated|shown|inferred|supported|claimed|justified|warranted|proven)\b",
+            rf"\b(?:not|no)\s+(?:an?\s+|the\s+)?{causal_term}\b",
+        )
+        if saw_disclaimed_direct or any(re.search(pattern, value, flags=re.I) for pattern in limitation_patterns):
+            return "EXPLICIT_CAUSAL_LIMITATION"
         return "NON_CAUSAL"
 
     @classmethod