L2 front matter language authority and delivery-first hardening

core/generation_runner.py

@@ -1810,6 +1810,62 @@ class GenerationRunner:
 
         return rows
 
+    def _delivery_first_final_abstract_language_guard(
+        self, *, mode: str, agent, quant_engine, final_abstract: str, reader_topic_en: str,
+        final_title: str, data_metadata: dict, hypothesis_ledger: dict | None, quality_notes: list,
+    ) -> tuple[str, bool]:
+        """Repair residual non-English script without turning a reader-surface defect into P0 failure.
+
+        Returns ``(abstract, rebuilt)``. Explicit P1 integrity validators remain
+        authoritative; only the language-surface defect is downgraded when a
+        provider-neutral clean reconstruction cannot be proven safe.
+        """
+        current = str(final_abstract or "").strip()
+        if not agent._english_reader_surface_has_foreign_script(current):
+            return current, False
+
+        language_fallback = self._deterministic_final_abstract(
+            mode, reader_topic_en or final_title, data_metadata, hypothesis_ledger=hypothesis_ledger
+        )
+        language_rows = self._study_fact_violations(
+            mode, agent, quant_engine, language_fallback, data_metadata, section_name="Final Abstract"
+        )
+        if hypothesis_ledger:
+            language_rows.extend(self._hypothesis_contract_violations(
+                language_fallback, hypothesis_ledger, section_name="Final Abstract"
+            ))
+        language_rows.extend(self._final_abstract_finding_violations(
+            mode, language_fallback, data_metadata, hypothesis_ledger=hypothesis_ledger
+        ))
+        fatal_language_rows, advisory_language_rows = self._partition_section_violations(
+            language_rows, section_name="Final Abstract", mode=mode, exhausted=True
+        )
+        if (
+            language_fallback
+            and not agent._english_reader_surface_has_foreign_script(language_fallback)
+            and not fatal_language_rows
+        ):
+            self._record_adaptive_completion_residual(
+                quality_notes,
+                advisory_language_rows + [{
+                    "type": "final_english_front_matter_language_fallback",
+                    "reason": "Residual non-English script in the Final Abstract was replaced by a deterministic English reconstruction from frozen research authority."
+                }],
+                section_name="Final Abstract",
+                stage="delivery_first_front_matter_language_fallback",
+            )
+            return language_fallback, True
+
+        self._record_adaptive_completion_residual(
+            quality_notes, [{
+                "type": "final_english_front_matter_language_warning",
+                "reason": "Final Abstract still contains non-English script after deterministic language fallback; delivery was preserved under the P0 Delivery-First contract."
+            }],
+            section_name="Final Abstract",
+            stage="delivery_first_front_matter_language_warning",
+        )
+        return current, False
+
     @staticmethod
     def _section_retry_feedback(violations, *, limit=8) -> str:
         rows = []
@@ -1898,10 +1954,11 @@ class GenerationRunner:
             ))
             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"
+        # Delivery-First hardening: unknown/new validators are not allowed to
+        # create an accidental terminal path merely because a developer forgot
+        # to register their severity.  Only explicit P1/hard-failure identities
+        # above may remain fatal after bounded repair.
+        return "ADVISORY"
 
     @staticmethod
     def _section_violation_lineage(row) -> str:
