U1-R2C-G1 Delivery-First Completion

app.py

@@ -1333,6 +1333,21 @@ def _render_task_status_once(lang_code: str) -> None:
         st.error(error_message)
         if translation_note:
             st.warning(translation_note)
+        recovery_path = _task_output_path(task)
+        if (
+            recovery_path
+            and recovery_path.is_file()
+            and recovery_path.name.startswith("SmartPaper_Recovery_")
+            and recovery_path.suffix.lower() == ".zip"
+        ):
+            st.warning(tr(lang_code, "recovery_notice"))
+            st.download_button(
+                label=tr(lang_code, "recovery_download"),
+                data=recovery_path.read_bytes(),
+                file_name=recovery_path.name,
+                mime="application/zip",
+                key=f"sp_recovery_download_{task_id}",
+            )
     elif status == "cancelled":
         detail = str(task.get("safe_error_detail") or "")
         if detail:

core/generation_runner.py

@@ -28,7 +28,7 @@ from core.checkpoint_store import TaskCheckpointStore
 from core.meta_extractor import extract_golden_metadata
 from core.model_gateway import resolve_plan_policy_snapshot
 from core.pipeline_agent import SSciPipelineAgent, PipelineCancelledError
-from core.production_guard import build_validated_package, replace_abstract
+from core.production_guard import build_validated_package, build_recovery_package, replace_abstract
 from core.qualitative_engine import get_qualitative_engine
 from core.quantitative_engine import get_quantitative_engine
 from core.rag_engine import initialize_rag
@@ -91,6 +91,20 @@ def _clean_markdown_headings(text):
     return re.sub(r"(?m)^#+\s+(.*)$", r"**\1**", str(text or ""))
 
 
+def _regex_sub_literal(pattern, replacement, text, *, flags=0, count=0):
+    r"""Run regex matching while treating replacement bytes as literal text.
+
+    Dynamic evidence/RAG payloads may legitimately contain backslashes such as
+    ``\h`` or LaTeX commands. Passing those strings directly as ``re.sub``'s
+    replacement template can raise ``re.error: bad escape``. A callable
+    replacement preserves the payload byte-for-byte and disables replacement
+    template interpretation without changing regex matching semantics.
+    """
+    literal = str(replacement if replacement is not None else "")
+    source = str(text if text is not None else "")
+    return re.sub(pattern, lambda _match, value=literal: value, source, count=count, flags=flags)
+
+
 def _canonical_request_hash(request_data: dict[str, Any]) -> str:
     material = dict(request_data or {})
     material.pop("request_hash", None)
@@ -1364,6 +1378,268 @@ class GenerationRunner:
                 })
         return current, diagnostics
 
+    @classmethod
+    def _deterministic_final_abstract(
+        cls, mode: str, topic: str, data_metadata: dict, *, hypothesis_ledger: dict | None = None,
+    ) -> str:
+        """Build a conservative provider-free Final Abstract from frozen authority.
+
+        Delivery-first rule: this fallback is used only after bounded model/front-
+        matter repair has failed.  It never invents coefficients, sample sizes,
+        methods, or causal claims; it uses the user topic plus frozen finding/theme
+        ledgers already owned by the completed manuscript.
+        """
+        clean_topic = re.sub(r"\s+", " ", str(topic or "")).strip().rstrip(".")
+        if str(mode or "").strip().lower() == "amj":
+            lead = f"This survey-based quantitative study examines {clean_topic}." if clean_topic else (
+                "This survey-based quantitative study examines the associations specified in the frozen research model."
+            )
+            method = (
+                "The analysis evaluates the prespecified construct associations using the completed study evidence and reports the results conservatively."
+            )
+            wanted = {
+                "final_abstract_missing_empirical_finding",
+                "final_abstract_missing_unsupported_finding",
+                "final_abstract_missing_moderation_finding",
+                "final_abstract_missing_indirect_finding",
+            }
+        else:
+            lead = f"This qualitative study examines {clean_topic}." if clean_topic else (
+                "This qualitative study examines the focal management phenomenon defined by the completed manuscript."
+            )
+            method = (
+                "The analysis develops an evidence-bounded account from the completed qualitative study and preserves the study's documented design and evidence identities."
+            )
+            wanted = {
+                "final_abstract_missing_qualitative_finding",
+                "final_abstract_missing_qualitative_theme_findings",
+            }
+        findings = cls._canonical_final_abstract_findings(
+            mode, data_metadata, hypothesis_ledger=hypothesis_ledger, violation_types=wanted,
+        )
+        if not findings:
+            findings = [
+                "The principal findings and their boundary conditions are reported directly from the completed analysis without adding unsupported numerical or causal claims."
+            ]
+        contribution = (
+            "The study clarifies the focal relationships and boundary conditions in a form that remains aligned with the completed evidence and research design."
+            if str(mode or "").strip().lower() == "amj"
+            else "The study offers a structured interpretation of the focal process while keeping the claims aligned with the completed qualitative evidence."
+        )
+        return " ".join([lead, method, *findings, contribution]).strip()
+
+    @classmethod
+    def _deterministic_research_logic_contract(
+        cls, mode: str, topic: str, keywords: str, data_metadata: dict, *, hypothesis_ledger: dict | None = None,
+    ) -> str:
+        """Provider-free planning contract used when research-logic generation fails.
+
+        This is an internal planning anchor, not reader-facing manuscript prose.  It
+        intentionally stays conservative: no result direction, coefficient, sample
+        value, causal claim, or bibliographic assertion is introduced.  Quantitative
+        H-identities are copied only from the frozen ledger.
+        """
+        clean_topic = re.sub(r"\s+", " ", str(topic or "")).strip()
+        clean_keywords = ", ".join(parse_keyword_terms(keywords, explicit_user=True))
+        if str(mode or "").strip().casefold() == "amj":
+            parts = [
+                f"Research focus: {clean_topic}." if clean_topic else "Research focus: the management problem defined by the frozen user request.",
+                "Design: survey-based quantitative research.",
+                "Planning rule: develop the management problem and theoretical rationale without predicting coefficients, significance, support, or causal effects.",
+                "Evidence rule: use only the documented study design and completed analysis as authority for empirical facts and results.",
+            ]
+            items = [item for item in (dict(hypothesis_ledger or {}).get("items") or []) if isinstance(item, dict)]
+            for item in items:
+                hid = re.sub(r"\s+", " ", str(item.get("hypothesis_id") or "")).strip().upper()
+                predictor = re.sub(r"\s+", " ", str(item.get("predictor") or "")).strip()
+                outcome = re.sub(r"\s+", " ", str(item.get("outcome") or "")).strip()
+                moderator = re.sub(r"\s+", " ", str(item.get("moderator") or "")).strip()
+                relation = str(item.get("relation") or "association").strip().casefold()
+                if not hid or not predictor or not outcome:
+                    continue
+                if relation == "moderation" and moderator:
+                    parts.append(
+                        f"{hid} defines {moderator} as the moderator of the association between {predictor} and {outcome}."
+                    )
+                else:
+                    parts.append(f"{hid} defines the association between {predictor} and {outcome}.")
+        else:
+            parts = [
+                f"Research focus: {clean_topic}." if clean_topic else "Research focus: the management phenomenon defined by the frozen user request.",
+                "Design: qualitative research.",
+                "Planning rule: examine the organizational process, meanings, mechanisms, and boundary conditions without predicting an emergent resolution.",
+                "Evidence rule: use only the documented qualitative setting, data sources, and completed evidence; do not invent participants, quotations, counts, or findings.",
+            ]
+        if clean_keywords:
+            parts.insert(1, f"User keywords: {clean_keywords}.")
+        return " ".join(part for part in parts if str(part or "").strip()).strip()
+
+    @classmethod
+    def _deterministic_working_front_matter(
+        cls, mode: str, topic: str, keywords: str, data_metadata: dict, *, hypothesis_ledger: dict | None = None,
+    ) -> dict:
+        """Build provisional front matter without a provider call.
+
+        Working front matter is only a continuity anchor before body drafting.  The
+        fallback therefore avoids empirical findings entirely and prefers frozen
+        user keyword authority.  Final front matter remains independently rebuilt
+        from the completed manuscript later.
+        """
+        clean_topic = re.sub(r"\s+", " ", str(topic or "")).strip().rstrip(".")
+        candidate_keywords = parse_keyword_terms(keywords, explicit_user=True)
+        resolved_keywords = cls._canonical_quantitative_front_matter_keywords(
+            mode, candidate_keywords, data_metadata
+        )
+        if not resolved_keywords:
+            resolved_keywords = cls._front_matter_keyword_fallback_terms(data_metadata)
+        if not resolved_keywords:
+            resolved_keywords = ["Management Research"]
+
+        if str(mode or "").strip().casefold() == "amj":
+            lead = (
+                f"This survey-based quantitative study examines {clean_topic}."
+                if clean_topic else
+                "This survey-based quantitative study examines the management relationships defined by the user request."
+            )
+            abstract = " ".join([
+                lead,
+                "The manuscript develops the theoretical rationale for the prespecified research model and evaluates the documented associations using the completed study evidence.",
+                "At this planning stage, no coefficient, significance test, hypothesis support status, or causal conclusion is anticipated.",
+                "The final abstract will be reconstructed from the completed analysis and frozen evidence before delivery.",
+            ])
+        else:
+            lead = (
+                f"This qualitative study examines {clean_topic}."
+                if clean_topic else
+                "This qualitative study examines the management phenomenon defined by the user request."
+            )
+            abstract = " ".join([
+                lead,
+                "The manuscript uses the documented qualitative study design to examine the focal process, meanings, mechanisms, and boundary conditions.",
+                "At this planning stage, no theme, quotation, participant detail, or substantive finding is anticipated beyond the frozen study design.",
+                "The final abstract will be reconstructed from the completed qualitative evidence before delivery.",
+            ])
+        return {"abstract": abstract.strip(), "keywords": list(resolved_keywords)}
+
+    @classmethod
+    def _deterministic_hypothesis_reconcile(cls, text: str, ledger: dict) -> tuple[str, list[dict]]:
+        """Reconcile H-identities without model rewriting or result-driven remapping.
+
+        Unsafe H-labelled sentences are removed rather than semantically rewritten;
+        missing canonical identities are then appended from the frozen ledger.  This
+        keeps P1 identity integrity while avoiding a task kill for mechanically
+        repairable H-label defects.
+        """
+        value = cls._normalize_hypothesis_ordinal_aliases(str(text or ""), ledger)
+        rows = cls._hypothesis_contract_violations(value, ledger, section_name="Full Manuscript", require_all=True)
+        diagnostics: list[dict] = []
+        if not rows:
+            return value, diagnostics
+
+        # Remove whole sentences whose H label is unknown or attached to the wrong
+        # construct pairing.  Violation previews are intentionally capped, so
+        # match them as normalized prefixes against the original sentence spans
+        # rather than deleting only the truncated preview and leaving a prose tail.
+        bad_sentence_previews = []
+        unknown_ids = set()
+        for row in rows:
+            kind = str(row.get("type") or "")
+            if kind == "hypothesis_contract_pairing_mismatch" and str(row.get("sentence") or "").strip():
+                bad_sentence_previews.append(re.sub(r"\s+", " ", str(row.get("sentence") or "")).strip())
+            if kind == "hypothesis_contract_unknown_id":
+                unknown_ids.update(str(x or "").upper() for x in (row.get("observed") or []) if str(x or "").strip())
+
+        spans_to_remove = []
+        for start, end, sentence in SSciPipelineAgent._sentence_spans(value):
+            normalized_sentence = re.sub(r"\s+", " ", str(sentence or "")).strip()
+            pairing_match = any(
+                normalized_sentence == preview
+                or normalized_sentence.startswith(preview)
+                or preview.startswith(normalized_sentence)
+                for preview in bad_sentence_previews if preview
+            )
+            unknown_match = bool(unknown_ids) and any(
+                re.search(rf"\b{re.escape(hid)}\b", sentence, flags=re.I) for hid in unknown_ids
+            )
+            if pairing_match or unknown_match:
+                spans_to_remove.append((start, end, pairing_match, unknown_match))
+
+        removed_pairing = 0
+        removed_unknown = 0
+        for start, end, pairing_match, unknown_match in reversed(spans_to_remove):
+            value = value[:start] + value[end:]
+            removed_pairing += int(pairing_match)
+            removed_unknown += int(unknown_match)
+        if removed_pairing:
+            diagnostics.append({
+                "type": "hypothesis_pairing_sentence_removed",
+                "count": removed_pairing,
+                "reason": "Sentence(s) attaching a frozen H-identifier to the wrong construct pairing were removed before deterministic H-identity reconstruction.",
+            })
+        if removed_unknown:
+            diagnostics.append({
+                "type": "unknown_hypothesis_sentence_removed",
+                "count": removed_unknown,
+                "reason": "Sentence(s) containing H-identifiers outside the frozen hypothesis ledger were removed before deterministic H-identity reconstruction.",
+            })
+
+        # A residual unknown token can still exist in a heading/range that the
+        # sentence scanner did not classify as prose.  Remove the identifier only
+        # as a final mechanical cleanup; substantive H-content is never rewritten.
+        if unknown_ids:
+            pattern = r"\b(?:" + "|".join(re.escape(x) for x in sorted(unknown_ids, key=len, reverse=True)) + r")\b"
+            value, count = re.subn(pattern, "", value, flags=re.I)
+            if count:
+                diagnostics.append({
+                    "type": "unknown_hypothesis_identifier_removed",
+                    "count": count,
+                    "reason": "Residual unknown H-identifiers were removed; only the frozen hypothesis ledger may own H-label identity.",
+                })
+
+        value = re.sub(r"[ \t]{2,}", " ", value)
+        value = re.sub(r"\n{3,}", "\n\n", value).strip()
+        remaining = cls._hypothesis_contract_violations(value, ledger, section_name="Full Manuscript", require_all=True)
+        missing = []
+        for row in remaining:
+            if str(row.get("type") or "") == "hypothesis_contract_missing_id":
+                missing.extend(str(x or "").upper() for x in (row.get("expected") or []) if str(x or "").strip())
+                observed = {str(x or "").upper() for x in (row.get("observed") or []) if str(x or "").strip()}
+                missing = [x for x in missing if x not in observed]
+        if missing:
+            items = {
+                str(item.get("hypothesis_id") or "").strip().upper(): item
+                for item in (dict(ledger or {}).get("items") or []) if isinstance(item, dict)
+            }
+            lines = []
+            for hid in dict.fromkeys(missing):
+                item = dict(items.get(hid) or {})
+                predictor = re.sub(r"\s+", " ", str(item.get("predictor") or "")).strip()
+                outcome = re.sub(r"\s+", " ", str(item.get("outcome") or "")).strip()
+                moderator = re.sub(r"\s+", " ", str(item.get("moderator") or "")).strip()
+                relation = str(item.get("relation") or "association").strip().lower()
+                if not predictor or not outcome:
+                    continue
+                if relation == "moderation" and moderator:
+                    lines.append(f"{hid} states that {moderator} moderates the association between {predictor} and {outcome}.")
+                else:
+                    lines.append(f"{hid} states that {predictor} is associated with {outcome}.")
+            if lines:
+                paragraph = "Hypothesis identity summary: " + " ".join(lines)
+                # Put the deterministic summary at the end of the theory section
+                # when that structural boundary is available; otherwise append to
+                # the frozen body.  No new empirical result is introduced.
+                match = re.search(r"(?m)^#\s+(?:Methodology|Methods|Research Setting and Methods)\s*$", value)
+                if match:
+                    value = value[:match.start()].rstrip() + "\n\n" + paragraph + "\n\n" + value[match.start():].lstrip()
+                else:
+                    value = value.rstrip() + "\n\n" + paragraph
+                diagnostics.append({
+                    "type": "missing_hypothesis_identity_reconstructed",
+                    "count": len(lines),
+                    "reason": "Missing H-identifiers were reconstructed only from the frozen hypothesis ledger.",
+                })
+        return value, diagnostics
+
     @classmethod
     def _final_abstract_finding_violations(
         cls, mode: str, text: str, data_metadata: dict, *, hypothesis_ledger: dict | None = None
@@ -2320,6 +2596,55 @@ class GenerationRunner:
             "sha256": cls._section_integrity_hash(section_text),
         }
 
+    def _refresh_section_integrity_seals_after_delivery_repair(
+        self, manuscript: str, section_order, section_integrity_seals: dict,
+    ) -> list[tuple[str, str, str]]:
+        """Reseal only body sections changed by a deterministic P0 repair.
+
+        G1 permits a narrow deterministic hypothesis-identity repair after the
+        ordinary section seals have been created.  Any changed section must be
+        resealed immediately so resume/checkpoint authority never points at stale
+        bytes.  The returned section text also lets the caller invalidate/requeue
+        target translation from the exact repaired English source.
+        """
+        changed: list[tuple[str, str, str]] = []
+        seals = section_integrity_seals
+        for section_key, section_title in list(section_order or []):
+            section_text = self._extract_top_level_section(manuscript, section_title)
+            if not section_text:
+                raise RuntimeError(f"delivery_first_reseal_source_missing:{section_title}")
+            current = dict(seals.get(section_key) or {})
+            updated = self._build_section_integrity_seal(section_key, section_title, section_text)
+            if str(current.get("sha256") or "") != str(updated.get("sha256") or ""):
+                seals[section_key] = updated
+                changed.append((str(section_key), str(section_title), section_text))
+        return changed
+
+    def _requeue_resealed_translation_sections(
+        self, *, changed_sections, mode: str, quant_engine, data_metadata: dict,
+        translation_agent, translation_sections: dict, english_commit_text: str,
+    ) -> int:
+        """Requeue translations whose frozen English source changed after repair."""
+        queued = 0
+        for section_key, section_title, section_text in list(changed_sections or []):
+            source_text = section_text
+            if (
+                str(mode or "").strip().casefold() == "amj"
+                and str(section_key).strip().casefold() == "results"
+                and dict(data_metadata or {}).get("quantitative_table_manifest")
+            ):
+                source_text = quant_engine._strip_target_quantitative_markdown_tables(section_text)
+            if self._enqueue_translation_unit(
+                translation_agent,
+                unit_key=section_key,
+                unit_name=section_title,
+                source_text=source_text,
+                translation_sections=translation_sections,
+                english_commit_text=english_commit_text,
+            ):
+                queued += 1
+        return queued
+
     @staticmethod
     def _replace_top_level_section(manuscript: str, title: str, replacement: str) -> str:
         value = str(manuscript or "")
@@ -2436,7 +2761,13 @@ class GenerationRunner:
         rendered = re.sub(r"\s+", " ", str(title or "")).strip()
         if not rendered:
             return value
-        updated, count = re.subn(r"(?m)^#\s+.+$", f"# {rendered}", value, count=1)
+        title_replacement = f"# {rendered}"
+        updated, count = re.subn(
+            r"(?m)^#\s+.+$",
+            lambda _match, replacement=title_replacement: replacement,
+            value,
+            count=1,
+        )
         if count == 1:
             return updated
         logging.warning("Manuscript title heading missing during final reconciliation; reconstructing canonical H1 instead of failing generation.")
@@ -4064,9 +4395,28 @@ class GenerationRunner:
                     "but do not invent citations or bibliographic metadata and do not present general model knowledge as retrieved evidence."
                 )
             else:
-                synthesis_result = self._stream_text(
-                    agent.stream_synthesize_literature(topic, refs_text, user_core_literature=user_lit_context)
-                )
+                try:
+                    synthesis_result = self._stream_text(
+                        agent.stream_synthesize_literature(topic, refs_text, user_core_literature=user_lit_context)
+                    )
+                except PipelineCancelledError:
+                    raise
+                except Exception as exc:
+                    synthesis_result = (
+                        "LITERATURE-SYNTHESIS FALLBACK: retrieved literature remains available through the citation/RAG layer, "
+                        "but the narrative synthesis provider step was unavailable. Continue from the frozen study evidence and "
+                        "claim-specific retrieved sources; do not invent bibliographic facts, expected findings, or causal claims."
+                    )
+                    quality_diagnostics.append({
+                        "scope": "delivery_first_literature_synthesis_fallback",
+                        "name": "Literature Synthesis",
+                        "status": "deterministic_fallback",
+                        "reason": type(exc).__name__,
+                    })
+                    logging.warning(
+                        "Delivery-first literature synthesis fallback activated | error=%s",
+                        type(exc).__name__,
+                    )
             agent.record_prewrite_stage("literature_synthesis", prewrite_started, prewrite_calls)
             synthesis_result = re.sub(r"<think>.*?</think>", "", synthesis_result, flags=re.DOTALL).strip()
             state.update({"synthesis_result": synthesis_result, "agent_state": agent.export_state()})
@@ -4077,43 +4427,93 @@ class GenerationRunner:
             self._progress(14, "research_logic")
             prewrite_started = time.monotonic()
             prewrite_calls = agent.ds_calls