@@ -2687,6 +2744,73 @@ class GenerationRunner:
         """
         return str(topic or "").strip(), str(keywords or "").strip()
 
+    @staticmethod
+    def _front_matter_language_projection_hash(topic: str, keywords) -> str:
+        terms = parse_keyword_terms(keywords, explicit_user=True)
+        payload = json.dumps(
+            {"title": str(topic or "").strip(), "keywords": terms},
+            ensure_ascii=False, sort_keys=True, separators=(",", ":"),
+        ).encode("utf-8")
+        return hashlib.sha256(payload).hexdigest()
+
+    def _resolve_english_front_matter_projection(
+        self, agent, state: dict, data_metadata: dict, source_topic: str, source_keywords
+    ) -> dict:
+        """Resolve/cache a language-only English projection without mutating user intent."""
+        meta = dict(data_metadata or {})
+        contract = dict(meta.get("study_design_contract") or meta.get("front_matter_contract") or {})
+        intent = dict(contract.get("user_research_intent") or meta.get("user_research_intent") or {})
+        semantic_title = str(intent.get("source_topic") or source_topic or "").strip()
+        semantic_keywords = list(intent.get("source_keywords") or parse_keyword_terms(source_keywords, explicit_user=True))
+        source_hash = self._front_matter_language_projection_hash(semantic_title, semantic_keywords)
+        expected_count = len(semantic_keywords)
+
+        cached = dict(
+            state.get("front_matter_language_projection")
+            or meta.get("front_matter_language_projection")
+            or {}
+        )
+        if (
+            cached.get("version") == "front-matter-language-authority-v1"
+            and cached.get("source_hash") == source_hash
+        ):
+            try:
+                validated = agent.validate_reader_facing_english_projection(
+                    cached, expected_keyword_count=expected_count
+                )
+                return {
+                    "version": "front-matter-language-authority-v1",
+                    "source_hash": source_hash,
+                    "english_title": validated["title"],
+                    "english_keywords": list(validated["keywords"]),
+                    "translation_applied": bool(cached.get("translation_applied")),
+                }
+            except Exception as exc:
+                logging.warning(
+                    "Cached English front-matter language projection invalid; rebuilding | error=%s",
+                    type(exc).__name__,
+                )
+
+        resolved = agent.normalize_reader_facing_front_matter_to_english(
+            semantic_title, semantic_keywords
+        )
+        validated = agent.validate_reader_facing_english_projection(
+            resolved, expected_keyword_count=expected_count
+        )
+        projection = {
+            "version": "front-matter-language-authority-v1",
+            "source_hash": source_hash,
+            "english_title": validated["title"],
+            "english_keywords": list(validated["keywords"]),
+            "translation_applied": bool(resolved.get("translation_applied")),
+        }
+        # Presentation projection is additive metadata only. Original source_topic
+        # and source_keywords inside user_research_intent remain byte-for-byte authority.
+        data_metadata["front_matter_language_projection"] = dict(projection)
+        state["front_matter_language_projection"] = dict(projection)
+        state["data_metadata"] = data_metadata
+        return projection
+
     @staticmethod
     def _resolved_front_matter_keyword_policy(data_metadata: dict):
         """Resolve one policy object for Working/Final/retry/resume/fallback paths."""
@@ -2698,8 +2822,12 @@ class GenerationRunner:
 
     @staticmethod
     def _front_matter_keyword_fallback_terms(data_metadata: dict):
-        """Derive recoverable generated keywords only from frozen study-design authority."""
+        """Derive recoverable English reader keywords from frozen study-design authority."""
         meta = dict(data_metadata or {})
+        projection = dict(meta.get("front_matter_language_projection") or {})
+        projected = [str(item).strip() for item in (projection.get("english_keywords") or []) if str(item).strip()]
+        if projected:
+            return projected
         contract = dict(meta.get("study_design_contract") or {})
         values = []
         quant = dict(contract.get("quantitative") or {})
@@ -2733,10 +2861,12 @@ class GenerationRunner:
         policy = keyword_policy_from_contract(contract)
         if str(policy.get("authority_source") or "").upper() != "USER_EXPLICIT":
             return []
+        projection = dict(meta.get("front_matter_language_projection") or {})
+        authoritative = list(projection.get("english_keywords") or intent.get("source_keywords") or [])
         return apply_keyword_policy(
-            intent.get("source_keywords") or [],
+            authoritative,
             policy,
-            authoritative=intent.get("source_keywords") or [],
+            authoritative=authoritative,
         )
 
     @staticmethod
@@ -2746,7 +2876,8 @@ class GenerationRunner:
         contract = dict(meta.get("study_design_contract") or meta.get("front_matter_contract") or {})
         intent = dict(contract.get("user_research_intent") or meta.get("user_research_intent") or {})
         policy = keyword_policy_from_contract(contract)
-        authoritative = list(intent.get("source_keywords") or [])
+        projection = dict(meta.get("front_matter_language_projection") or {})
+        authoritative = list(projection.get("english_keywords") or intent.get("source_keywords") or [])
         values = apply_keyword_policy(
             candidate_keywords,
             policy,
@@ -2798,13 +2929,26 @@ class GenerationRunner:
 
 
     @staticmethod
-    def _target_keyword_heading_aliases(target_lang: str) -> list[str]:
-        code = str(target_lang or '').strip().casefold().replace('_', '-')
-        if code.startswith('zh'):
+    def _target_front_matter_language_code(target_lang: str) -> str:
+        value = str(target_lang or '').strip().casefold().replace('_', '-')
+        if value.startswith('zh') or 'chinese' in value or '中文' in value or '简体中文' in value:
+            return 'zh'
+        if value.startswith('ru') or 'russian' in value or 'рус' in value:
+            return 'ru'
+        if value.startswith('kk') or value.startswith('kz') or 'kazakh' in value or 'қазақ' in value:
+            return 'kk'
+        if value.startswith('en') or 'english' in value:
+            return 'en'
+        return value
+
+    @classmethod
+    def _target_keyword_heading_aliases(cls, target_lang: str) -> list[str]:
+        code = cls._target_front_matter_language_code(target_lang)
+        if code == 'zh':
             return ['关键词', '关键字']
-        if code.startswith('ru'):
+        if code == 'ru':
             return ['Ключевые слова']
-        if code.startswith('kk') or code.startswith('kz'):
+        if code == 'kk':
             return ['Түйін сөздер', 'Кілт сөздер']
         return ['Keywords']
 
@@ -2815,7 +2959,13 @@ class GenerationRunner:
         meta = dict(data_metadata or {})
         contract = dict(meta.get('study_design_contract') or {})
         intent = dict(contract.get('user_research_intent') or meta.get('user_research_intent') or {})
-        source_keywords = list(intent.get('source_keywords') or contract.get('source_keywords') or [])
+        projection = dict(meta.get('front_matter_language_projection') or {})
+        source_keywords = list(
+            projection.get('english_keywords')
+            or intent.get('source_keywords')
+            or contract.get('source_keywords')
+            or []
+        )
         mapping = dict(glossary or {})
         diagnostics = []
 
@@ -2847,7 +2997,7 @@ class GenerationRunner:
                         values = [str(row.get('label') or '') for row in construct_rows if role in (row.get('roles') or []) and row.get('label')]
                         if len(values) == 1:
                             role_label[role] = re.sub(r'\s+', ' ', str(mapping.get(values[0]) or values[0])).strip()
-                    code = str(target_lang or '').strip().casefold().replace('_', '-')
+                    code = cls._target_front_matter_language_code(target_lang)
                     suffixes = []
                     if 'moderation' in missing:
                         label = role_label.get('moderator') or ''
@@ -2958,7 +3108,8 @@ class GenerationRunner:
         # R7-R2C-R5: user keywords are research-intent identities, not expendable
         # planning seeds.  Freeze them for multilingual inheritance while still
         # distinguishing them from empirical construct roles.
-        intent_keywords = list(user_intent.get('source_keywords') or [])
+        projection = dict(meta.get('front_matter_language_projection') or {})
+        intent_keywords = list(projection.get('english_keywords') or user_intent.get('source_keywords') or [])
         if not intent_keywords:
             intent_keywords = parse_keyword_terms(keywords, explicit_user=True)
         for seed in intent_keywords:
@@ -4329,6 +4480,15 @@ class GenerationRunner:
         state.pop("effective_topic", None)
         state.pop("effective_keywords", None)
 
+        front_matter_projection = self._resolve_english_front_matter_projection(
+            agent, state, data_metadata, source_topic, source_keywords
+        )
+        reader_topic_en = str(front_matter_projection.get("english_title") or source_topic or topic or "").strip()
+        reader_keywords_en = list(front_matter_projection.get("english_keywords") or [])
+        reader_keywords_prompt = "; ".join(reader_keywords_en)
+        data_metadata["front_matter_language_projection"] = dict(front_matter_projection)
+        state["data_metadata"] = data_metadata
+
         hypothesis_ledger = dict(data_metadata.get("hypothesis_ledger") or {}) if mode == "amj" else {}
         hypothesis_contract = self._hypothesis_contract_directive(hypothesis_ledger)
         if hypothesis_ledger:
@@ -4496,7 +4656,7 @@ class GenerationRunner:
             working_front_error = None
             try:
                 working_front = agent.generate_working_front_matter(
-                    topic, keywords, research_logic_contract, synthesis_result,
+                    reader_topic_en, reader_keywords_prompt, research_logic_contract, synthesis_result,
                     data_evidence=planning_data_md, mode=mode,
                     hypothesis_contract=hypothesis_contract,
                     authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
@@ -4522,7 +4682,7 @@ class GenerationRunner:
             )
             if not working_abstract or not working_keywords:
                 fallback = self._deterministic_working_front_matter(
-                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                    mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                 )
                 working_abstract = str(fallback.get("abstract") or "").strip()
                 working_keywords = list(fallback.get("keywords") or [])
@@ -4553,7 +4713,7 @@ class GenerationRunner:
                 )
                 try:
                     working_front = agent.generate_working_front_matter(
-                        topic, keywords, research_logic_contract, synthesis_result,
+                        reader_topic_en, reader_keywords_prompt, research_logic_contract, synthesis_result,
                         data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
                         hypothesis_contract=hypothesis_contract,
                         authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
@@ -4599,7 +4759,7 @@ class GenerationRunner:
                 })
             if not working_abstract or not working_keywords:
                 fallback = self._deterministic_working_front_matter(
-                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                    mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                 )
                 working_abstract = str(fallback.get("abstract") or "").strip()
                 working_keywords = list(fallback.get("keywords") or [])
@@ -4620,7 +4780,7 @@ class GenerationRunner:
             plan = agent.generate_outline(topic, keywords, mode=mode, data_metadata=data_metadata)
             # User title is reader-facing authority.  Outline planning may adapt
             # structure but may not silently rename the study.
-            en_topic = str(source_topic or topic).strip()
+            en_topic = str(reader_topic_en or source_topic or topic).strip()
             structure = plan.get("structure", [])
             full_draft_en = (
                 f"# {en_topic}\n\n## Abstract\n{working_abstract}\n\n"
@@ -4815,7 +4975,7 @@ class GenerationRunner:
         else:
             full_draft_en = str(state.get("full_draft_en") or "")
             if str(data_metadata.get("source_type") or "") == "system_generated" and mode == "amj":
-                full_draft_en = self._replace_title(full_draft_en, topic)
+                full_draft_en = self._replace_title(full_draft_en, reader_topic_en)
                 canonical_resume_keywords = self._canonical_quantitative_front_matter_keywords(
                     mode, [], data_metadata
                 )
@@ -4934,7 +5094,7 @@ class GenerationRunner:
                 feedback = self._section_retry_feedback(resumed_front_rows)
                 try:
                     working_front = agent.generate_working_front_matter(
-                        topic, keywords, research_logic_contract, synthesis_result,
+                        reader_topic_en, reader_keywords_prompt, research_logic_contract, synthesis_result,
                         data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
                         hypothesis_contract=hypothesis_contract,
                         authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
@@ -4957,13 +5117,13 @@ class GenerationRunner:
                         type(exc).__name__,
                     )
                     fallback = self._deterministic_working_front_matter(
-                        mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                        mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                     )
                     candidate_abstract = str(fallback.get("abstract") or "").strip()
                     candidate_keywords = list(fallback.get("keywords") or [])
                 if not candidate_abstract or not candidate_keywords:
                     fallback = self._deterministic_working_front_matter(
-                        mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                        mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                     )
                     candidate_abstract = str(fallback.get("abstract") or "").strip()
                     candidate_keywords = list(fallback.get("keywords") or [])
@@ -4985,7 +5145,7 @@ class GenerationRunner:
                         logging.warning("Adaptive resumed Working Abstract accepted with residual diagnostics | count=%s", len(residual_front_rows))
                 if not candidate_abstract:
                     fallback = self._deterministic_working_front_matter(
-                        mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                        mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                     )
                     candidate_abstract = str(fallback.get("abstract") or "").strip()
                     candidate_keywords = list(fallback.get("keywords") or candidate_keywords)
@@ -5188,7 +5348,7 @@ class GenerationRunner:
         if not working_abstract and not state.get("english_finalized"):
             try:
                 working_front = agent.generate_working_front_matter(
-                    topic, keywords, research_logic_contract, synthesis_result,
+                    reader_topic_en, reader_keywords_prompt, research_logic_contract, synthesis_result,
                     data_evidence=planning_data_md, mode=mode,
                     hypothesis_contract=hypothesis_contract,
                     authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
@@ -5203,7 +5363,7 @@ class GenerationRunner:
                     type(exc).__name__,
                 )
                 working_front = self._deterministic_working_front_matter(
-                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                    mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                 )
                 quality_diagnostics.append({
                     "scope": "delivery_first_working_front_matter_fallback",
@@ -5221,7 +5381,7 @@ class GenerationRunner:
             )
             if not working_abstract or not working_keywords:
                 working_front = self._deterministic_working_front_matter(
-                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                    mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                 )
                 working_abstract = str(working_front.get("abstract") or "").strip()
                 working_keywords = list(working_front.get("keywords") or [])
@@ -5242,7 +5402,7 @@ class GenerationRunner:
                 feedback = self._section_retry_feedback(front_fact_rows)
                 try:
                     working_front = agent.generate_working_front_matter(
-                        topic, keywords, research_logic_contract, synthesis_result,
+                        reader_topic_en, reader_keywords_prompt, research_logic_contract, synthesis_result,
                         data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
                         hypothesis_contract=hypothesis_contract,
                         authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
@@ -5288,7 +5448,7 @@ class GenerationRunner:
                 })
             if not working_abstract or not working_keywords:
                 working_front = self._deterministic_working_front_matter(
-                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                    mode, reader_topic_en, reader_keywords_prompt, data_metadata, hypothesis_ledger=hypothesis_ledger
                 )
                 working_abstract = str(working_front.get("abstract") or "").strip()
                 working_keywords = list(working_front.get("keywords") or [])
@@ -6440,7 +6600,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             self._progress(92, "final_front_matter")
             try:
                 final_front = agent.generate_final_front_matter(
-                    topic, keywords, body_only, data_evidence=user_data_md, mode=mode,
+                    reader_topic_en, reader_keywords_prompt, body_only, data_evidence=user_data_md, mode=mode,
                     hypothesis_contract=hypothesis_contract,
                     authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
                     keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
@@ -6452,10 +6612,10 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 )
                 final_front = {
                     "abstract": self._deterministic_final_abstract(
-                        mode, source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
+                        mode, reader_topic_en or source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
                     ),
                     "keywords": self._canonical_quantitative_front_matter_keywords(
-                        mode, parse_keyword_terms(source_keywords or keywords, explicit_user=True), data_metadata
+                        mode, list(reader_keywords_en), data_metadata
                     ) or self._front_matter_keyword_fallback_terms(data_metadata),
                 }
                 self._record_adaptive_completion_residual(
@@ -6474,10 +6634,10 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             )
             if not final_abstract or not final_keywords:
                 final_abstract = final_abstract or self._deterministic_final_abstract(
-                    mode, source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
+                    mode, reader_topic_en or source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
                 )
                 final_keywords = final_keywords or self._canonical_quantitative_front_matter_keywords(
-                    mode, parse_keyword_terms(source_keywords or keywords, explicit_user=True), data_metadata
+                    mode, list(reader_keywords_en), data_metadata
                 ) or self._front_matter_keyword_fallback_terms(data_metadata)
                 self._record_adaptive_completion_residual(
                     quality_notes, [{
@@ -6505,7 +6665,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 )
                 try:
                     final_front = agent.generate_final_front_matter(
-                        topic, keywords, body_only, data_evidence=user_data_md, mode=mode,
+                        reader_topic_en, reader_keywords_prompt, body_only, data_evidence=user_data_md, mode=mode,
                         validation_feedback=feedback, hypothesis_contract=hypothesis_contract,
                         authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
                         keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
@@ -6517,10 +6677,10 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     )
                     final_front = {
                         "abstract": self._deterministic_final_abstract(
-                            mode, source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
+                            mode, reader_topic_en or source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
                         ),
                         "keywords": final_keywords or self._canonical_quantitative_front_matter_keywords(
-                            mode, parse_keyword_terms(source_keywords or keywords, explicit_user=True), data_metadata
+                            mode, list(reader_keywords_en), data_metadata
                         ) or self._front_matter_keyword_fallback_terms(data_metadata),
                     }
                 final_abstract = str(final_front.get("abstract") or "").strip()
@@ -6533,10 +6693,10 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 )
                 if not final_abstract or not final_keywords:
                     final_abstract = final_abstract or self._deterministic_final_abstract(
-                        mode, source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
+                        mode, reader_topic_en or source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
                     )
                     final_keywords = final_keywords or self._canonical_quantitative_front_matter_keywords(
-                        mode, parse_keyword_terms(source_keywords or keywords, explicit_user=True), data_metadata
+                        mode, list(reader_keywords_en), data_metadata
                     ) or self._front_matter_keyword_fallback_terms(data_metadata)
                     self._record_adaptive_completion_residual(
                         quality_notes, [{
@@ -6587,7 +6747,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     )
                     if residual_final_rows:
                         deterministic_abstract = self._deterministic_final_abstract(
-                            mode, source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
+                            mode, reader_topic_en or source_topic or topic, data_metadata, hypothesis_ledger=hypothesis_ledger
                         )
                         hard_rows = self._study_fact_violations(
                             mode, agent, quant_engine, deterministic_abstract, data_metadata, section_name="Final Abstract"
@@ -6616,12 +6776,12 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     "examples":final_front_fact_rows[:3],
                 })
             state.pop("final_abstract_retry_exhausted", None)
-            final_title = str(source_topic or topic or "").strip()
+            final_title = str(reader_topic_en or source_topic or topic or "").strip()
             if mode == "amj" and str(data_metadata.get("source_type") or "") == "system_generated":
                 semantic_contract = dict(data_metadata.get("front_matter_contract") or {})
                 reconciled_front = agent.reconcile_quantitative_reader_front_matter(
                     topic=source_topic, keywords=source_keywords, contract=semantic_contract,
-                    manuscript=body_only,
+                    manuscript=body_only, reader_title=reader_topic_en, reader_keywords=reader_keywords_en,
                 )
                 candidate_title = re.sub(r"\s+", " ", str(reconciled_front.get("title") or "")).strip()
                 candidate_keywords = self._canonical_quantitative_front_matter_keywords(
@@ -6640,6 +6800,17 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             full_draft_en = replace_abstract(full_draft_en, final_abstract)
             full_draft_en = self._replace_title(full_draft_en, final_title)
             full_draft_en = self._replace_keywords(full_draft_en, final_keywords)
+            agent.validate_reader_facing_english_projection(
+                {"title": final_title, "keywords": final_keywords},
+                expected_keyword_count=(len(reader_keywords_en) if reader_keywords_en else None),
+            )
+            final_abstract, language_rebuilt = self._delivery_first_final_abstract_language_guard(
+                mode=mode, agent=agent, quant_engine=quant_engine, final_abstract=final_abstract,
+                reader_topic_en=reader_topic_en, final_title=final_title, data_metadata=data_metadata,
+                hypothesis_ledger=hypothesis_ledger, quality_notes=quality_notes,
+            )
+            if language_rebuilt:
+                full_draft_en = replace_abstract(full_draft_en, final_abstract)
             if mode == "amj" and hypothesis_ledger:
                 post_front_hypothesis_rows = self._hypothesis_contract_violations(
                     full_draft_en, hypothesis_ledger, section_name="Full Manuscript", require_all=True

core/pipeline_agent.py

@@ -1640,6 +1640,117 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         return cleaned
 
 
+    @staticmethod
+    def _english_reader_surface_has_foreign_script(value):
+        """Return True only for scripts that cannot remain in an English reader-facing surface.
+
+        User intent may be entered in Chinese, Russian, or Kazakh, but the English
+        manuscript projection must not leak Han or Cyrillic source text. Latin
+        acronyms, numerals, punctuation, and ordinary English remain untouched.
+        """
+        return bool(re.search(r'[\u3400-\u9fff\u3040-\u30ff\u0400-\u052f]', str(value or '')))
+
+    @classmethod
+    def validate_reader_facing_english_projection(cls, projection, *, expected_keyword_count=None):
+        """Validate a language projection without changing the user's semantic authority."""
+        record = dict(projection or {})
+        title = str(record.get('title') or record.get('english_title') or '').strip()
+        raw_keywords = record.get('keywords')
+        if raw_keywords is None:
+            raw_keywords = record.get('english_keywords')
+        keywords = [str(item).strip() for item in (raw_keywords or []) if str(item).strip()]
+        if not title:
+            raise DeepSeekRequestError('English reader-facing title is empty.')
+        if expected_keyword_count is not None and len(keywords) != int(expected_keyword_count):
+            raise DeepSeekRequestError(
+                f'English reader-facing keyword cardinality changed: expected={int(expected_keyword_count)} observed={len(keywords)}.'
+            )
+        combined = ' '.join([title] + keywords)
+        if cls._english_reader_surface_has_foreign_script(combined):
+            raise DeepSeekRequestError('English reader-facing title/keywords contain untranslated Han or Cyrillic source text.')
+        return {'title': title, 'keywords': keywords}
+
+    def normalize_reader_facing_front_matter_to_english(self, topic, keywords):
+        """Create a faithful English language projection of user Title/Keywords.
+
+        The original user strings remain semantic authority in the study contract.
+        This method is presentation-only: English input is passed through without a
+        model call; non-English/mixed-script input is translated, never optimized.
+        Keyword cardinality and order are invariant.
+        """
+        source_title = str(topic or '').strip()
+        source_keywords = parse_keyword_terms(keywords, explicit_user=True)
+        if not source_title:
+            raise DeepSeekRequestError('User research title is empty before English language projection.')
+
+        source_surface = ' '.join([source_title] + source_keywords)
+        if not self._english_reader_surface_has_foreign_script(source_surface):
+            return {
+                'title': source_title,
+                'keywords': list(source_keywords),
+                'translation_applied': False,
+            }
+
+        indexed_keywords = [
+            {'index': index, 'text': value}
+            for index, value in enumerate(source_keywords, start=1)
+        ]
+        payload = {'title': source_title, 'keywords': indexed_keywords}
+        base_prompt = f"""Translate ONLY the supplied user research Title and Keywords into academic English.
+Return JSON only: {{"title":"...","keywords":["..."]}}.
+
+LANGUAGE-AUTHORITY CONTRACT:
+1. This is translation, not title optimization, rewriting, topic expansion, or keyword generation.
+2. Preserve the complete research meaning: focal constructs, direction/relationship wording, mediator/moderator roles, population/context, method qualifiers, acronyms, numbers, and scope.
+3. Do not add, delete, merge, split, reorder, generalize, specialize, or reinterpret any keyword. Return exactly {len(source_keywords)} keywords in the supplied order.
+4. Preserve existing English technical identities and acronyms. Translate/transliterate Han or Cyrillic reader-facing text so the returned Title/Keywords contain no Chinese, Russian, or Kazakh script.
+5. Do not mention the translation process and do not wrap JSON in Markdown.
+
+<SOURCE_USER_FRONT_MATTER>
+{json.dumps(payload, ensure_ascii=False)}
+</SOURCE_USER_FRONT_MATTER>"""
+        current_prompt = base_prompt
+        last_error = None
+        for local_attempt in range(2):
+            response = self.call_deepseek(
+                'You are a precise academic translator. Preserve research semantics exactly and change only language surface.',
+                current_prompt,
+                max_tokens=max(700, min(1600, 500 + len(source_keywords) * 90)),
+                temp=0.0,
+                model_role='writing',
+                thinking=False,
+                allow_fallback=False,
+                response_format={'type': 'json_object'},
+                logical_node='front_matter_writing',
+            )
+            try:
+                parsed = self._extract_json_from_text(response)
+                if not isinstance(parsed, dict):
+                    raise DeepSeekRequestError('English front-matter language projection was not valid JSON.')
+                candidate = self.validate_reader_facing_english_projection(
+                    parsed, expected_keyword_count=len(source_keywords)
+                )
+                return {
+                    'title': candidate['title'],
+                    'keywords': candidate['keywords'],
+                    'translation_applied': True,
+                }
+            except DeepSeekRequestError as exc:
+                last_error = exc
+                if local_attempt >= 1:
+                    raise
+                logging.warning(
+                    'English front-matter language projection retry | attempt=%s/2 | reason=%s',
+                    local_attempt + 1, self._safe_source_error(exc),
+                )
+                current_prompt = base_prompt + (
+                    '\n\nLOCAL VALIDATION RETRY: The prior response violated the language/cardinality contract. '
+                    'Return a fresh complete JSON object only. Translate the same Title and the same indexed Keywords '
+                    'faithfully into English, preserve keyword order/count exactly, and remove all Han/Cyrillic source text. '
+                    'Do not optimize or rewrite the research intent.'
+                )
+        raise DeepSeekRequestError(f'English front-matter language projection failed: {last_error}')
+
     @staticmethod
     def _front_matter_has_citation(value):
         text = str(value or '')