-            research_logic_response = self._stream_text(
-                agent.stream_generate_research_logic(
-                    topic, keywords, synthesis_result, refs_text,
-                    data_evidence=planning_data_md, mode=mode,
-                    hypothesis_contract=hypothesis_contract,
+            research_logic_error = None
+            try:
+                research_logic_response = self._stream_text(
+                    agent.stream_generate_research_logic(
+                        topic, keywords, synthesis_result, refs_text,
+                        data_evidence=planning_data_md, mode=mode,
+                        hypothesis_contract=hypothesis_contract,
+                    )
                 )
-            )
-            agent.record_prewrite_stage("research_logic", prewrite_started, prewrite_calls)
+            except PipelineCancelledError:
+                raise
+            except Exception as exc:
+                research_logic_error = exc
+                research_logic_response = ""
+                logging.warning(
+                    "Delivery-first research-logic provider path unavailable; using deterministic planning fallback | error=%s",
+                    type(exc).__name__,
+                )
+            finally:
+                agent.record_prewrite_stage("research_logic", prewrite_started, prewrite_calls)
             research_logic_contract = re.sub(
                 r"<think>.*?</think>", "", research_logic_response, flags=re.DOTALL
             ).strip()
-            if hypothesis_ledger:
+            logic_hypothesis_rows = []
+            if hypothesis_ledger and research_logic_contract:
                 research_logic_contract = self._normalize_hypothesis_ordinal_aliases(
                     research_logic_contract, hypothesis_ledger
                 )
                 logic_hypothesis_rows = self._hypothesis_contract_violations(
                     research_logic_contract, hypothesis_ledger, section_name="Research Logic Contract"
                 )
-                if logic_hypothesis_rows:
+            if not research_logic_contract or logic_hypothesis_rows:
+                fallback_reason = (
+                    "provider_error" if research_logic_error is not None
+                    else "empty_contract" if not research_logic_contract
+                    else "hypothesis_identity_mismatch"
+                )
+                research_logic_contract = self._deterministic_research_logic_contract(
+                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                )
+                fallback_rows = (
+                    self._hypothesis_contract_violations(
+                        research_logic_contract, hypothesis_ledger, section_name="Research Logic Contract"
+                    ) if hypothesis_ledger else []
+                )
+                if fallback_rows:
+                    # This is an internal deterministic invariant failure, not a
+                    # model-quality rejection.  Do not continue with a corrupted
+                    # H-identity planning contract.
                     raise RuntimeError(
-                        f"research_logic_hypothesis_contract_invalid:{logic_hypothesis_rows[:3]}"
+                        f"delivery_first_research_logic_fallback_invalid:{fallback_rows[:3]}"
                     )
-            if not research_logic_contract:
-                raise RuntimeError("research_logic_contract_missing")
+                quality_diagnostics.append({
+                    "scope": "delivery_first_research_logic_fallback",
+                    "name": "Research Logic Contract",
+                    "status": "deterministic_fallback",
+                    "reason": fallback_reason,
+                    "prior_violations": logic_hypothesis_rows[:3],
+                })
+                logging.warning(
+                    "Delivery-first Research Logic fallback activated | reason=%s prior_violations=%s",
+                    fallback_reason, len(logic_hypothesis_rows),
+                )
             agent.global_memory["research_logic_contract"] = research_logic_contract
 
             self._progress(14, "working_front_matter")
-            working_front = agent.generate_working_front_matter(
-                topic, keywords, 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),
-                keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
-                fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
-            )
+            working_front_error = None
+            try:
+                working_front = agent.generate_working_front_matter(
+                    topic, keywords, 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),
+                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
+                )
+            except PipelineCancelledError:
+                raise
+            except Exception as exc:
+                working_front_error = exc
+                working_front = {}
+                logging.warning(
+                    "Delivery-first Working Front Matter provider path unavailable; preparing deterministic fallback | error=%s",
+                    type(exc).__name__,
+                )
             working_abstract = str(working_front.get("abstract") or "").strip()
-            if hypothesis_ledger:
+            if hypothesis_ledger and working_abstract:
                 working_abstract = self._normalize_hypothesis_ordinal_aliases(
                     working_abstract, hypothesis_ledger
                 )
@@ -4121,7 +4521,22 @@ class GenerationRunner:
                 mode, list(working_front.get("keywords") or []), data_metadata
             )
             if not working_abstract or not working_keywords:
-                raise RuntimeError("working_front_matter_missing")
+                fallback = self._deterministic_working_front_matter(
+                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                )
+                working_abstract = str(fallback.get("abstract") or "").strip()
+                working_keywords = list(fallback.get("keywords") or [])
+                quality_diagnostics.append({
+                    "scope": "delivery_first_working_front_matter_fallback",
+                    "name": "Working Abstract",
+                    "status": "deterministic_fallback",
+                    "reason": "provider_error" if working_front_error is not None else "missing_component",
+                })
+                logging.warning(
+                    "Delivery-first Working Front Matter fallback activated | reason=%s",
+                    "provider_error" if working_front_error is not None else "missing_component",
+                )
+
             _, front_fact_rows = self._repair_validate_study_fact_contract(
                 mode, agent, quant_engine, working_abstract, data_metadata,
                 section_name="Working Abstract",
@@ -4136,22 +4551,32 @@ class GenerationRunner:
                     "Working Abstract fact validation retry | mode=%s count=%s types=%s",
                     mode, len(front_fact_rows), sorted({str(row.get("type") or "") for row in front_fact_rows}),
                 )
-                working_front = agent.generate_working_front_matter(
-                    topic, keywords, research_logic_contract, synthesis_result,
-                    data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
-                    hypothesis_contract=hypothesis_contract,
-                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
-                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
-                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
-                )
-                working_abstract = str(working_front.get("abstract") or "").strip()
-                if hypothesis_ledger:
-                    working_abstract = self._normalize_hypothesis_ordinal_aliases(
-                        working_abstract, hypothesis_ledger
+                try:
+                    working_front = agent.generate_working_front_matter(
+                        topic, keywords, 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),
+                        keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                        fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
+                    )
+                    retry_abstract = str(working_front.get("abstract") or "").strip()
+                    if hypothesis_ledger and retry_abstract:
+                        retry_abstract = self._normalize_hypothesis_ordinal_aliases(
+                            retry_abstract, hypothesis_ledger
+                        )
+                    retry_keywords = self._canonical_quantitative_front_matter_keywords(
+                        mode, list(working_front.get("keywords") or []), data_metadata
+                    )
+                    if retry_abstract and retry_keywords:
+                        working_abstract, working_keywords = retry_abstract, retry_keywords
+                except PipelineCancelledError:
+                    raise
+                except Exception as exc:
+                    logging.warning(
+                        "Working Front Matter validation retry unavailable; continuing to deterministic/adaptive fallback | error=%s",
+                        type(exc).__name__,
                     )
-                working_keywords = self._canonical_quantitative_front_matter_keywords(
-                    mode, list(working_front.get("keywords") or []), data_metadata
-                )
                 _, remaining_front_rows = self._repair_validate_study_fact_contract(
                     mode, agent, quant_engine, working_abstract, data_metadata, section_name="Working Abstract",
                 )
@@ -4172,8 +4597,23 @@ class GenerationRunner:
                     "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
                     "status":"section_retry_resolved", "count":len(front_fact_rows), "examples":front_fact_rows[:3],
                 })
-            if not working_abstract:
-                raise RuntimeError("working_front_matter_empty_after_validation")
+            if not working_abstract or not working_keywords:
+                fallback = self._deterministic_working_front_matter(
+                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                )
+                working_abstract = str(fallback.get("abstract") or "").strip()
+                working_keywords = list(fallback.get("keywords") or [])
+                quality_diagnostics.append({
+                    "scope": "delivery_first_working_front_matter_fallback",
+                    "name": "Working Abstract",
+                    "status": "deterministic_fallback_after_validation",
+                    "reason": "empty_after_repair",
+                })
+                logging.warning(
+                    "Delivery-first Working Front Matter fallback activated after validation repair exhaustion."
+                )
+            agent.global_memory["working_abstract"] = working_abstract
+
             agent.global_memory["working_abstract"] = working_abstract
 
             self._progress(14, "outline_planning")
@@ -4393,6 +4833,7 @@ class GenerationRunner:
         research_logic_contract = str(
             state.get("research_logic_contract") or agent.global_memory.get("research_logic_contract") or ""
         )
+        resumed_logic_rows = []
         if hypothesis_ledger and research_logic_contract:
             research_logic_contract = self._normalize_hypothesis_ordinal_aliases(
                 research_logic_contract, hypothesis_ledger
@@ -4400,11 +4841,33 @@ class GenerationRunner:
             resumed_logic_rows = self._hypothesis_contract_violations(
                 research_logic_contract, hypothesis_ledger, section_name="Research Logic Contract"
             )
-            if resumed_logic_rows:
+        if not research_logic_contract or resumed_logic_rows:
+            research_logic_contract = self._deterministic_research_logic_contract(
+                mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+            )
+            fallback_rows = (
+                self._hypothesis_contract_violations(
+                    research_logic_contract, hypothesis_ledger, section_name="Research Logic Contract"
+                ) if hypothesis_ledger else []
+            )
+            if fallback_rows:
                 raise RuntimeError(
-                    f"research_logic_hypothesis_contract_invalid_on_resume:{resumed_logic_rows[:3]}"
+                    f"delivery_first_research_logic_fallback_invalid_on_resume:{fallback_rows[:3]}"
                 )
-            agent.global_memory["research_logic_contract"] = research_logic_contract
+            quality_diagnostics.append({
+                "scope": "delivery_first_research_logic_fallback",
+                "name": "Research Logic Contract",
+                "status": "deterministic_resume_fallback",
+                "reason": "missing_contract" if not str(state.get("research_logic_contract") or "").strip() else "hypothesis_identity_mismatch",
+                "prior_violations": resumed_logic_rows[:3],
+            })
+            state["research_logic_contract"] = research_logic_contract
+            state["quality_diagnostics"] = quality_diagnostics
+            logging.warning(
+                "Delivery-first Research Logic resume fallback activated | prior_violations=%s",
+                len(resumed_logic_rows),
+            )
+        agent.global_memory["research_logic_contract"] = research_logic_contract
         citation_plan = state.get("citation_plan") if isinstance(state.get("citation_plan"), dict) else {}
         citation_minimums = dict(state.get("citation_minimums") or {})
         citation_reserve_targets = dict(state.get("citation_reserve_targets") or {})
@@ -4469,22 +4932,41 @@ class GenerationRunner:
                 ))
             if resumed_front_rows:
                 feedback = self._section_retry_feedback(resumed_front_rows)
-                working_front = agent.generate_working_front_matter(
-                    topic, keywords, research_logic_contract, synthesis_result,
-                    data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
-                    hypothesis_contract=hypothesis_contract,
-                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
-                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
-                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
-                )
-                candidate_abstract = str(working_front.get("abstract") or "").strip()
-                if hypothesis_ledger:
-                    candidate_abstract = self._normalize_hypothesis_ordinal_aliases(
-                        candidate_abstract, hypothesis_ledger
+                try:
+                    working_front = agent.generate_working_front_matter(
+                        topic, keywords, 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),
+                        keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                        fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
                     )
-                candidate_keywords = self._canonical_quantitative_front_matter_keywords(
-                    mode, list(working_front.get("keywords") or []), data_metadata
-                )
+                    candidate_abstract = str(working_front.get("abstract") or "").strip()
+                    if hypothesis_ledger and candidate_abstract:
+                        candidate_abstract = self._normalize_hypothesis_ordinal_aliases(
+                            candidate_abstract, hypothesis_ledger
+                        )
+                    candidate_keywords = self._canonical_quantitative_front_matter_keywords(
+                        mode, list(working_front.get("keywords") or []), data_metadata
+                    )
+                except PipelineCancelledError:
+                    raise
+                except Exception as exc:
+                    logging.warning(
+                        "Resumed Working Front Matter repair unavailable; using deterministic fallback | error=%s",
+                        type(exc).__name__,
+                    )
+                    fallback = self._deterministic_working_front_matter(
+                        mode, topic, keywords, 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
+                    )
+                    candidate_abstract = str(fallback.get("abstract") or "").strip()
+                    candidate_keywords = list(fallback.get("keywords") or [])
                 _, remaining = self._repair_validate_study_fact_contract(
                     mode, agent, quant_engine, candidate_abstract, data_metadata, section_name="Working Abstract",
                 )
@@ -4501,6 +4983,12 @@ class GenerationRunner:
                     )
                     if residual_front_rows:
                         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
+                    )
+                    candidate_abstract = str(fallback.get("abstract") or "").strip()
+                    candidate_keywords = list(fallback.get("keywords") or candidate_keywords)
                 working_abstract = candidate_abstract
                 working_keywords = candidate_keywords or working_keywords
                 state["working_abstract"] = working_abstract
@@ -4698,16 +5186,33 @@ class GenerationRunner:
         # an outline but no Working Abstract yet. Generate it once, checkpoint it,
         # and reuse it thereafter; never regenerate an unchanged anchor on resume.
         if not working_abstract and not state.get("english_finalized"):
-            working_front = agent.generate_working_front_matter(
-                topic, keywords, 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),
-                keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
-                fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
-            )
+            try:
+                working_front = agent.generate_working_front_matter(
+                    topic, keywords, 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),
+                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
+                )
+            except PipelineCancelledError:
+                raise
+            except Exception as exc:
+                logging.warning(
+                    "Migrated Working Front Matter provider path unavailable; using deterministic fallback | error=%s",
+                    type(exc).__name__,
+                )
+                working_front = self._deterministic_working_front_matter(
+                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                )
+                quality_diagnostics.append({
+                    "scope": "delivery_first_working_front_matter_fallback",
+                    "name": "Working Abstract",
+                    "status": "deterministic_resume_migration_fallback",
+                    "reason": "provider_error",
+                })
             working_abstract = str(working_front.get("abstract") or "").strip()
-            if hypothesis_ledger:
+            if hypothesis_ledger and working_abstract:
                 working_abstract = self._normalize_hypothesis_ordinal_aliases(
                     working_abstract, hypothesis_ledger
                 )
@@ -4715,7 +5220,17 @@ class GenerationRunner:
                 mode, list(working_front.get("keywords") or []), data_metadata
             )
             if not working_abstract or not working_keywords:
-                raise RuntimeError("working_front_matter_missing")
+                working_front = self._deterministic_working_front_matter(
+                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                )
+                working_abstract = str(working_front.get("abstract") or "").strip()
+                working_keywords = list(working_front.get("keywords") or [])
+                quality_diagnostics.append({
+                    "scope": "delivery_first_working_front_matter_fallback",
+                    "name": "Working Abstract",
+                    "status": "deterministic_resume_migration_fallback",
+                    "reason": "missing_component",
+                })
             _, front_fact_rows = self._repair_validate_study_fact_contract(
                 mode, agent, quant_engine, working_abstract, data_metadata, section_name="Working Abstract",
             )
@@ -4725,22 +5240,32 @@ class GenerationRunner:
                 ))
             if front_fact_rows:
                 feedback = self._section_retry_feedback(front_fact_rows)
-                working_front = agent.generate_working_front_matter(
-                    topic, keywords, research_logic_contract, synthesis_result,
-                    data_evidence=planning_data_md, mode=mode, validation_feedback=feedback,
-                    hypothesis_contract=hypothesis_contract,
-                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
-                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
-                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
-                )
-                working_abstract = str(working_front.get("abstract") or "").strip()
-                if hypothesis_ledger:
-                    working_abstract = self._normalize_hypothesis_ordinal_aliases(
-                        working_abstract, hypothesis_ledger
+                try:
+                    working_front = agent.generate_working_front_matter(
+                        topic, keywords, 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),
+                        keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
+                        fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
+                    )
+                    retry_abstract = str(working_front.get("abstract") or "").strip()
+                    if hypothesis_ledger and retry_abstract:
+                        retry_abstract = self._normalize_hypothesis_ordinal_aliases(
+                            retry_abstract, hypothesis_ledger
+                        )
+                    retry_keywords = self._canonical_quantitative_front_matter_keywords(
+                        mode, list(working_front.get("keywords") or []), data_metadata
+                    )
+                    if retry_abstract and retry_keywords:
+                        working_abstract, working_keywords = retry_abstract, retry_keywords
+                except PipelineCancelledError:
+                    raise
+                except Exception as exc:
+                    logging.warning(
+                        "Migrated Working Front Matter validation retry unavailable; preserving deterministic-safe candidate | error=%s",
+                        type(exc).__name__,
                     )
-                working_keywords = self._canonical_quantitative_front_matter_keywords(
-                    mode, list(working_front.get("keywords") or []), data_metadata
-                )
                 _, remaining = self._repair_validate_study_fact_contract(
                     mode, agent, quant_engine, working_abstract, data_metadata, section_name="Working Abstract",
                 )
@@ -4761,8 +5286,13 @@ class GenerationRunner:
                     "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract",
                     "status":"section_retry_resolved", "count":len(front_fact_rows), "examples":front_fact_rows[:3],
                 })
-            if not working_abstract:
-                raise RuntimeError("working_front_matter_empty_after_validation")
+            if not working_abstract or not working_keywords:
+                working_front = self._deterministic_working_front_matter(
+                    mode, topic, keywords, data_metadata, hypothesis_ledger=hypothesis_ledger
+                )
+                working_abstract = str(working_front.get("abstract") or "").strip()
+                working_keywords = list(working_front.get("keywords") or [])
+                logging.warning("Delivery-first migrated Working Front Matter fallback activated after repair exhaustion.")
             agent.global_memory["working_abstract"] = working_abstract
             full_draft_en = replace_abstract(full_draft_en, working_abstract)
             full_draft_en = self._replace_keywords(full_draft_en, working_keywords)
@@ -4771,6 +5301,7 @@ class GenerationRunner:
                 "working_keywords": working_keywords,
                 "front_matter_phase": "working",
                 "full_draft_en": full_draft_en,
+                "quality_diagnostics": quality_diagnostics,
                 "agent_state": agent.export_state(),
             })
             self._publish_preview(full_draft_en, "working_front_matter", force=True)