@@ -1679,6 +1790,8 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         word_count = cls._working_abstract_word_count(text)
         if cls._front_matter_has_citation(text):
             raise DeepSeekRequestError('Working Abstract must not contain citations.')
+        if cls._english_reader_surface_has_foreign_script(text):
+            raise DeepSeekRequestError('Working Abstract contains untranslated Han or Cyrillic source text.')
         forbidden = (
             ('RESULT_ASSERTION', r'\bwe\s+(?:find|found|show|demonstrate|observe|establish)\b'),
             ('RESULT_ASSERTION', r'\b(?:results?|findings?)\s+(?:show|indicate|reveal|demonstrate|confirm|support)\b'),
@@ -1875,24 +1988,28 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
         return issues
 
 
-    def reconcile_quantitative_reader_front_matter(self, *, topic, keywords, contract, manuscript=""):
-        """Preserve R5 user intent while retaining legacy contract reconciliation.
+    def reconcile_quantitative_reader_front_matter(
+        self, *, topic, keywords, contract, manuscript="", reader_title=None, reader_keywords=None
+    ):
+        """Preserve source semantic authority while allowing a faithful English surface.
 
-        Contracts that carry ``user_research_intent`` are authoritative: the
-        finalizer must not repair an upstream design mismatch by silently
-        renaming the user study.  Older contracts without that explicit R5
-        authority retain the pre-R5 bounded reconciliation path so historical
-        checkpoints and compatibility behavior remain stable.
+        ``user_research_intent`` remains the immutable semantic authority.  When
+        ``reader_title``/``reader_keywords`` are supplied they are a language-only
+        English projection of that same authority; they may change script/language
+        but not constructs, relations, scope, order, or keyword cardinality.
         """
         record = dict(contract or {})
         intent = dict(record.get("user_research_intent") or {})
         policy = keyword_policy_from_contract(record, fallback_keywords=keywords)
         source_topic = str(intent.get("source_topic") or record.get("source_topic") or topic or "").strip()
         source_keywords = list(intent.get("source_keywords") or [])
+        reader_source_topic = str(reader_title or source_topic or topic or "").strip()
+        reader_source_keywords = parse_keyword_terms(reader_keywords, explicit_user=True) if reader_keywords is not None else list(source_keywords)
         if str(policy.get("authority_source") or "").upper() == "USER_EXPLICIT" and source_keywords:
+            authoritative_reader = reader_source_keywords or source_keywords
             return {
-                "title": source_topic,
-                "keywords": apply_keyword_policy(source_keywords, policy, authoritative=source_keywords),
+                "title": reader_source_topic,
+                "keywords": apply_keyword_policy(authoritative_reader, policy, authoritative=authoritative_reader),
             }
 
         # No explicit user keyword authority: retain user-authoritative title when
@@ -1910,7 +2027,7 @@ Return a concise PROVISIONAL RESEARCH LOGIC CONTRACT in plain English prose. It
             logging.warning("Front-matter semantic contract missing frozen constructs; using degraded fallback.")
             fallback = self._quantitative_front_matter_fallback(record)
             if intent and source_topic:
-                fallback["title"] = source_topic
+                fallback["title"] = reader_source_topic
             return fallback
 
         source_keywords = list(record.get("source_keywords") or [])
@@ -1999,7 +2116,7 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
             extra = [str(item).strip() for item in ((critic or {}).get("unmeasured_focal_constructs") or []) if str(item).strip()]
             missing = [str(item).strip() for item in ((critic or {}).get("missing_frozen_constructs") or []) if str(item).strip()]
             if not invalid and not extra and not missing:
-                resolved_title = source_topic if intent else re.sub(r"\s+", " ", str(candidate.get("title") or "")).strip()
+                resolved_title = reader_source_topic if intent else re.sub(r"\s+", " ", str(candidate.get("title") or "")).strip()
                 return {
                     "title": resolved_title,
                     "keywords": apply_keyword_policy(candidate.get("keywords") or [], policy),
@@ -2013,7 +2130,11 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
 
         fallback = self._quantitative_front_matter_fallback(record)
         if intent and source_topic:
-            fallback["title"] = source_topic
+            fallback["title"] = reader_source_topic
+        if str(policy.get("authority_source") or "").upper() == "USER_EXPLICIT" and reader_source_keywords:
+            fallback["keywords"] = apply_keyword_policy(
+                reader_source_keywords, policy, authoritative=reader_source_keywords
+            )
         logging.warning("Adaptive front-matter reconciliation exhausted; using deterministic degraded fallback instead of failing manuscript generation.")
         return fallback
 
@@ -2028,7 +2149,7 @@ A contextual/domain term is allowed. Mark INVALID only when the proposal makes a
         prompt = f"""Create WORKING front matter for an empirical {str(mode).upper()}-oriented manuscript before body drafting.
 Return JSON only: {{\"abstract\":\"...\",\"keywords\":[\"...\"]}}.
 Aim for approximately 120-180 English words, have no citations, and function only as a continuity anchor. Word count is guidance, not a failure condition; prioritize Conciseness, Accuracy, and Rigor over mechanical compliance. It may state the research problem, theoretical tension, focal concepts, question/hypothesis architecture, design or method class, and intended contribution domain. It MUST NOT state coefficients, p-values, fit statistics, sample statistics unless already indispensable to define the design, supported/unsupported hypotheses, result direction, substantive findings, or a conclusion that has not yet been frozen. Do not predict findings.
-Preserve the user keyword identities and research scope. Do not silently replace, delete, or reinterpret an explicit construct/relation because direct literature support is sparse. The user-supplied keyword list is authoritative for reader-facing keyword identity; wording changes belong only to target-language translation.
+Preserve the supplied English keyword identities and research scope. They are a faithful language projection of the user input. Do not silently replace, delete, reorder, or reinterpret an explicit construct/relation because direct literature support is sparse. Do not translate them back into the source input language.
 TOPIC: {topic}
 USER KEYWORDS: {keywords}
 <PROVISIONAL_RESEARCH_LOGIC>
@@ -2065,7 +2186,7 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
         prompt = f"""Write FINAL English front matter for this completed {str(mode).upper()} manuscript.
 Return JSON only: {{\"abstract\":\"...\",\"keywords\":[\"...\"]}}.
 Aim for a concise Abstract of roughly 120-200 words, use no citations, and accurately state the research problem, validated design, main frozen finding(s), and contribution. Word count is guidance, not a failure condition; prioritize Conciseness, Accuracy, and Rigor. Include only the minimum numerical evidence required; do not list coefficients, p-values, and fit statistics unless one value is indispensable. Do not infer beyond the frozen evidence or completed manuscript.
-Preserve the user keyword identities and research scope. Do not silently replace, delete, or reinterpret an explicit construct/relation because direct literature support is sparse. The user-supplied keyword list is authoritative for reader-facing keyword identity; wording changes belong only to target-language translation.
+Preserve the supplied English keyword identities and research scope. They are a faithful language projection of the user input. Do not silently replace, delete, reorder, or reinterpret an explicit construct/relation because direct literature support is sparse. Do not translate them back into the source input language.
 TOPIC: {topic}
 USER KEYWORDS: {keywords}
 <FROZEN_RESEARCH_EVIDENCE>
@@ -2085,6 +2206,8 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
             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.')
+            if self._english_reader_surface_has_foreign_script(abstract):
+                raise DeepSeekRequestError('Final Abstract contains untranslated Han or Cyrillic source text.')
             meta_findings = self.inspect_authorial_ai_meta(abstract)
             if meta_findings:
                 raise DeepSeekRequestError('Final Abstract contains high-confidence authorial AI-writing meta-commentary.')

core/production_guard.py

@@ -848,6 +848,40 @@ def _aigc_visible_label_for_docx(filename: str) -> str:
     return visible_label_for_language(filename)
 
 
+def _apply_aigc_label_delivery_first(docx_bytes: bytes, filename: str):
+    """Apply/validate the AI-output label without making it a delivery blocker.
+
+    The label subsystem already has two independent writer/validator paths.  If
+    both mechanisms fail unexpectedly, P0 Delivery-First keeps the already
+    validated DOCX bytes and records an advisory.  Academic manuscript text and
+    P1 result/reference identity are never changed by this fallback.
+    """
+    original = bytes(docx_bytes or b"")
+    validate_docx_bytes(original)
+    label = _aigc_visible_label_for_docx(filename)
+    try:
+        labeled = ensure_aigc_output_label(original, visible_label=label)
+        validate_docx_bytes(labeled)
+        validate_aigc_output_label_resilient(labeled, expected_visible_label=label)
+        return labeled, label, True, None
+    except Exception as exc:
+        # Never keep a partially-mutated artifact after label failure. Reuse the
+        # pre-label DOCX only after independently proving it still opens.
+        validate_docx_bytes(original)
+        warning = {
+            "severity": "ADVISORY",
+            "section": "Output package",
+            "issue_type": "aigc_output_label_best_effort_fallback",
+            "detail": (
+                "AI output labeling could not be completed after the independent label "
+                f"fallbacks ({type(exc).__name__}). The validated pre-label DOCX was preserved "
+                "so generation/download was not interrupted."
+            ),
+            "user_action": "Add/verify the required AI-output label before external distribution if this warning appears.",
+        }
+        return original, label, False, warning
+
+
 def _validate_zip_bytes(
     zip_bytes: bytes,
     *,
@@ -859,6 +893,7 @@ def _validate_zip_bytes(
     quantitative_manifest: dict[str, Any] | None,
     qualitative_mode: bool,
     safe_mode: bool,
+    aigc_label_status: dict[str, bool] | None = None,
 ) -> None:
     try:
         with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as archive:
@@ -874,12 +909,13 @@ def _validate_zip_bytes(
             for name in (english_filename, translated_filename):
                 archived_docx = archive.read(name)
                 validate_docx_bytes(archived_docx)
-                try:
-                    validate_aigc_output_label_resilient(
-                        archived_docx, expected_visible_label=_aigc_visible_label_for_docx(name)
-                    )
-                except Exception as exc:
-                    raise OutputValidationError(f"Archived DOCX AI output label invalid: {exc}") from exc
+                if (aigc_label_status or {}).get(name, True):
+                    try:
+                        validate_aigc_output_label_resilient(
+                            archived_docx, expected_visible_label=_aigc_visible_label_for_docx(name)
+                        )
+                    except Exception as exc:
+                        raise OutputValidationError(f"Archived DOCX AI output label invalid: {exc}") from exc
                 if quantitative_manifest:
                     archived_markdown = english_markdown if name == english_filename else translated_markdown
                     validate_quantitative_table_contract(
@@ -920,22 +956,21 @@ def _validated_package_attempt(
     translated_docx = _render_delivery_docx(
         translated_markdown, qualitative_mode=qualitative_mode, safe_mode=safe_mode,
     )
-    # E3-B: label only the already-rendered delivery artifact. Manuscript
-    # Markdown, generation, translation and academic-content gates stay frozen.
-    try:
-        english_label = _aigc_visible_label_for_docx(english_filename)
-        translated_label = _aigc_visible_label_for_docx(translated_filename)
-        english_docx = ensure_aigc_output_label(english_docx, visible_label=english_label)
-        translated_docx = ensure_aigc_output_label(translated_docx, visible_label=translated_label)
-    except Exception as exc:
-        raise OutputValidationError(f"Post-generation AI output labeling failed: {exc}") from exc
+    # E3-B/E3-C + Delivery-First hardening: label only the already-rendered
+    # delivery artifact. A total label-subsystem failure is advisory rather than
+    # a reason to destroy an otherwise valid downloadable paper.
+    english_docx, english_label, english_labeled, english_label_warning = _apply_aigc_label_delivery_first(
+        english_docx, english_filename
+    )
+    translated_docx, translated_label, translated_labeled, translated_label_warning = _apply_aigc_label_delivery_first(
+        translated_docx, translated_filename
+    )
     validate_docx_bytes(english_docx)
     validate_docx_bytes(translated_docx)
-    try:
-        validate_aigc_output_label_resilient(english_docx, expected_visible_label=english_label)
-        validate_aigc_output_label_resilient(translated_docx, expected_visible_label=translated_label)
-    except Exception as exc:
-        raise OutputValidationError(f"Post-generation AI output label validation failed: {exc}") from exc
+    aigc_label_status = {
+        english_filename: bool(english_labeled),
+        translated_filename: bool(translated_labeled),
+    }
 
     quantitative_manifest = (
         (data_metadata or {}).get("quantitative_table_manifest")
@@ -955,6 +990,9 @@ def _validated_package_attempt(
 
     appendix_prefix = "Appendix/" if organize_appendix else ""
     notes = list(quality_notes or [])
+    for label_warning in (english_label_warning, translated_label_warning):
+        if label_warning:
+            notes.append(label_warning)
     if safe_mode:
         notes.append({
             "severity": "ADVISORY",
@@ -995,6 +1033,7 @@ def _validated_package_attempt(
                 quantitative_manifest=quantitative_manifest,
                 qualitative_mode=qualitative_mode,
                 safe_mode=safe_mode,
+                aigc_label_status=aigc_label_status,
             )
             return PackageResult(
                 zip_bytes=zip_bytes,
@@ -1118,37 +1157,36 @@ def build_recovery_package(
         raise OutputValidationError("Recovery package requires a materially complete frozen English manuscript.")
     try:
         english_docx = _render_delivery_docx(english, qualitative_mode=False, safe_mode=True)
-        english_label = _aigc_visible_label_for_docx(english_filename)
-        english_docx = ensure_aigc_output_label(english_docx, visible_label=english_label)
         validate_docx_bytes(english_docx)
-        validate_aigc_output_label_resilient(english_docx, expected_visible_label=english_label)
     except Exception:
         english_docx = _plain_recovery_docx(english)
-        english_label = _aigc_visible_label_for_docx(english_filename)
-        english_docx = ensure_aigc_output_label(english_docx, visible_label=english_label)
         validate_docx_bytes(english_docx)
-        validate_aigc_output_label_resilient(english_docx, expected_visible_label=english_label)
+    english_docx, english_label, english_labeled, english_label_warning = _apply_aigc_label_delivery_first(
+        english_docx, english_filename
+    )
 
     entries: dict[str, bytes] = {english_filename: english_docx}
+    recovery_label_status = {english_filename: bool(english_labeled)}
+    recovery_label_warnings = [english_label_warning] if english_label_warning else []
     target = str(translated_markdown or "").strip()
     if len(target) >= 800:
         try:
             target_docx = _render_delivery_docx(target, qualitative_mode=False, safe_mode=True)
-            target_label = _aigc_visible_label_for_docx(translated_filename)
-            target_docx = ensure_aigc_output_label(target_docx, visible_label=target_label)
             validate_docx_bytes(target_docx)
-            validate_aigc_output_label_resilient(target_docx, expected_visible_label=target_label)
         except Exception:
             try:
                 target_docx = _plain_recovery_docx(target)
-                target_label = _aigc_visible_label_for_docx(translated_filename)
-                target_docx = ensure_aigc_output_label(target_docx, visible_label=target_label)
                 validate_docx_bytes(target_docx)
-                validate_aigc_output_label_resilient(target_docx, expected_visible_label=target_label)
             except Exception:
                 target_docx = b""
         if target_docx:
+            target_docx, target_label, target_labeled, target_label_warning = _apply_aigc_label_delivery_first(
+                target_docx, translated_filename
+            )
             entries[translated_filename] = target_docx
+            recovery_label_status[translated_filename] = bool(target_labeled)
+            if target_label_warning:
+                recovery_label_warnings.append(target_label_warning)
 
     note = (
         "SmartPaper Recovery Package\n\n"
@@ -1157,6 +1195,12 @@ def build_recovery_package(
         "This recovery artifact must not be represented as a completed paid paper.\n\n"
         f"Terminal detail: {str(error_detail or '')[:1000]}\n"
     )
+    if recovery_label_warnings:
+        note += (
+            "\nAI-output labeling warning: the label subsystem failed after independent fallbacks. "
+            "The validated pre-label DOCX was preserved so the recovery download remains usable. "
+            "Add/verify the required AI-output label before external distribution.\n"
+        )
     entries["RECOVERY_NOTES.txt"] = note.encode("utf-8")
     manifest = _manifest(entries)
     entries["FILE_MANIFEST_SHA256.txt"] = manifest.encode("utf-8")
@@ -1171,15 +1215,17 @@ def build_recovery_package(
                     raise OutputValidationError(f"Recovery ZIP integrity failed at {bad}.")
                 archived_english = archive.read(english_filename)
                 validate_docx_bytes(archived_english)
-                validate_aigc_output_label_resilient(
-                    archived_english, expected_visible_label=_aigc_visible_label_for_docx(english_filename)
-                )
+                if recovery_label_status.get(english_filename, True):
+                    validate_aigc_output_label_resilient(
+                        archived_english, expected_visible_label=_aigc_visible_label_for_docx(english_filename)
+                    )
                 if translated_filename in archive.namelist():
                     archived_target = archive.read(translated_filename)
                     validate_docx_bytes(archived_target)
-                    validate_aigc_output_label_resilient(
-                        archived_target, expected_visible_label=_aigc_visible_label_for_docx(translated_filename)
-                    )
+                    if recovery_label_status.get(translated_filename, True):
+                        validate_aigc_output_label_resilient(
+                            archived_target, expected_visible_label=_aigc_visible_label_for_docx(translated_filename)
+                        )
             return PackageResult(
                 zip_bytes=zip_bytes,
                 file_names=tuple(sorted(entries)),