@@ -5164,17 +5695,19 @@ Output only final academic prose. Do not report the word count."""
                 candidate_user_prompt = user_prompt
                 if retry_excluded_tags:
                     retry_claim_plan = agent.format_claim_evidence_plan(planned_items)
-                    candidate_user_prompt = re.sub(
+                    candidate_user_prompt = _regex_sub_literal(
                         r"<CLAIM_EVIDENCE_PLAN>.*?</CLAIM_EVIDENCE_PLAN>",
                         f"<CLAIM_EVIDENCE_PLAN>\n{retry_claim_plan}\n</CLAIM_EVIDENCE_PLAN>",
-                        candidate_user_prompt, flags=re.S,
+                        candidate_user_prompt,
+                        flags=re.S,
                     )
                     retry_capability_context = self._section_citation_capability_context(agent, planned_items)
-                    candidate_user_prompt = re.sub(
+                    candidate_user_prompt = _regex_sub_literal(
                         r"\n<TARGETED_RAG>.*?</TARGETED_RAG>\n",
                         (f"\n<TARGETED_RAG>\n{retry_capability_context}\n</TARGETED_RAG>\n"
                          if retry_capability_context else "\n"),
-                        candidate_user_prompt, flags=re.S,
+                        candidate_user_prompt,
+                        flags=re.S,
                     )
                 if attempt_feedback:
                     excluded_text = ", ".join(sorted(retry_excluded_tags, key=agent._citation_tag_number)) or "None"
@@ -5758,10 +6291,41 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     full_draft_en, hypothesis_ledger, section_name="Full Manuscript", require_all=True
                 )
                 if hypothesis_rows:
-                    self._raise_if_fatal_completion_residual(
-                        hypothesis_rows, section_name="Full Manuscript", mode=mode,
-                        error_code="final_hypothesis_contract_unresolved"
+                    reconciled, reconciliation_notes = self._deterministic_hypothesis_reconcile(
+                        full_draft_en, hypothesis_ledger
+                    )
+                    reconciled_rows = self._hypothesis_contract_violations(
+                        reconciled, hypothesis_ledger, section_name="Full Manuscript", require_all=True
                     )
+                    if reconciliation_notes:
+                        self._record_adaptive_completion_residual(
+                            quality_notes, reconciliation_notes, section_name="Full Manuscript",
+                            stage="delivery_first_hypothesis_reconciliation"
+                        )
+                    if reconciled_rows:
+                        self._raise_if_fatal_completion_residual(
+                            reconciled_rows, section_name="Full Manuscript", mode=mode,
+                            error_code="final_hypothesis_contract_unresolved"
+                        )
+                    else:
+                        full_draft_en = reconciled
+                        changed_sections = self._refresh_section_integrity_seals_after_delivery_repair(
+                            full_draft_en, section_order_for_fact, section_integrity_seals
+                        )
+                        if changed_sections:
+                            requeued = self._requeue_resealed_translation_sections(
+                                changed_sections=changed_sections, mode=mode, quant_engine=quant_engine,
+                                data_metadata=data_metadata, translation_agent=translation_agent,
+                                translation_sections=translation_sections, english_commit_text=full_draft_en,
+                            )
+                            self._record_adaptive_completion_residual(
+                                quality_notes, [{
+                                    "type": "delivery_first_section_resealed",
+                                    "count": len(changed_sections),
+                                    "requeued_translations": requeued,
+                                    "reason": "Deterministic H-identity repair changed sealed English bytes; section seals and target translation source authority were refreshed before finalization.",
+                                }], section_name="Full Manuscript", stage="delivery_first_hypothesis_reseal"
+                            )
                     self._record_adaptive_completion_residual(
                         quality_notes, hypothesis_rows, section_name="Full Manuscript", stage="final_hypothesis_review"
                     )
@@ -5874,13 +6438,32 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             # from the completed manuscript/evidence, then replace the provisional
             # front matter deterministically. No body rebalance may occur afterward.
             self._progress(92, "final_front_matter")
-            final_front = agent.generate_final_front_matter(
-                topic, keywords, 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),
-                fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
-            )
+            try:
+                final_front = agent.generate_final_front_matter(
+                    topic, keywords, 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),
+                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
+                )
+            except Exception as exc:
+                logging.warning(
+                    "Delivery-first Final Front Matter fallback | mode=%s error=%s", mode, type(exc).__name__
+                )
+                final_front = {
+                    "abstract": self._deterministic_final_abstract(
+                        mode, 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
+                    ) or self._front_matter_keyword_fallback_terms(data_metadata),
+                }
+                self._record_adaptive_completion_residual(
+                    quality_notes, [{
+                        "type": "final_front_matter_provider_fallback",
+                        "reason": f"Final Front Matter model/validation failed ({type(exc).__name__}); deterministic frozen-authority fallback was used."
+                    }], section_name="Final Abstract", stage="delivery_first_front_matter_fallback"
+                )
             final_abstract = str(final_front.get("abstract") or "").strip()
             if hypothesis_ledger:
                 final_abstract = self._normalize_hypothesis_ordinal_aliases(
@@ -5890,7 +6473,18 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 mode, list(final_front.get("keywords") or []), data_metadata
             )
             if not final_abstract or not final_keywords:
-                raise RuntimeError("final_front_matter_missing")
+                final_abstract = final_abstract or self._deterministic_final_abstract(
+                    mode, 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
+                ) or self._front_matter_keyword_fallback_terms(data_metadata)
+                self._record_adaptive_completion_residual(
+                    quality_notes, [{
+                        "type": "final_front_matter_missing_fallback",
+                        "reason": "Missing Final Front Matter component was reconstructed deterministically from frozen title/keyword/finding authority."
+                    }], section_name="Final Abstract", stage="delivery_first_front_matter_fallback"
+                )
             final_front_fact_rows = self._study_fact_violations(
                 mode, agent, quant_engine, final_abstract, data_metadata,
                 section_name="Final Abstract",
@@ -5909,13 +6503,26 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     mode, len(final_front_fact_rows),
                     sorted({str(row.get("type") or "") for row in final_front_fact_rows if isinstance(row, dict)}),
                 )
-                final_front = agent.generate_final_front_matter(
-                    topic, keywords, body_only, data_evidence=user_data_md, mode=mode,
-                    validation_feedback=feedback, hypothesis_contract=hypothesis_contract,
-                    authoritative_keywords=self._user_authoritative_front_matter_keywords(data_metadata),
-                    keyword_policy=self._resolved_front_matter_keyword_policy(data_metadata),
-                    fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
-                )
+                try:
+                    final_front = agent.generate_final_front_matter(
+                        topic, keywords, 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),
+                        fallback_keywords=self._front_matter_keyword_fallback_terms(data_metadata),
+                    )
+                except Exception as exc:
+                    logging.warning(
+                        "Delivery-first Final Front Matter retry fallback | mode=%s error=%s", mode, type(exc).__name__
+                    )
+                    final_front = {
+                        "abstract": self._deterministic_final_abstract(
+                            mode, 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
+                        ) or self._front_matter_keyword_fallback_terms(data_metadata),
+                    }
                 final_abstract = str(final_front.get("abstract") or "").strip()
                 if hypothesis_ledger:
                     final_abstract = self._normalize_hypothesis_ordinal_aliases(
@@ -5925,7 +6532,18 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     mode, list(final_front.get("keywords") or []), data_metadata
                 )
                 if not final_abstract or not final_keywords:
-                    raise RuntimeError("final_front_matter_missing_after_retry")
+                    final_abstract = final_abstract or self._deterministic_final_abstract(
+                        mode, 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
+                    ) or self._front_matter_keyword_fallback_terms(data_metadata)
+                    self._record_adaptive_completion_residual(
+                        quality_notes, [{
+                            "type": "final_front_matter_retry_exhausted_fallback",
+                            "reason": "Final Front Matter remained incomplete after retry; deterministic frozen-authority fallback was used."
+                        }], section_name="Final Abstract", stage="delivery_first_front_matter_fallback"
+                    )
                 remaining_final_fact_rows = self._study_fact_violations(
                     mode, agent, quant_engine, final_abstract, data_metadata,
                     section_name="Final Abstract",
@@ -5968,11 +6586,30 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                         quality_notes=quality_notes, label="Final Abstract",
                     )
                     if residual_final_rows:
-                        self._raise_if_fatal_completion_residual(
-                            residual_final_rows, section_name="Final Abstract", mode=mode,
-                            error_code="final_abstract_finding_contract_unresolved"
+                        deterministic_abstract = self._deterministic_final_abstract(
+                            mode, 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"
+                        )
+                        if hypothesis_ledger:
+                            hard_rows.extend(self._hypothesis_contract_violations(
+                                deterministic_abstract, hypothesis_ledger, section_name="Final Abstract"
+                            ))
+                        if hard_rows:
+                            self._raise_if_fatal_completion_residual(
+                                hard_rows, section_name="Final Abstract", mode=mode,
+                                error_code="final_abstract_integrity_unresolved"
+                            )
+                        final_abstract = deterministic_abstract
+                        self._record_adaptive_completion_residual(
+                            quality_notes, residual_final_rows, section_name="Final Abstract",
+                            stage="delivery_first_final_abstract_fallback"
+                        )
+                        logging.warning(
+                            "Delivery-first Final Abstract deterministic fallback accepted | residual_diagnostics=%s",
+                            len(residual_final_rows)
                         )
-                        logging.warning("Adaptive Final Abstract accepted with residual diagnostics | count=%s", len(residual_final_rows))
                 quality_diagnostics.append({
                     "scope":"final_abstract_study_fact_integrity", "name":"Final Abstract",
                     "status":"front_matter_retry_resolved", "count":len(final_front_fact_rows),
@@ -6008,10 +6645,41 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     full_draft_en, hypothesis_ledger, section_name="Full Manuscript", require_all=True
                 )
                 if post_front_hypothesis_rows:
-                    self._raise_if_fatal_completion_residual(
-                        post_front_hypothesis_rows, section_name="Full Manuscript", mode=mode,
-                        error_code="post_front_matter_hypothesis_contract_unresolved"
+                    reconciled, reconciliation_notes = self._deterministic_hypothesis_reconcile(
+                        full_draft_en, hypothesis_ledger
+                    )
+                    remaining_post_rows = self._hypothesis_contract_violations(
+                        reconciled, hypothesis_ledger, section_name="Full Manuscript", require_all=True
                     )
+                    if reconciliation_notes:
+                        self._record_adaptive_completion_residual(
+                            quality_notes, reconciliation_notes, section_name="Final Abstract",
+                            stage="delivery_first_post_front_hypothesis_reconciliation"
+                        )
+                    if remaining_post_rows:
+                        self._raise_if_fatal_completion_residual(
+                            remaining_post_rows, section_name="Full Manuscript", mode=mode,
+                            error_code="post_front_matter_hypothesis_contract_unresolved"
+                        )
+                    else:
+                        full_draft_en = reconciled
+                        changed_sections = self._refresh_section_integrity_seals_after_delivery_repair(
+                            full_draft_en, section_order_for_fact, section_integrity_seals
+                        )
+                        if changed_sections:
+                            requeued = self._requeue_resealed_translation_sections(
+                                changed_sections=changed_sections, mode=mode, quant_engine=quant_engine,
+                                data_metadata=data_metadata, translation_agent=translation_agent,
+                                translation_sections=translation_sections, english_commit_text=full_draft_en,
+                            )
+                            self._record_adaptive_completion_residual(
+                                quality_notes, [{
+                                    "type": "delivery_first_section_resealed",
+                                    "count": len(changed_sections),
+                                    "requeued_translations": requeued,
+                                    "reason": "Post-front-matter H-identity repair changed sealed English bytes; section seals and target translation source authority were refreshed before checkpointing.",
+                                }], section_name="Full Manuscript", stage="delivery_first_post_front_reseal"
+                            )
                     self._record_adaptive_completion_residual(
                         quality_notes, post_front_hypothesis_rows, section_name="Final Abstract",
                         stage="post_front_matter_hypothesis_review"
@@ -6387,6 +7055,57 @@ The previous attempt for this same manuscript unit was rejected by validation. R
             })
         return result
 
+    def build_recovery_artifact(self, error: Exception) -> dict[str, str]:
+        """Best-effort failed/refunded download when English authority is already frozen.
+
+        This method never changes task success semantics.  The Executor may expose
+        the returned ZIP only on a FAILED task after quota release.
+        """
+        if "cancel" in str(error or "").casefold():
+            return {}
+        latest_path = self.task_dir / "checkpoints" / "latest.json"
+        try:
+            payload = json.loads(latest_path.read_text(encoding="utf-8"))
+        except Exception:
+            return {}
+        if not isinstance(payload, dict) or payload.get("request_hash") != self.request_hash:
+            return {}
+        if not bool(payload.get("english_finalized")):
+            return {}
+        english = str(payload.get("full_draft_en") or "").strip()
+        if len(english) < 1500:
+            return {}
+        translated = str(payload.get("full_draft_translated") or "").strip()
+        target_lang = str(self.request.get("target_lang") or "").strip()
+        output_suffix = str(self.request.get("output_suffix") or "Target").strip()
+        target_file_language = (
+            "Chinese" if "chinese" in target_lang.casefold() or "中文" in target_lang
+            else "Russian" if "russian" in target_lang.casefold() or "рус" in target_lang.casefold()
+            else "Kazakh" if "kazakh" in target_lang.casefold() or "қазақ" in target_lang.casefold()
+            else re.sub(r"[^A-Za-z]+", "", output_suffix) or "Target"
+        )
+        try:
+            package = build_recovery_package(
+                english_markdown=english,
+                translated_markdown=translated,
+                english_filename="Paper_English.docx",
+                translated_filename=f"Paper_{target_file_language}_Partial.docx",
+                error_detail=str(error or "")[:1000],
+            )
+        except Exception as recovery_exc:
+            logging.warning(
+                "Delivery-first recovery package unavailable | error=%s original=%s",
+                type(recovery_exc).__name__, type(error).__name__,
+            )
+            return {}
+        bj_time = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d_%H%M")
+        output_filename = f"SmartPaper_Recovery_{bj_time}.zip"
+        pending = self.output_dir / (".pending_" + output_filename)
+        pending.write_bytes(package.zip_bytes)
+        final = self.output_dir / output_filename
+        os.replace(pending, final)
+        return {"output_filename": output_filename, "output_path": str(final)}
+
     def run(self) -> dict[str, Any]:
         try:
             return self._run_impl()

core/pipeline_agent.py

@@ -2931,8 +2931,17 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
                 future_lane[executor.submit(self._s2_worker, query, 0)] = 'method_authority'
                 future_lane[executor.submit(self._oa_worker, query, 1, "language:en,type:article,is_paratext:false")] = 'method_authority'
             for future in concurrent.futures.as_completed(future_lane):
-                results = future.result()
                 lane = future_lane[future]
+                try:
+                    results = future.result()
+                except PipelineCancelledError:
+                    raise
+                except Exception as exc:
+                    logging.warning(
+                        "Delivery-first METHOD_AUTHORITY source skipped | lane=%s error=%s",
+                        lane, type(exc).__name__,
+                    )
+                    continue
                 with self.lock:
                     for res in results:
                         result_keys = self._reference_keys(res)
@@ -2966,28 +2975,44 @@ If the immutable hypothesis contract is non-empty, preserve its H-identifiers an
         sys_prompt = """You are an expert academic librarian. Generate exactly 3 concise English search queries.
 [CRITICAL SEARCH QUERY RULES]
 1. DO NOT use complex, deeply nested boolean logic (e.g., Avoid `(A OR B) AND (C OR D)`).
-2. DO NOT use excessive quotation marks for long phrases. 
-3. Output simple, space-separated keyword combinations or simple AND logic. 
+2. DO NOT use excessive quotation marks for long phrases.
+3. Output simple, space-separated keyword combinations or simple AND logic.
    Good: "educational chatbot" "recommendation system" context-aware
    Bad: ("educational chatbot" OR "pedagogical agent") AND ("recommendation system")
 4. Keep the query strictly UNDER 5 terms to avoid triggering server API timeouts.
 Return ONLY a valid JSON array of strings."""
         user_prompt = f"Topic: {topic}\nKeywords: {keywords}"
         for attempt in range(3):
-            queries = self._extract_json_from_text(self.call_deepseek(
-                sys_prompt,
-                user_prompt,
-                temp=0.1 + attempt * 0.1,
-                model_role="fast",
-                logical_node='search_jit',
-            ))
+            try:
+                queries = self._extract_json_from_text(self.call_deepseek(
+                    sys_prompt,
+                    user_prompt,
+                    temp=0.1 + attempt * 0.1,
+                    model_role="fast",
+                    logical_node='search_jit',
+                ))
+            except PipelineCancelledError:
+                raise
+            except Exception as exc:
+                logging.warning(
+                    "Delivery-first search-query generation fallback | attempt=%s/3 error=%s",
+                    attempt + 1, type(exc).__name__,
+                )
+                queries = []
             if queries and isinstance(queries, list) and len(queries) > 0:
                 final_queries = [str(q).strip() for q in queries if self._is_english_dominant(str(q).strip())]
-                if final_queries: 
+                if final_queries:
                     return final_queries[:3]
             time.sleep(1)
-            
-        clean_chunks = [c.strip() for c in re.findall(r'[a-zA-Z0-9\s\-]{4,}', f"{topic} {keywords}") if self._is_english_dominant(c.strip())]
+
+        clean_chunks = [
+            c.strip() for c in re.findall(r'[a-zA-Z0-9\s\-]{4,}', f"{topic} {keywords}")
+            if self._is_english_dominant(c.strip())
+        ]
+        if clean_chunks:
+            logging.warning(
+                "Delivery-first search-query provider exhausted; using deterministic English user-text query fallback."
+            )
         return clean_chunks[:3] if clean_chunks else []
 
     @staticmethod
@@ -3418,7 +3443,18 @@ Return ONLY a valid JSON array of strings."""
                 for future in concurrent.futures.as_completed(future_sources):
                     self._check_cancelled()
                     source = future_sources[future]
-                    collected[source].extend(future.result())
+                    try:
+                        rows = future.result()
+                    except PipelineCancelledError:
+                        raise
+                    except Exception as exc:
+                        self._record_source_error(source, 'RETRIEVAL_FUTURE_ERROR', detail=exc)
+                        logging.warning(
+                            "Delivery-first literature source future skipped | source=%s error=%s",
+                            source, type(exc).__name__,
+                        )
+                        rows = []
+                    collected[source].extend(rows)
                     if progress_callback:
                         progress_callback()
 
@@ -3427,15 +3463,24 @@ Return ONLY a valid JSON array of strings."""
             if phase == 0:
                 terms = self.generate_search_queries(topic, keywords)
             else:
-                response = self.call_deepseek(
-                    'You are an academic librarian. Return only a JSON array with three new concise English search queries, each under five terms.',
-                    f"Topic: {topic}\nKeywords: {keywords}\nUsed: {sorted(all_used_terms)}",
-                    temp=0.4,
-                    model_role='fast',
-                    logical_node='search_jit',
-                )
-                parsed = self._extract_json_from_text(response)
-                terms = parsed if isinstance(parsed, list) else []
+                try:
+                    response = self.call_deepseek(
+                        'You are an academic librarian. Return only a JSON array with three new concise English search queries, each under five terms.',
+                        f"Topic: {topic}\nKeywords: {keywords}\nUsed: {sorted(all_used_terms)}",
+                        temp=0.4,
+                        model_role='fast',
+                        logical_node='search_jit',
+                    )
+                    parsed = self._extract_json_from_text(response)
+                    terms = parsed if isinstance(parsed, list) else []
+                except PipelineCancelledError:
+                    raise
+                except Exception as exc:
+                    logging.warning(
+                        'Delivery-first auxiliary search-query phase skipped | phase=%s error=%s',
+                        phase, type(exc).__name__,
+                    )
+                    terms = []
 
             normalized_terms = []
             for value in terms:

core/production_guard.py

@@ -33,7 +33,7 @@ def replace_abstract(manuscript: str, abstract: str) -> str:
     replacement = f"## Abstract\n{str(abstract or '').strip()}\n\n"
     updated, count = re.subn(
         r"(?ms)^##\s+Abstract\s*$\n.*?(?=^##\s+Keywords\s*$)",
-        replacement,
+        lambda _match, value=replacement: value,
         value,
         count=1,
     )
@@ -469,6 +469,65 @@ def find_ai_writing_trace_findings(text: str) -> tuple[dict[str, Any], ...]:
     return tuple(findings)
 
 
+def repair_markdown_for_delivery(text: str, *, language_name: str) -> tuple[str, list[dict[str, Any]]]:
+    """Deterministically remove noncritical process/meta leakage before packaging.
+
+    Unresolved citation identities are intentionally not repaired here because
+    they are P1 authority.  High-confidence authorial/process leakage is removed
+    at paragraph granularity and recorded outside the manuscript.
+    """
+    value = str(text or "").strip()
+    notes: list[dict[str, Any]] = []
+    findings = [
+        dict(item) for item in find_ai_writing_trace_findings(value)
+        if str(item.get("relation") or "") != "unresolved_citation"
+        and str(item.get("kind") or "") in {"prompt_leak", "meta"}
+    ]
+    if findings:
+        spans = []
+        for item in findings:
+            start = max(0, int(item.get("start") or 0))
+            end = min(len(value), int(item.get("end") or start))
+            block_start = value.rfind("\n\n", 0, start)
+            block_start = 0 if block_start < 0 else block_start + 2
+            block_end = value.find("\n\n", end)
+            block_end = len(value) if block_end < 0 else block_end
+            spans.append((block_start, block_end))
+        merged = []
+        for start, end in sorted(spans):
+            if merged and start <= merged[-1][1]:
+                merged[-1] = (merged[-1][0], max(merged[-1][1], end))
+            else:
+                merged.append((start, end))
+        for start, end in reversed(merged):
+            value = value[:start] + value[end:]
+        value = re.sub(r"\n{3,}", "\n\n", value).strip()
+        notes.append({
+            "severity": "ADVISORY",
+            "section": language_name,
+            "issue_type": "delivery_first_reader_facing_leak_removed",
+            "detail": (
+                f"{language_name} packaging removed {len(findings)} high-confidence internal/process or AI-meta item(s) "
+                "deterministically instead of terminating an otherwise complete manuscript."
+            ),
+            "user_action": "Review the affected location during final editorial polishing if desired.",
+        })
+    if value and not re.search(r"(?m)^#\s+", value):
+        lines = value.splitlines()
+        first_index = next((i for i, line in enumerate(lines) if line.strip()), None)
+        if first_index is not None:
+            lines[first_index] = "# " + re.sub(r"^#+\s*", "", lines[first_index]).strip()
+            value = "\n".join(lines).strip()
+            notes.append({
+                "severity": "ADVISORY",
+                "section": language_name,
+                "issue_type": "delivery_first_heading_reconstructed",
+                "detail": f"{language_name} manuscript had no Markdown heading marker; the first nonempty line was rebound as the title heading.",
+                "user_action": "Verify title styling before formal submission.",
+            })
+    return value, notes
+
+
 def validate_markdown_manuscript(text: str, *, language_name: str) -> None:
     value = str(text or "").strip()
     if len(value) < 1000:
@@ -672,68 +731,125 @@ def _manifest(entries: dict[str, bytes]) -> str:
     return "\n".join(rows) + "\n"
 
 
-def build_validated_package(
-    *,
-    english_markdown: str,
-    translated_markdown: str,
-    english_filename: str,
-    translated_filename: str,
-    data_metadata: dict[str, Any],
-    mode: str,
-    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,
-    organize_appendix: bool = False,
-) -> PackageResult:
-    validate_markdown_manuscript(english_markdown, language_name="English")
-    validate_markdown_manuscript(translated_markdown, language_name="Target-language")
+def _safe_markdown_docx(markdown: str) -> bytes:
+    """Independent minimal DOCX renderer for the P0 safe-rebuild path.
 
-    qualitative_mode = str(mode or "").strip().casefold() == "asq"
+    The ordinary renderer owns publication styling.  This fallback owns only
+    delivery: headings, paragraph text and Markdown tables are preserved in a
+    standards-compliant DOCX without advanced section/layout transformations.
+    It deliberately keeps tables so quantitative table-identity validation can
+    still run unchanged.
+    """
+    document = Document()
+    lines = str(markdown or "").replace("\r\n", "\n").replace("\r", "\n").split("\n")
+
+    def table_cells(line: str) -> list[str]:
+        value = str(line or "").strip()
+        if value.startswith("|"):
+            value = value[1:]
+        if value.endswith("|"):
+            value = value[:-1]
+        return [cell.strip() for cell in value.split("|")]
+
+    def is_separator(line: str) -> bool:
+        cells = table_cells(line)
+        return bool(cells) and all(bool(re.fullmatch(r":?-{3,}:?", cell.replace(" ", ""))) for cell in cells)
+
+    index = 0
+    while index < len(lines):
+        raw = lines[index]
+        line = raw.strip()
+        if not line:
+            index += 1
+            continue
+        heading = re.match(r"^(#{1,6})\s+(.+)$", line)
+        if heading:
+            document.add_heading(heading.group(2).strip(), level=min(6, len(heading.group(1))))
+            index += 1
+            continue
+        if (
+            "|" in line
+            and index + 1 < len(lines)
+            and "|" in lines[index + 1]
+            and is_separator(lines[index + 1])
+        ):
+            header = table_cells(line)
+            rows: list[list[str]] = []
+            index += 2
+            while index < len(lines):
+                candidate = lines[index].strip()
+                if not candidate or "|" not in candidate:
+                    break
+                if is_separator(candidate):
+                    index += 1
+                    continue
+                cells = table_cells(candidate)
+                if len(cells) != len(header):
+                    break
+                rows.append(cells)
+                index += 1
+            table = document.add_table(rows=1, cols=max(1, len(header)))
+            for col, cell in enumerate(header):
+                table.rows[0].cells[col].text = re.sub(r"[*_`]+", "", cell)
+            for row in rows:
+                cells = table.add_row().cells
+                for col, cell in enumerate(row):
+                    cells[col].text = re.sub(r"[*_`]+", "", cell)
+            continue
+        bullet = re.match(r"^[-*+]\s+(.+)$", line)
+        if bullet:
+            document.add_paragraph(re.sub(r"[*_`]+", "", bullet.group(1)), style="List Bullet")
+        else:
+            document.add_paragraph(re.sub(r"[*_`]+", "", line))
+        index += 1
+
+    stream = io.BytesIO()
+    document.save(stream)
+    return stream.getvalue()
+
+
+def _render_delivery_docx(markdown: str, *, qualitative_mode: bool, safe_mode: bool = False) -> bytes:
+    """Render normal publication DOCX, then use an independent safe renderer.
+
+    Safe mode changes presentation only: manuscript text and Markdown table
+    identity remain authoritative, while advanced Appendix/layout logic is
+    intentionally omitted.  Quantitative table validation still runs after the
+    safe rebuild, so P1 result identity cannot be bypassed.
+    """
+    if safe_mode:
+        try:
+            return _safe_markdown_docx(markdown)
+        except Exception as exc:
+            raise OutputValidationError(f"Safe DOCX rebuild failed: {exc}") from exc
     try:
-        english_docx = text_to_docx(
-            english_markdown, qualitative_appendix=qualitative_mode
-        ).getvalue()
-        translated_docx = text_to_docx(
-            translated_markdown, qualitative_appendix=qualitative_mode
+        return text_to_docx(
+            markdown,
+            qualitative_appendix=bool(qualitative_mode),
         ).getvalue()
     except DocxBuildError as exc:
         raise OutputValidationError(str(exc)) from exc
-    validate_docx_bytes(english_docx)
-    validate_docx_bytes(translated_docx)
-    quantitative_manifest = (data_metadata or {}).get("quantitative_table_manifest") if str(mode or "").strip().casefold() == "amj" else None
-    if quantitative_manifest:
-        validate_quantitative_table_contract(
-            english_markdown, english_docx, quantitative_manifest, label="English DOCX"
-        )
-        validate_quantitative_table_contract(
-            translated_markdown, translated_docx, quantitative_manifest, label="Target-language DOCX"
-        )
-    if qualitative_mode:
-        validate_qualitative_appendix_docx_layout(english_docx, label="English DOCX")
-        validate_qualitative_appendix_docx_layout(translated_docx, label="Target-language DOCX")
 
-    appendix_prefix = "Appendix/" if organize_appendix else ""
-    entries: dict[str, bytes] = {
-        english_filename: english_docx,
-        translated_filename: translated_docx,
-        appendix_prefix + "GENERATION_NOTES.txt": build_generation_notes(data_metadata, mode).encode("utf-8"),
-        appendix_prefix + "CITATION_AUDIT.json": json.dumps(citation_audit or [], ensure_ascii=False, indent=2).encode("utf-8"),
-        appendix_prefix + "LITERATURE_METRICS.json": json.dumps(literature_metrics or {}, ensure_ascii=False, indent=2).encode("utf-8"),
-        appendix_prefix + "WORD_COUNT_QUALITY_REPORT.txt": build_word_count_quality_report(quality_diagnostics).encode("utf-8"),
-        appendix_prefix + "SmartPaper_Quality_Notes.txt": build_quality_notes(quality_notes).encode("utf-8"),
-        appendix_prefix + "EVIDENCE_COVERAGE_REPORT.txt": build_evidence_coverage_report(evidence_coverage).encode("utf-8"),
-    }
-    manifest = _manifest(entries)
-    entries[appendix_prefix + "FILE_MANIFEST_SHA256.txt"] = manifest.encode("utf-8")
 
+def _build_zip_bytes(entries: dict[str, bytes]) -> bytes:
     buffer = io.BytesIO()
     with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED) as archive:
         for name, content in entries.items():
             archive.writestr(name, content)
-    zip_bytes = buffer.getvalue()
+    return buffer.getvalue()
 
+
+def _validate_zip_bytes(
+    zip_bytes: bytes,
+    *,
+    entries: dict[str, bytes],
+    english_filename: str,
+    translated_filename: str,
+    english_markdown: str,
+    translated_markdown: str,
+    quantitative_manifest: dict[str, Any] | None,
+    qualitative_mode: bool,
+    safe_mode: bool,
+) -> None:
     try:
         with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as archive:
             bad = archive.testzip()
@@ -754,7 +870,7 @@ def build_validated_package(
                         archived_markdown, archived_docx, quantitative_manifest,
                         label="Archived English DOCX" if name == english_filename else "Archived target-language DOCX",
                     )
-                if qualitative_mode:
+                if qualitative_mode and not safe_mode:
                     validate_qualitative_appendix_docx_layout(
                         archived_docx,
                         label="Archived English DOCX" if name == english_filename else "Archived target-language DOCX",
@@ -764,4 +880,261 @@ def build_validated_package(
     except Exception as exc:
         raise OutputValidationError(f"ZIP cannot be reopened: {exc}") from exc
 
-    return PackageResult(zip_bytes=zip_bytes, file_names=tuple(sorted(entries)), manifest=manifest)
+
+def _validated_package_attempt(
+    *,
+    english_markdown: str,
+    translated_markdown: str,
+    english_filename: str,
+    translated_filename: str,
+    data_metadata: dict[str, Any],
+    mode: str,
+    citation_audit: list[dict[str, Any]],
+    literature_metrics: dict[str, Any],
+    quality_diagnostics: list[dict[str, Any]] | None,
+    quality_notes: list[dict[str, Any]] | None,
+    evidence_coverage: dict[str, Any] | None,
+    organize_appendix: bool,
+    safe_mode: bool,
+) -> PackageResult:
+    qualitative_mode = str(mode or "").strip().casefold() == "asq"
+    english_docx = _render_delivery_docx(
+        english_markdown, qualitative_mode=qualitative_mode, safe_mode=safe_mode,
+    )
+    translated_docx = _render_delivery_docx(
+        translated_markdown, qualitative_mode=qualitative_mode, safe_mode=safe_mode,
+    )
+    validate_docx_bytes(english_docx)
+    validate_docx_bytes(translated_docx)
+
+    quantitative_manifest = (
+        (data_metadata or {}).get("quantitative_table_manifest")
+        if str(mode or "").strip().casefold() == "amj" else None
+    )
+    if quantitative_manifest:
+        # P1 identity gate remains strict in both normal and safe-format modes.
+        validate_quantitative_table_contract(
+            english_markdown, english_docx, quantitative_manifest, label="English DOCX"
+        )
+        validate_quantitative_table_contract(
+            translated_markdown, translated_docx, quantitative_manifest, label="Target-language DOCX"
+        )
+    if qualitative_mode and not safe_mode:
+        validate_qualitative_appendix_docx_layout(english_docx, label="English DOCX")
+        validate_qualitative_appendix_docx_layout(translated_docx, label="Target-language DOCX")
+
+    appendix_prefix = "Appendix/" if organize_appendix else ""
+    notes = list(quality_notes or [])
+    if safe_mode:
+        notes.append({
+            "severity": "ADVISORY",
+            "section": "Output package",
+            "issue_type": "delivery_first_safe_docx_rebuild",
+            "detail": (
+                "The normal rich-output package did not pass final output validation. "
+                "SmartPaper rebuilt the DOCX package in safe-format mode without changing manuscript text, evidence, or quantitative table identity."
+            ),
+            "user_action": "Review final page orientation and table layout before formal submission.",
+        })
+    entries: dict[str, bytes] = {
+        english_filename: english_docx,
+        translated_filename: translated_docx,
+        appendix_prefix + "GENERATION_NOTES.txt": build_generation_notes(data_metadata, mode).encode("utf-8"),
+        appendix_prefix + "CITATION_AUDIT.json": json.dumps(citation_audit or [], ensure_ascii=False, indent=2).encode("utf-8"),
+        appendix_prefix + "LITERATURE_METRICS.json": json.dumps(literature_metrics or {}, ensure_ascii=False, indent=2).encode("utf-8"),
+        appendix_prefix + "WORD_COUNT_QUALITY_REPORT.txt": build_word_count_quality_report(quality_diagnostics).encode("utf-8"),
+        appendix_prefix + "SmartPaper_Quality_Notes.txt": build_quality_notes(notes).encode("utf-8"),
+        appendix_prefix + "EVIDENCE_COVERAGE_REPORT.txt": build_evidence_coverage_report(evidence_coverage).encode("utf-8"),
+    }
+    manifest = _manifest(entries)
+    entries[appendix_prefix + "FILE_MANIFEST_SHA256.txt"] = manifest.encode("utf-8")
+
+    # Delivery-first ZIP recovery: a clean second build is mandatory before
+    # declaring output corruption terminal.
+    last_error: Exception | None = None
+    for _attempt in range(2):
+        zip_bytes = _build_zip_bytes(entries)
+        try:
+            _validate_zip_bytes(
+                zip_bytes,
+                entries=entries,
+                english_filename=english_filename,
+                translated_filename=translated_filename,
+                english_markdown=english_markdown,
+                translated_markdown=translated_markdown,
+                quantitative_manifest=quantitative_manifest,
+                qualitative_mode=qualitative_mode,
+                safe_mode=safe_mode,
+            )
+            return PackageResult(
+                zip_bytes=zip_bytes,
+                file_names=tuple(sorted(entries)),
+                manifest=manifest,
+            )
+        except OutputValidationError as exc:
+            last_error = exc
+    raise OutputValidationError(
+        f"ZIP validation failed after clean rebuild: {last_error}"
+    ) from last_error
+
+
+def build_validated_package(
+    *,
+    english_markdown: str,
+    translated_markdown: str,
+    english_filename: str,
+    translated_filename: str,
+    data_metadata: dict[str, Any],
+    mode: str,
+    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,
+    organize_appendix: bool = False,
+) -> PackageResult:
+    # Delivery-first cleanup removes only high-confidence process/meta leakage
+    # and repairs missing heading markup.  Citation identity remains fail-closed.
+    english_markdown, english_repair_notes = repair_markdown_for_delivery(
+        english_markdown, language_name="English"
+    )
+    translated_markdown, target_repair_notes = repair_markdown_for_delivery(
+        translated_markdown, language_name="Target-language"
+    )
+    merged_quality_notes = list(quality_notes or []) + english_repair_notes + target_repair_notes
+
+    # Content completeness/integrity remains P1/P0 authority and is never
+    # bypassed by safe-format packaging.
+    validate_markdown_manuscript(english_markdown, language_name="English")
+    validate_markdown_manuscript(translated_markdown, language_name="Target-language")
+
+    try:
+        return _validated_package_attempt(
+            english_markdown=english_markdown,
+            translated_markdown=translated_markdown,
+            english_filename=english_filename,
+            translated_filename=translated_filename,
+            data_metadata=data_metadata,
+            mode=mode,
+            citation_audit=citation_audit,
+            literature_metrics=literature_metrics,
+            quality_diagnostics=quality_diagnostics,
+            quality_notes=merged_quality_notes,
+            evidence_coverage=evidence_coverage,
+            organize_appendix=organize_appendix,
+            safe_mode=False,
+        )
+    except OutputValidationError as first_error:
+        # P0 delivery-first: presentation/build failures get one deterministic
+        # safe-format rebuild before the task may fail.  Quantitative table
+        # identity, DOCX reopenability and ZIP integrity remain strict in the
+        # second attempt.
+        try:
+            return _validated_package_attempt(
+                english_markdown=english_markdown,
+                translated_markdown=translated_markdown,
+                english_filename=english_filename,
+                translated_filename=translated_filename,
+                data_metadata=data_metadata,
+                mode=mode,
+                citation_audit=citation_audit,
+                literature_metrics=literature_metrics,
+                quality_diagnostics=quality_diagnostics,
+                quality_notes=merged_quality_notes,
+                evidence_coverage=evidence_coverage,
+                organize_appendix=organize_appendix,
+                safe_mode=True,
+            )
+        except OutputValidationError as second_error:
+            raise OutputValidationError(
+                "Validated package failed normal and safe-format rebuilds: "
+                f"normal={first_error}; safe={second_error}"
+            ) from second_error
+
+
+def _plain_recovery_docx(markdown: str) -> bytes:
+    """Last-resort readable DOCX for FAILED/refunded Recovery Packages only."""
+    document = Document()
+    for raw_line in str(markdown or "").replace("\r\n", "\n").replace("\r", "\n").split("\n"):
+        line = raw_line.strip()
+        if not line:
+            continue
+        heading = re.match(r"^(#{1,6})\s+(.+)$", line)
+        if heading:
+            document.add_heading(heading.group(2).strip(), level=min(6, len(heading.group(1))))
+        else:
+            document.add_paragraph(re.sub(r"[*_`]+", "", line))
+    stream = io.BytesIO()
+    document.save(stream)
+    return stream.getvalue()
+
+
+def build_recovery_package(
+    *,
+    english_markdown: str,
+    translated_markdown: str = "",
+    english_filename: str = "Paper_English.docx",
+    translated_filename: str = "Paper_Target_Partial.docx",
+    error_detail: str = "",
+) -> PackageResult:
+    """Build a non-success Recovery Package from already-frozen manuscript text.
+
+    This artifact is intentionally independent from the Success Contract.  It is
+    used only on FAILED/refunded tasks and therefore may omit an incomplete target
+    manuscript, but the English DOCX must still reopen successfully.
+    """
+    english = str(english_markdown or "").strip()
+    if len(english) < 1500:
+        raise OutputValidationError("Recovery package requires a materially complete frozen English manuscript.")
+    try:
+        english_docx = _render_delivery_docx(english, qualitative_mode=False, safe_mode=True)
+        validate_docx_bytes(english_docx)
+    except Exception:
+        english_docx = _plain_recovery_docx(english)
+        validate_docx_bytes(english_docx)
+
+    entries: dict[str, bytes] = {english_filename: english_docx}
+    target = str(translated_markdown or "").strip()
+    if len(target) >= 800:
+        try:
+            target_docx = _render_delivery_docx(target, qualitative_mode=False, safe_mode=True)
+            validate_docx_bytes(target_docx)
+        except Exception:
+            try:
+                target_docx = _plain_recovery_docx(target)
+                validate_docx_bytes(target_docx)
+            except Exception:
+                target_docx = b""
+        if target_docx:
+            entries[translated_filename] = target_docx
+
+    note = (
+        "SmartPaper Recovery Package\n\n"
+        "This package was created after the full Success Contract could not be completed.\n"
+        "The task must remain FAILED and any reserved quota must be released/refunded.\n"
+        "This recovery artifact must not be represented as a completed paid paper.\n\n"
+        f"Terminal detail: {str(error_detail or '')[:1000]}\n"
+    )
+    entries["RECOVERY_NOTES.txt"] = note.encode("utf-8")
+    manifest = _manifest(entries)
+    entries["FILE_MANIFEST_SHA256.txt"] = manifest.encode("utf-8")
+
+    last_error: Exception | None = None
+    for _attempt in range(2):
+        zip_bytes = _build_zip_bytes(entries)
+        try:
+            with zipfile.ZipFile(io.BytesIO(zip_bytes), "r") as archive:
+                bad = archive.testzip()
+                if bad:
+                    raise OutputValidationError(f"Recovery ZIP integrity failed at {bad}.")
+                validate_docx_bytes(archive.read(english_filename))
+            return PackageResult(
+                zip_bytes=zip_bytes,
+                file_names=tuple(sorted(entries)),
+                manifest=manifest,
+            )
+        except Exception as exc:
+            last_error = exc
+    raise OutputValidationError(
+        f"Recovery ZIP could not be validated after rebuild: {last_error}"
+    ) from last_error

core/quantitative_engine.py

@@ -270,7 +270,13 @@ class QuantitativeEngine:
         for source, target in sorted(
             dict(profile.get("phrases") or {}).items(), key=lambda row: len(row[0]), reverse=True
         ):
-            text = re.sub(re.escape(source), str(target), text, flags=re.I)
+            replacement = str(target)
+            text = re.sub(
+                re.escape(source),
+                lambda _match, value=replacement: value,
+                text,
+                flags=re.I,
+            )
         return text
 
     @classmethod

i18n.py

@@ -216,6 +216,8 @@ TEXTS: dict[str, dict[str, str]] = {
         "product_generation_not_active": "产品版任务生命周期尚未启用，请勿继续生成。",
         "success": "🎉 流水线推演完毕！双语文档已就绪。",
         "download": "📥 下载文件",
+        "recovery_download": "📥 下载恢复文件",
+        "recovery_notice": "本次任务未满足完整成功条件，额度将按失败规则处理。系统已保留可恢复的论文文件供下载；该恢复文件不视为成功生成的付费论文。",
         "system_busy_title": "⏳ 系统忙碌",
         "system_busy_warning": "🚨 有用户正在生成论文",
         "system_busy_info": "💡 预计25–35分钟后可正常使用",
@@ -450,6 +452,8 @@ TEXTS: dict[str, dict[str, str]] = {
         "product_generation_not_active": "The product-tier task lifecycle is not active yet. Generation has been blocked to protect your balance.",
         "success": "🎉 The pipeline is complete. The bilingual documents are ready.",
         "download": "📥 Download Files",
+        "recovery_download": "📥 Download Recovery Files",
+        "recovery_notice": "This task did not satisfy the full success contract, so quota is handled under the failure/refund rule. SmartPaper preserved recoverable manuscript files for download; this recovery package is not counted as a successfully generated paid paper.",
         "system_busy_title": "⏳ System Busy",
         "system_busy_warning": "🚨 Another user is generating a paper",
         "system_busy_info": "💡 The system is expected to be available in 25–35 minutes",
@@ -684,6 +688,8 @@ TEXTS: dict[str, dict[str, str]] = {
         "product_generation_not_active": "Жизненный цикл задач продуктовых версий ещё не активирован. Генерация заблокирована для защиты баланса.",
         "success": "🎉 Работа конвейера завершена. Двуязычные документы готовы.",
         "download": "📥 Скачать файлы",
+        "recovery_download": "📥 Скачать восстановленные файлы",
+        "recovery_notice": "Задача не выполнила полный контракт успешного завершения, поэтому квота обрабатывается по правилу сбоя/возврата. SmartPaper сохранил доступные для восстановления материалы статьи; этот пакет не считается успешно сгенерированной оплаченной статьёй.",
         "system_busy_title": "⏳ Система занята",
         "system_busy_warning": "🚨 Другой пользователь формирует статью",
         "system_busy_info": "💡 Ожидаемое время доступности — через 25–35 минут",
@@ -918,6 +924,8 @@ TEXTS: dict[str, dict[str, str]] = {
         "product_generation_not_active": "Өнімдік нұсқалардың тапсырма өмірлік циклі әлі іске қосылған жоқ. Балансты қорғау үшін генерация бұғатталды.",
         "success": "🎉 Жұмыс үдерісі аяқталды. Екі тілдегі құжаттар дайын.",
         "download": "📥 Файлдарды жүктеу",
+        "recovery_download": "📥 Қалпына келтіру файлдарын жүктеу",
+        "recovery_notice": "Бұл тапсырма толық сәтті аяқталу талаптарына сәйкес келмеді, сондықтан квота сәтсіздік/қайтару ережесі бойынша өңделеді. SmartPaper қалпына келтіруге болатын мақала материалдарын жүктеу үшін сақтады; бұл пакет сәтті жасалған ақылы мақала ретінде есептелмейді.",
         "system_busy_title": "⏳ Жүйе бос емес",
         "system_busy_warning": "🚨 Басқа пайдаланушы мақала дайындап жатыр",
         "system_busy_info": "💡 Жүйе шамамен 25–35 минуттан кейін қолжетімді болады",

task_executor.py

@@ -126,34 +126,69 @@ def _is_local_checkpoint_authority_failure(exc: Exception) -> bool:
     return isinstance(exc, RuntimeError) and str(exc or "").startswith("checkpoint_authority_")
 
 
-def _is_recoverable_generation_failure(exc: Exception) -> bool:
-    """Classify only post-claim execution failures.
-
-    This deliberately avoids importing the generation pipeline at module import
-    time. Provider/model exception class names are inspected only after
-    claim/bootstrap has succeeded, so future provider adapters do not require
-    executor imports or vendor-specific branching.
-    """
+def _generation_failure_family(exc: Exception) -> str:
+    """Classify post-claim execution failures for the P0 delivery-first contract."""
     if isinstance(exc, ExecutorApiError):
-        return exc.status_code == 429 or exc.status_code >= 500
+        return "TRANSPORT" if exc.status_code == 429 or exc.status_code >= 500 else "AUTH_POLICY_SECURITY"
     class_name = exc.__class__.__name__
+    text = str(exc or "").casefold()
+
     if class_name in {
-        "DeepSeekAuthenticationError", "DeepSeekBalanceError", "DeepSeekRequestError",
-        "ModelAuthenticationError", "ModelBalanceError", "ModelRequestError",
-        "ModelProviderUnavailableError", "ModelGatewayError",
+        "DeepSeekAuthenticationError", "DeepSeekBalanceError",
+        "ModelAuthenticationError", "ModelBalanceError",
     }:
-        return False
+        return "AUTH_POLICY_SECURITY"
+    stable_policy_markers = (
+        "trusted_product_policy", "trusted_product_route_", "request_hash_mismatch",
+        "policy_snapshot", "entitlement", "authentication", "balance is insufficient",
+        "rejected the request format (400)", "not found (404)", "parameters are invalid (422)",
+        "unknown or unconfigured deepseek model role", "no deepseek model configured",
+    )
+    if any(marker in text for marker in stable_policy_markers):
+        return "AUTH_POLICY_SECURITY"
     if isinstance(exc, (requests.exceptions.Timeout, requests.exceptions.ConnectionError)):
-        return True
-    if class_name == "DeepSeekAPIError":
-        return True
-    text = str(exc or "").lower()
+        return "TRANSPORT"
+    if isinstance(exc, requests.exceptions.HTTPError):
+        response = getattr(exc, "response", None)
+        status = int(getattr(response, "status_code", 0) or 0)
+        if status in {408, 409, 425, 429, 500, 502, 503, 504}:
+            return "TRANSPORT"
+    if class_name in {"ModelProviderUnavailableError", "DeepSeekAPIError"}:
+        return "TRANSPORT"
+    if class_name == "DeepSeekRequestError":
+        content_validation_markers = (
+            "final abstract", "front matter", "translation", "citation", "hypothesis",
+            "manuscript", "qualitative", "glossary", "marker mismatch", "semantic",
+            "must not contain", "empty after validation", "structured validation",
+        )
+        return "CONTENT_VALIDATION" if any(marker in text for marker in content_validation_markers) else "AUTH_POLICY_SECURITY"
+    if class_name == "ModelRequestError":
+        # Gateway request errors are trusted route/request-shape failures; provider
+        # availability has its own recoverable exception class.
+        return "AUTH_POLICY_SECURITY"
+    if class_name == "ModelGatewayError":
+        transient_markers = ("unavailable", "timeout", "timed out", "connection", "provider")
+        return "TRANSPORT" if any(marker in text for marker in transient_markers) else "CONTENT_VALIDATION"
     transient_markers = (
         "connection", "timed out", "timeout", "temporarily unavailable",
         "streaming failed after", "api failed after", "stream was interrupted",
         "insufficient eligible references",
     )
-    return any(marker in text for marker in transient_markers)
+    if any(marker in text for marker in transient_markers):
+        return "TRANSPORT"
+    return "TERMINAL_INTEGRITY"
+
+
+def _is_recoverable_generation_failure(exc: Exception) -> bool:
+    """Return True only for failures that should resume instead of refund/fail.
+
+    CONTENT_VALIDATION reaches this layer only after the runner's local repair/
+    fallback paths were unable to complete the current attempt.  A new executor
+    attempt is still preferable to destroying the paper because model output is
+    nondeterministic; stable auth/policy/request-shape faults remain terminal.
+    """
+    return _generation_failure_family(exc) in {"TRANSPORT", "CONTENT_VALIDATION"}
+
 
 
 class ExecutorApi:
@@ -694,6 +729,7 @@ def main() -> int:
             state.stop_heartbeat()
         return 2
 
+    runner = None
     try:
         # Heavy paper modules are deliberately imported only after D1 claim and
         # the local claim handshake. Any import/init failure is now observable
@@ -761,6 +797,15 @@ def main() -> int:
             logging.exception("[SP_EXECUTOR] recoverable_interruption task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
             return 75
         logging.exception("[SP_EXECUTOR] failed task_id=%s epoch=%s error=%s", args.task_id, epoch, _safe(exc, 240))
+        recovery = {}
+        if runner is not None:
+            try:
+                recovery = dict(runner.build_recovery_artifact(exc) or {})
+            except Exception as recovery_exc:
+                logging.warning(
+                    "[SP_EXECUTOR] recovery_package_failed task_id=%s error=%s",
+                    args.task_id, _safe(recovery_exc, 220),
+                )
         try:
             api.finish(
                 {
@@ -772,6 +817,7 @@ def main() -> int:
                     "safe_error_code": "E_GENERATION_FAILED",
                     "safe_error_detail": _safe(exc, 800),
                     "current_node": "generation_failed",
+                    "output_filename": str(recovery.get("output_filename") or "")[:240],
                 }
             )
         except Exception as finish_exc: