S0-R1-C5 Dual-Engine Final Quality Contracts

core/generation_runner.py

@@ -36,6 +36,7 @@ from utils.file_parser import extract_text_from_files
 ProgressCallback = Callable[..., None]
 
 CITATION_SEMANTIC_SURVIVAL_RATIO = 0.78
+EVIDENCE_COMPLETION_MAX_PER_TOP_LEVEL = 4
 CITATION_ROLE_RULES = {
     "introduction": (3, 180),
     "theory": (3, 180),
@@ -976,6 +977,220 @@ class GenerationRunner:
                 tags.add(f'[{token}]')
         return tags
 
+    @staticmethod
+    def _author_facing_internal_workflow_violations(text: str) -> list[dict]:
+        """Detect SmartPaper control vocabulary that must stay out of author prose.
+
+        The check is intentionally phrase-scoped.  A generic word such as
+        ``frozen`` can be legitimate academic content, but phrases naming the
+        private evidence/state machinery are never reader-facing manuscript
+        language.
+        """
+        value = str(text or "")
+        patterns = [
+            r"\bfrozen\s+(?:research\s+)?evidence\b",
+            r"\bfrozen\s+evidence\s+packet\b",
+            r"\bevidence\s+packet\b",
+            r"\bfrozen\s+(?:theme\s+summary|thematic\s+structure|statistical\s+evidence|measurement\s+design|association\s+model|bootstrap\s+and\s+collinearity\s+checks|task-level\s+concepts?)\b",
+            r"\b(?:study|result)-owned\s+fact\b",
+            r"\bevidence\s+ledger\b",
+        ]
+        rows = []
+        for pattern in patterns:
+            for match in re.finditer(pattern, value, flags=re.I):
+                start = max(0, match.start() - 80)
+                end = min(len(value), match.end() + 120)
+                preview = re.sub(r"\s+", " ", value[start:end]).strip()
+                rows.append({
+                    "type": "author_facing_internal_workflow_language",
+                    "preview": preview[:320],
+                    "reason": "Private SmartPaper evidence/state vocabulary surfaced in reader-facing manuscript prose.",
+                    "repair_instruction": (
+                        "Rewrite the sentence as ordinary academic prose describing the study, evidence, method, or limitation directly. "
+                        "Do not mention frozen evidence, evidence packets, ledgers, study-owned/result-owned facts, or other orchestration labels."
+                    ),
+                })
+                if len(rows) >= 6:
+                    return rows
+        return rows
+
+    @staticmethod
+    def _hypothesis_contract_directive(ledger: dict) -> str:
+        record = dict(ledger or {})
+        items = [item for item in (record.get("items") or []) if isinstance(item, dict)]
+        if not items:
+            return ""
+        lines = []
+        for item in items:
+            hid = str(item.get("hypothesis_id") or "").strip()
+            predictor = str(item.get("predictor") or "").strip()
+            outcome = str(item.get("outcome") or "").strip()
+            if hid and predictor and outcome:
+                lines.append(f"{hid}: {predictor} is associated with {outcome}.")
+        if not lines:
+            return ""
+        indirect = dict(record.get("indirect_analysis") or {})
+        indirect_line = ""
+        if indirect:
+            p = str(indirect.get("predictor") or "").strip()
+            m = str(indirect.get("mediator") or "").strip()
+            o = str(indirect.get("outcome") or "").strip()
+            if p and m and o:
+                indirect_line = (
+                    f"The estimated indirect association {p} -> {m} -> {o} is an analysis target, not a separately numbered hypothesis."
+                )
+        rendered = " ".join(lines + ([indirect_line] if indirect_line else []))
+        return (
+            "IMMUTABLE HYPOTHESIS CONTRACT: " + rendered
+            + " Use H1/H2/H3-style identifiers exactly as listed whenever hypotheses are defined or referenced. "
+              "Do not renumber, reorder, create an additional hypothesis, replace an H-label with 'first/second/third hypothesis', "
+              "or change the construct pairing to fit observed results. The contract is result-independent."
+        )
+
+    @classmethod
+    def _hypothesis_contract_violations(
+        cls, text: str, ledger: dict, *, section_name: str = "", require_all: bool = False
+    ) -> list[dict]:
+        record = dict(ledger or {})
+        items = [item for item in (record.get("items") or []) if isinstance(item, dict)]
+        if not items:
+            return []
+        value = str(text or "")
+        expected = {
+            str(item.get("hypothesis_id") or "").upper(): item
+            for item in items if str(item.get("hypothesis_id") or "").strip()
+        }
+        if not expected:
+            return []
+        rows = []
+        observed = {f"H{n}" for n in re.findall(r"\bH(\d{1,2})\b", value, flags=re.I)}
+        unknown = sorted(observed - set(expected))
+        if unknown:
+            rows.append({
+                "type": "hypothesis_contract_unknown_id",
+                "observed": unknown,
+                "expected": sorted(expected),
+                "reason": "The manuscript introduced a hypothesis identifier outside the frozen result-independent hypothesis ledger.",
+                "repair_instruction": "Use only the frozen H-identifiers and keep the indirect association unnumbered unless the ledger explicitly numbers it.",
+            })
+        ordinal = re.search(r"\b(?:first|second|third|fourth|fifth|sixth)\s+hypothesis\b", value, flags=re.I)
+        if ordinal:
+            rows.append({
+                "type": "hypothesis_contract_ordinal_alias",
+                "preview": re.sub(r"\s+", " ", value[max(0, ordinal.start()-80):ordinal.end()+120]).strip()[:320],
+                "reason": "Ordinal hypothesis aliases can silently reorder the canonical H1-Hn identity.",
+                "repair_instruction": "Refer to hypotheses only by their frozen H1-Hn identifiers.",
+            })
+        definition_scope = bool(re.search(r"Hypothesis Development|Integrated Research Model", str(section_name or ""), flags=re.I))
+        if definition_scope:
+            all_constructs = {
+                re.sub(r"\s+", " ", str(item.get(key) or "")).strip()
+                for item in items for key in ("predictor", "outcome")
+                if str(item.get(key) or "").strip()
+            }
+            for start, end, sentence in SSciPipelineAgent._sentence_spans(value):
+                ids = list(dict.fromkeys(
+                    f"H{n}" for n in re.findall(r"\bH(\d{1,2})\b", sentence, flags=re.I)
+                ))
+                # Collective references such as "H1-H3 are summarized below"
+                # are not definitions and must not be mistaken for remapping.
+                if len(ids) != 1:
+                    continue
+                hid = ids[0]
+                item = expected.get(hid)
+                if not item:
+                    continue
+                predictor = re.sub(r"\s+", " ", str(item.get("predictor") or "")).strip()
+                outcome = re.sub(r"\s+", " ", str(item.get("outcome") or "")).strip()
+                lower_sentence = re.sub(r"\s+", " ", sentence).casefold()
+                mentioned_constructs = {
+                    term for term in all_constructs if term.casefold() in lower_sentence
+                }
+                # Validate pairing only when the sentence actually names one or
+                # more constructs. A bare cross-reference ("H1 is summarized")
+                # remains legal and cannot silently redefine the hypothesis.
+                if mentioned_constructs and predictor and outcome and (
+                    predictor.casefold() not in lower_sentence or outcome.casefold() not in lower_sentence
+                ):
+                    rows.append({
+                        "type": "hypothesis_contract_pairing_mismatch",
+                        "id": hid,
+                        "sentence": re.sub(r"\s+", " ", sentence).strip()[:420],
+                        "expected": f"{predictor} -> {outcome}",
+                        "reason": "A canonical hypothesis label was attached to a different construct pairing.",
+                        "repair_instruction": f"Define {hid} only for {predictor} and {outcome}, using association language.",
+                    })
+        if require_all:
+            missing = sorted(set(expected) - observed)
+            if missing:
+                rows.append({
+                    "type": "hypothesis_contract_missing_id",
+                    "observed": sorted(observed),
+                    "expected": sorted(expected),
+                    "reason": "The completed hypothesis architecture omitted one or more frozen hypothesis identifiers.",
+                    "repair_instruction": "Preserve every canonical H1-Hn identity from theory through the integrated model and downstream reporting.",
+                })
+        return rows[:8]
+
+    @staticmethod
+    def _strip_model_markdown_tables(text: str) -> str:
+        """Remove writer-created Markdown tables before deterministic AMJ table binding."""
+        lines = str(text or "").splitlines()
+        kept = []
+        index = 0
+        while index < len(lines):
+            line = lines[index]
+            if re.match(r"^\s*(?:#{1,6}\s*)?Table\s+\d+\s*[:.\-]", line, flags=re.I):
+                index += 1
+                while index < len(lines) and not lines[index].strip():
+                    index += 1
+                if index < len(lines) and lines[index].lstrip().startswith("|"):
+                    while index < len(lines) and lines[index].lstrip().startswith("|"):
+                        index += 1
+                if index < len(lines) and re.match(r"^\s*\*?Note\s*:", lines[index], flags=re.I):
+                    index += 1
+                continue
+            if line.lstrip().startswith("|"):
+                start = index
+                while index < len(lines) and lines[index].lstrip().startswith("|"):
+                    index += 1
+                block = lines[start:index]
+                if len(block) >= 2 and re.match(r"^\s*\|?\s*:?-{3,}", block[1]):
+                    continue
+                kept.extend(block)
+                continue
+            kept.append(line)
+            index += 1
+        return re.sub(r"\n{3,}", "\n\n", "\n".join(kept)).strip()
+
+    @classmethod
+    def _apply_quantitative_table_contract(cls, text: str, data_metadata: dict, display_name: str) -> str:
+        manifest = dict((data_metadata or {}).get("quantitative_table_manifest") or {})
+        tables = [item for item in (manifest.get("tables") or []) if isinstance(item, dict)]
+        if not tables:
+            return str(text or "")
+        value = cls._strip_model_markdown_tables(text)
+        assigned = []
+        for item in tables:
+            target = str(item.get("target_subsection") or "").strip()
+            if target and target.casefold() in str(display_name or "").casefold():
+                assigned.append(item)
+        if not assigned:
+            return value
+        blocks = []
+        for item in sorted(assigned, key=lambda row: int(row.get("number") or 0)):
+            number = int(item.get("number") or 0)
+            title = str(item.get("title") or "").strip()
+            markdown = str(item.get("markdown") or "").strip()
+            note = str(item.get("note") or "").strip()
+            if not number or not title or not markdown:
+                raise RuntimeError(f"quantitative_table_manifest_invalid:{display_name}")
+            block = f"Table {number}: {title}\n{markdown}"
+            if note:
+                block += f"\n\n*Note: {note}*"
+            blocks.append(block)
+        return (value.rstrip() + "\n\n" + "\n\n".join(blocks)).strip()
+
     @staticmethod
     def _snapshot_section_citation_state(agent, citation_plan: dict, unit_name: str) -> dict:
         """Snapshot only Section-owned citation state so a rejected attempt cannot leak forward."""
@@ -1007,7 +1222,7 @@ class GenerationRunner:
     def _section_validation_bundle(
         cls, mode: str, agent, quant_engine, text: str, data_metadata: dict, *,
         section_id: str, section_name: str, evidence_contract: str = "", include_critic: bool = True,
-        citation_violations=None,
+        citation_violations=None, hypothesis_ledger=None,
     ) -> list[dict]:
         """Collect typed S0-R1 validation feedback without mutating manuscript prose."""
         value = str(text or "")
@@ -1027,6 +1242,12 @@ class GenerationRunner:
         violations.extend(cls._study_fact_violations(
             mode, agent, quant_engine, value, data_metadata, section_name=section_name
         ))
+        violations.extend(cls._author_facing_internal_workflow_violations(value))
+        if str(mode or "").lower() == "amj":
+            violations.extend(cls._hypothesis_contract_violations(
+                value, hypothesis_ledger or {}, section_name=section_name,
+                require_all=bool(re.search(r"Integrated Research Model", str(section_name or ""), flags=re.I)),
+            ))
         violations.extend(copy.deepcopy(list(citation_violations or [])))
         return violations
 
@@ -1952,8 +2173,10 @@ class GenerationRunner:
                 "external_claims": external_claims,
                 "planning_minimum": int(planned_minimum),
             })
+        normalized_status = str(status or "adequate")
         return {
-            "status": str(status or "adequate"),
+            "status": normalized_status,
+            "submission_readiness": "review_required" if normalized_status == "limited" else "quality_gate_passed",
             "early_risk_detected": bool(early_risk),
             "final_reference_count": max(0, int(final_reference_count or 0)),
             "sections": sections,
@@ -1964,6 +2187,55 @@ class GenerationRunner:
             "events": list(events or []),
         }
 
+    def _bounded_top_level_evidence_completion(
+        self, agent, *, topic: str, section_id: str, section_name: str, section_text: str,
+        required: int, existing_tags, excluded_tags=None,
+    ) -> tuple[str, list[str]]:
+        """Give one completed top-level section a single bounded evidence top-up.
+
+        This is not a prose-generation path.  It issues at most one contextual JIT
+        retrieval and may attach only semantically validated citations to claims
+        that already exist.  If no directly supporting source is found, the prose
+        remains unchanged and the manuscript later reports evidence-limited quality.
+        """
+        value = str(section_text or "")
+        current = {str(tag or "").strip() for tag in (existing_tags or []) if str(tag or "").strip()}
+        needed = max(0, int(required or 0) - len(current))
+        needed = min(EVIDENCE_COMPLETION_MAX_PER_TOP_LEVEL, needed)
+        if needed <= 0 or not value.strip():
+            return value, []
+        before = set(getattr(agent, "used_citations", set()) or set())
+        items = agent.fetch_contextual_jit_recovery_items(
+            topic=topic,
+            section_name=section_name,
+            section_id=section_id,
+            section_context=value,
+            needed=needed,
+            excluded_tags=(before | set(excluded_tags or set())),
+        )
+        if not items:
+            logging.info(
+                "Bounded evidence completion found no eligible source | section=%s needed=%s",
+                section_name, needed,
+            )
+            return value, []
+        recovered = agent.recover_citations_on_existing_claims(
+            value, items, section_name=section_name, section_id=section_id, needed=needed
+        )
+        after = set(getattr(agent, "used_citations", set()) or set())
+        added = sorted(after - before, key=agent._citation_tag_number)
+        if recovered == value or not added:
+            logging.info(
+                "Bounded evidence completion made no validated attachment | section=%s candidates=%s",
+                section_name, len(items),
+            )
+            return value, []
+        logging.info(
+            "Bounded evidence completion accepted | section=%s added=%s required=%s",
+            section_name, len(added), required,
+        )
+        return recovered, added
+
     @staticmethod
     def _paragraph_token_set(value):
         text = re.sub(r'\[CITE-\d+\]', ' ', str(value or ''), flags=re.I)
@@ -2147,6 +2419,35 @@ class GenerationRunner:
                 )
                 self._checkpoint("quantitative_evidence_ledger_upgraded", state, 6)
 
+            if str(data_metadata.get("source_type") or "") == "system_generated":
+                table_manifest = dict(data_metadata.get("quantitative_table_manifest") or {})
+                expected_table_version = str(getattr(quant_engine, "QUANTITATIVE_TABLE_MANIFEST_VERSION", ""))
+                if table_manifest.get("version") != expected_table_version:
+                    rebuilt_tables = quant_engine.build_table_manifest_from_evidence_markdown(user_data_md)
+                    if not rebuilt_tables:
+                        raise RuntimeError("quantitative_table_manifest_rebuild_failed")
+                    data_metadata["quantitative_table_manifest"] = rebuilt_tables
+                    logging.warning(
+                        "Quantitative table manifest contract advanced; rebuilt locally version=%s",
+                        expected_table_version,
+                    )
+
+                hypothesis_ledger = dict(data_metadata.get("hypothesis_ledger") or {})
+                expected_hypothesis_version = str(getattr(quant_engine, "HYPOTHESIS_LEDGER_VERSION", ""))
+                if hypothesis_ledger.get("version") != expected_hypothesis_version:
+                    constructs = dict(data_metadata.get("constructs") or {})
+                    rebuilt_hypotheses = quant_engine.build_hypothesis_ledger([
+                        constructs.get("predictor"), constructs.get("mediator"), constructs.get("outcome")
+                    ])
+                    if not rebuilt_hypotheses:
+                        raise RuntimeError("quantitative_hypothesis_ledger_rebuild_failed")
+                    data_metadata["hypothesis_ledger"] = rebuilt_hypotheses
+                    logging.warning(
+                        "Quantitative hypothesis ledger contract advanced; rebuilt locally version=%s",
+                        expected_hypothesis_version,
+                    )
+                state["data_metadata"] = data_metadata
+
         self._qualitative_quote_sources = (
             self._qualitative_quote_source_map(data_metadata) if mode == "asq" else {}
         )
@@ -2643,6 +2944,27 @@ class GenerationRunner:
             state['canonical_concept_ledger'] = canonical_concept_ledger
             self._checkpoint('canonical_concept_ledger_ready', state, 15)
 
+        glossary_quality_version = str(
+            getattr(translation_agent, 'CANONICAL_GLOSSARY_QUALITY_VERSION', 'canonical-glossary-quality-v1')
+            or 'canonical-glossary-quality-v1'
+        )
+        if state.get('canonical_glossary_quality_version') != glossary_quality_version:
+            reviewed_glossary = translation_agent.validate_and_refine_canonical_concept_glossary(
+                canonical_concept_ledger, target_lang, translation_glossary
+            )
+            if reviewed_glossary != translation_glossary:
+                translation_glossary = dict(reviewed_glossary)
+                # A corrected canonical identity invalidates only target-language
+                # work. Sealed English remains authoritative and unchanged.
+                translation_sections = {}
+                state['translation_sections'] = {}
+                state['full_draft_translated'] = ''
+                state['translation_complete'] = False
+                state['package_validated'] = False
+            state['translation_glossary'] = translation_glossary
+            state['canonical_glossary_quality_version'] = glossary_quality_version
+            self._checkpoint('canonical_glossary_quality_ready', state, 15)
+
         current_entity_ledger_version = str(
             getattr(translation_agent, 'PROTECTED_ENTITY_LEDGER_VERSION', 'protected-entity-ledger-v1')
             or 'protected-entity-ledger-v1'
@@ -2807,6 +3129,7 @@ class GenerationRunner:
                 quote_invocation_plan, adaptive_structure, data_metadata
             )
 
+        hypothesis_ledger = dict(data_metadata.get("hypothesis_ledger") or {}) if mode == "amj" else {}
         total_steps = len(adaptive_structure)
         for sec in adaptive_structure:
             display_name = sec["display_name"]
@@ -2890,10 +3213,16 @@ class GenerationRunner:
                     " Use level-2 or level-3 headings only when they clarify the argument; "
                     "never use level-4 headings or numbered headings."
                 )
-            table_rule = (
-                "Present only frozen statistical evidence in standard Markdown tables with plain-text captions."
-                if sec["is_method_or_result"] and mode == "amj" else ""
-            )
+            if mode == "amj" and section_id == "results" and data_metadata.get("quantitative_table_manifest"):
+                table_rule = (
+                    "AUTHORITATIVE TABLE CONTRACT: Do not create, reproduce, renumber, or alter any Markdown statistical table. "
+                    "The Executor will attach the required reader-facing tables deterministically from the quantitative evidence manifest. "
+                    "You may discuss and refer to Table 1-Table 4 where relevant, but do not invent an additional table or change a table number."
+                )
+            elif sec["is_method_or_result"] and mode == "amj":
+                table_rule = "Present only study-supported statistical evidence; do not invent tables or numerical results."
+            else:
+                table_rule = ""
             concepts = ", ".join(sec.get("concepts") or [])
             theme_evidence = [item for item in (sec.get("theme_evidence") or []) if isinstance(item, dict)]
             evidence_profiles = []
@@ -2973,6 +3302,8 @@ Planning length: approximately {int(adjusted_target * min_ratio)}-{int(adjusted_
 {heading_directive}
 {citation_rule}
 {table_rule}
+AUTHOR-FACING PURITY: Write only reader-facing scholarly prose. Never mention SmartPaper control labels such as frozen evidence, evidence packet, evidence ledger, study-owned/result-owned fact, frozen theme summary, frozen thematic structure, or frozen task-level concepts. Describe the study, method, evidence, and limitations directly.
+{self._hypothesis_contract_directive(hypothesis_ledger)}
 {str(sec.get('required_content') or '')}
 {qualitative_rule}
 {anti_repeat}
@@ -3117,7 +3448,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     section_id=section_id, section_name=display_name,
                     evidence_contract=evidence_packet,
                     include_critic=not deterministic_appendix,
-                    citation_violations=citation_semantic_rows,
+                    citation_violations=citation_semantic_rows, hypothesis_ledger=hypothesis_ledger,
                 )
                 non_citation_count = max(
                     0, len(phase1_validation_rows) - len(citation_semantic_rows)
@@ -3219,6 +3550,17 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                         quote_invocation_plan=quote_invocation_plan,
                     )
 
+                if mode == "amj" and data_metadata.get("quantitative_table_manifest"):
+                    if section_id == "results":
+                        en_resolved = self._apply_quantitative_table_contract(
+                            en_resolved, data_metadata, display_name
+                        )
+                    else:
+                        # System-generated AMJ tables have one authoritative
+                        # manifest. Writer-created tables anywhere else are not
+                        # permitted to create a fifth/competing artifact table.
+                        en_resolved = self._strip_model_markdown_tables(en_resolved)
+
                 # S0-R1 phase 2: only local/idempotent structural normalizers may
                 # mutate the candidate before the final pure validation bundle.
                 en_resolved, enumeration_rows = agent.repair_and_validate_enumeration_contract(
@@ -3268,6 +3610,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                     mode, agent, quant_engine, en_resolved, data_metadata,
                     section_id=section_id, section_name=display_name,
                     evidence_contract=evidence_packet, include_critic=False,
+                    hypothesis_ledger=hypothesis_ledger,
                 )
                 if post_validation_rows:
                     if section_attempt < max_section_attempts:
@@ -3345,6 +3688,46 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 frozen_section = self._extract_top_level_section(full_draft_en, sec_title)
                 if not frozen_section:
                     raise RuntimeError(f"english_section_seal_source_missing:{sec_title}")
+
+                # C5: one bounded publication-quality evidence opportunity is
+                # allowed before the English top-level Section is sealed and
+                # before translation begins. It never writes new prose and may
+                # attach only citations that survive the same semantic auditor.
+                if section_key in citation_top_level_minimums:
+                    top_units = [
+                        unit for unit, owner in citation_unit_section_map.items()
+                        if str(owner) == str(section_key)
+                    ]
+                    top_validated = set()
+                    top_claims = 0
+                    for unit in top_units:
+                        top_validated.update(citation_validated_tags_by_unit.get(unit) or [])
+                        top_claims += int(citation_required_claim_counts_by_unit.get(unit, 0) or 0)
+                    top_required = min(
+                        int(citation_top_level_minimums.get(section_key, 0) or 0),
+                        max(0, top_claims),
+                    ) if top_claims else 0
+                    if len(top_validated) < top_required:
+                        completed_section, added_tags = self._bounded_top_level_evidence_completion(
+                            agent, topic=topic, section_id=section_key, section_name=sec_title,
+                            section_text=frozen_section, required=top_required,
+                            existing_tags=top_validated, excluded_tags=set(agent.used_citations),
+                        )
+                        if added_tags and completed_section != frozen_section:
+                            full_draft_en = self._replace_top_level_section(
+                                full_draft_en, sec_title, completed_section
+                            )
+                            frozen_section = completed_section
+                            merged_tags = set(citation_validated_tags_by_unit.get(display_name) or [])
+                            merged_tags.update(added_tags)
+                            citation_validated_tags_by_unit[display_name] = sorted(
+                                merged_tags, key=agent._citation_tag_number
+                            )
+                            logging.info(
+                                "Top-level evidence completion sealed | section=%s added=%s",
+                                sec_title, len(added_tags),
+                            )
+
                 section_integrity_seals[section_key] = self._build_section_integrity_seal(
                     section_key, sec_title, frozen_section
                 )
@@ -3424,6 +3807,17 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 agent.validate_enumeration_contract(
                     sealed_section, section_name=f"Final sealed section: {section_title}"
                 )
+                internal_rows = self._author_facing_internal_workflow_violations(sealed_section)
+                if internal_rows:
+                    raise RuntimeError(
+                        f"final_author_facing_purity_failed:{section_title}:{internal_rows[:3]}"
+                    )
+            if mode == "amj" and hypothesis_ledger:
+                hypothesis_rows = self._hypothesis_contract_violations(
+                    full_draft_en, hypothesis_ledger, section_name="Full Manuscript", require_all=True
+                )
+                if hypothesis_rows:
+                    raise RuntimeError(f"final_hypothesis_contract_failed:{hypothesis_rows[:3]}")
             # Validate the completed body while the Working Abstract is still in
             # place. Final front matter must not be generated until all body
             # sections, citation cleanup, and any Discussion rebalance are frozen.
@@ -3762,6 +4156,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                 if str(state.get("evidence_status") or "") == "limited":
                     result.update({
                         "quality_status": "evidence_limited",
+                        "submission_readiness": "review_required",
                         "safe_error_code": "W_EVIDENCE_COVERAGE_LIMITED",
                         "safe_error_detail": self._evidence_warning_detail(),
                     })
@@ -3807,6 +4202,7 @@ The previous attempt for this same manuscript unit was rejected by validation. R
         if str(state.get("evidence_status") or "") == "limited":
             result.update({
                 "quality_status": "evidence_limited",
+                "submission_readiness": "review_required",
                 "safe_error_code": "W_EVIDENCE_COVERAGE_LIMITED",
                 "safe_error_detail": self._evidence_warning_detail(),
             })

core/pipeline_agent.py

@@ -73,6 +73,7 @@ class SSciPipelineAgent:
     TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v8'
     MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS = 2
     PROTECTED_ENTITY_LEDGER_VERSION = 'protected-entity-ledger-v2'
+    CANONICAL_GLOSSARY_QUALITY_VERSION = 'canonical-glossary-quality-v1'
 
     def __init__(self, ds_key, s2_key=""):
         self.ds_key = str(ds_key or "").strip()
@@ -6491,6 +6492,106 @@ TEXT:
                 mapping[term] = target
         return mapping
 
+    @staticmethod
+    def _canonical_target_surface_suspicious(target, target_lang):
+        """Detect only high-confidence malformed canonical-term surfaces."""
+        value = re.sub(r'\s+', ' ', str(target or '')).strip()
+        if not value:
+            return True
+        profile = SSciPipelineAgent._translation_language_profile(target_lang)
+        if profile.get('language_code') == 'zh':
+            compact = re.sub(r'\s+', '', value)
+            # Repeated multi-character fragments such as 情境感知感知 are a
+            # high-confidence malformed canonical surface, while single-character
+            # reduplication is left alone because it can be grammatical Chinese.
+            if re.search(r'([\u4e00-\u9fff]{2,4})\1', compact):
+                return True
+        return False
+
+    def validate_and_refine_canonical_concept_glossary(self, concept_ledger, target_lang, glossary):
+        """Independently review strict canonical terms once before body translation.
+
+        The first glossary call establishes candidate identities.  This second,
+        bounded call verifies semantic identity and natural academic surface form
+        before those candidates become immutable for all later ZH/RU/KK sections.
+        At most one corrected target is accepted for each strict construct; no
+        section-level repair loop is involved.
+        """
+        mapping = dict(glossary or {})
+        rows = (dict(concept_ledger or {}).get('items') or []) if isinstance(concept_ledger, dict) else []
+        strict_sources = []
+        for row in rows:
+            if not isinstance(row, dict):
+                continue
+            if str(row.get('policy') or '').strip().upper() != 'STRICT_CONSTRUCT':
+                continue
+            source = re.sub(r'\s+', ' ', str(row.get('source') or '')).strip()
+            if source and source in mapping and source not in strict_sources:
+                strict_sources.append(source)
+        if not strict_sources:
+            return mapping
+
+        payload = [
+            {'source': source, 'candidate_target': str(mapping.get(source) or '').strip()}
+            for source in strict_sources
+        ]
+        if any(not row['candidate_target'] for row in payload):
+            raise DeepSeekRequestError('Canonical glossary quality review received an empty strict target.')
+        language_rule = self._translation_language_rule(target_lang)
+        response = self.call_deepseek(
+            (
+                f"Independently review strict research-construct translations for {target_lang}. {language_rule} "
+                "Judge semantic identity and natural academic terminology, not stylistic preference. "
+                "Reject duplicated/malformed wording, concept narrowing/broadening, or a target that changes the construct. "
+                "If the candidate is sound, return it unchanged with accepted=true. If it is materially flawed, return exactly one corrected canonical target with accepted=false. "
+                "Do not merge distinct constructs. Return JSON only as "
+                '{"terms":[{"source":"...","candidate_target":"...","accepted":true,"target":"..."}]}'
+                " and return every source exactly once."
+            ),
+            json.dumps({'terms': payload}, ensure_ascii=False),
+            max_tokens=max(1200, min(4200, 180 + len(payload) * 120)),
+            temp=0.0, model_role='translation', thinking=False, allow_fallback=False,
+        )
+        parsed = self._extract_json_from_text(response)
+        items = parsed.get('terms', []) if isinstance(parsed, dict) else []
+        returned = {}
+        for item in items:
+            if not isinstance(item, dict):
+                continue
+            source = re.sub(r'\s+', ' ', str(item.get('source') or '')).strip()
+            if source:
+                returned[source] = item
+        if set(returned) != set(strict_sources):
+            raise DeepSeekRequestError('Canonical glossary quality review omitted or changed a strict source identity.')
+
+        refined = dict(mapping)
+        for source in strict_sources:
+            item = returned[source]
+            candidate = str(mapping.get(source) or '').strip()
+            accepted_raw = item.get('accepted')
+            accepted = accepted_raw is True or str(accepted_raw).strip().casefold() == 'true'
+            reviewed_target = re.sub(r'\s+', ' ', str(item.get('target') or '')).strip()
+            suspicious = self._canonical_target_surface_suspicious(candidate, target_lang)
+            if accepted and not suspicious:
+                # An acceptance cannot silently remap the frozen candidate.
+                if reviewed_target and reviewed_target != candidate:
+                    raise DeepSeekRequestError('Canonical glossary review changed an accepted strict target.')
+                continue
+            if not reviewed_target or reviewed_target == candidate:
+                raise DeepSeekRequestError(
+                    f'Canonical concept quality review did not provide a usable correction: {source[:80]}'
+                )
+            if reviewed_target.casefold() == source.casefold() and re.search(r'[a-z]{4,}', source):
+                raise DeepSeekRequestError(f'Canonical concept remained untranslated after quality review: {source[:80]}')
+            if self._canonical_target_surface_suspicious(reviewed_target, target_lang):
+                raise DeepSeekRequestError(f'Canonical concept correction remains malformed: {source[:80]}')
+            refined[source] = reviewed_target
+            self._record_translation_assurance(
+                'CANONICAL_GLOSSARY_REFINED', target_lang=target_lang, severity='repair',
+                decision='REPAIR_LOCAL', detail={'source': source[:120]},
+            )
+        return refined
+
     def extend_translation_glossary(self, body, target_lang, existing=None):
         """Extend a task-level glossary without remapping an existing term."""
         mapping = dict(existing or {})

core/production_guard.py

@@ -56,6 +56,80 @@ def validate_docx_bytes(data: bytes, *, minimum_paragraphs: int = 8) -> None:
         )
 
 
+def _table_reference_numbers(markdown: str) -> set[int]:
+    value = str(markdown or "")
+    # The References tail can contain article titles with the word "Table"; it
+    # is bibliographic data and must not be mistaken for a manuscript table
+    # cross-reference.
+    reference_heading = re.search(
+        r"(?mi)^#\s+(?:References|Reference List|参考文献|Литература|Список литературы|Әдебиеттер|Пайдаланылған әдебиеттер)\s*$",
+        value,
+    )
+    if reference_heading:
+        value = value[:reference_heading.start()]
+    numbers = set()
+    patterns = [
+        r"\bTable\s+(\d+)\b",
+        r"表\s*(\d+)",
+        r"\bТаблиц(?:а|ы|е|у|ей|ами|ах)?\s*(\d+)\b",
+        r"\bКесте\s*(\d+)\b",
+        r"\b(\d+)\s*[-–—]?\s*кесте\b",
+    ]
+    for pattern in patterns:
+        for match in re.finditer(pattern, value, flags=re.I):
+            try:
+                numbers.add(int(match.group(1)))
+            except Exception:
+                continue
+    return numbers
+
+
+def validate_quantitative_table_contract(
+    markdown: str, data: bytes, manifest: dict[str, Any] | None, *, label: str = "DOCX"
+) -> None:
+    """Validate table inventory/reference parity for the system-generated AMJ path."""
+    record = dict(manifest or {})
+    rows = [item for item in (record.get("tables") or []) if isinstance(item, dict)]
+    if not rows:
+        return
+    expected_numbers = {
+        int(item.get("number") or 0) for item in rows if int(item.get("number") or 0) > 0
+    }
+    if not expected_numbers:
+        raise OutputValidationError(f"{label} quantitative table manifest is empty or invalid.")
+    try:
+        document = Document(io.BytesIO(data))
+    except Exception as exc:
+        raise OutputValidationError(f"{label} quantitative table contract cannot inspect DOCX: {exc}") from exc
+    if len(document.tables) != len(expected_numbers):
+        raise OutputValidationError(
+            f"{label} quantitative table inventory mismatch: actual={len(document.tables)}, expected={len(expected_numbers)}."
+        )
+    referenced = _table_reference_numbers(markdown)
+    unexpected = sorted(referenced - expected_numbers)
+    missing = sorted(expected_numbers - referenced)
+    if unexpected:
+        raise OutputValidationError(
+            f"{label} references unexpected quantitative table number(s): {unexpected}."
+        )
+    if missing:
+        raise OutputValidationError(
+            f"{label} is missing reader-facing table number(s): {missing}."
+        )
+    if str(label).casefold().startswith("english"):
+        for item in rows:
+            number = int(item.get("number") or 0)
+            title = str(item.get("title") or "").strip()
+            if not number or not title:
+                continue
+            if not re.search(
+                rf"(?mi)^\s*Table\s+{number}\s*:\s*{re.escape(title)}\s*$", str(markdown or "")
+            ):
+                raise OutputValidationError(
+                    f"English manuscript table caption mismatch for Table {number}: {title}."
+                )
+
+
 def validate_qualitative_appendix_docx_layout(data: bytes, *, label: str = "DOCX") -> None:
     """Fail closed if the qualitative evidence Appendix lost its layout contract.
 
@@ -512,6 +586,7 @@ def build_evidence_coverage_report(summary: dict[str, Any] | None) -> str:
         "SmartPaper Evidence Coverage Report",
         "",
         f"Overall evidence status: {'LIMITED' if limited else 'ADEQUATE'}",
+        f"Submission readiness: {'REVIEW REQUIRED — evidence coverage is limited' if limited else 'QUALITY GATE PASSED'}",
         f"Validated references in final manuscript: {max(0, int(data.get('final_reference_count') or 0))}",
         f"Early coverage risk detected: {'yes' if bool(data.get('early_risk_detected')) else 'no'}",
         "",
@@ -546,6 +621,7 @@ def build_evidence_coverage_report(summary: dict[str, Any] | None) -> str:
             "Author guidance:",
             "The manuscript was completed because structural, translation, citation-correctness, quotation/fact, DOCX, and ZIP integrity gates are independent from literature coverage quantity.",
             "Before formal submission, review the evidence gaps and consider adding more directly relevant core literature where appropriate.",
+            "Do not treat an evidence-limited package as submission-ready until the evidence gaps have been reviewed and, where possible, strengthened.",
         ])
     return "\n".join(lines).strip() + "\n"
 
@@ -585,6 +661,14 @@ def build_validated_package(
         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")
@@ -621,6 +705,12 @@ def build_validated_package(
             for name in (english_filename, translated_filename):
                 archived_docx = archive.read(name)
                 validate_docx_bytes(archived_docx)
+                if quantitative_manifest:
+                    archived_markdown = english_markdown if name == english_filename else translated_markdown
+                    validate_quantitative_table_contract(
+                        archived_markdown, archived_docx, quantitative_manifest,
+                        label="Archived English DOCX" if name == english_filename else "Archived target-language DOCX",
+                    )
                 if qualitative_mode:
                     validate_qualitative_appendix_docx_layout(
                         archived_docx,

core/quantitative_engine.py

@@ -18,6 +18,113 @@ class QuantitativeEngine:
     """Prepare one frozen quantitative evidence packet before manuscript writing."""
 
     QUANTITATIVE_EVIDENCE_LEDGER_VERSION = "quantitative-evidence-ledger-v2"
+    QUANTITATIVE_TABLE_MANIFEST_VERSION = "quantitative-table-manifest-v1"
+    HYPOTHESIS_LEDGER_VERSION = "quantitative-hypothesis-ledger-v1"
+
+    @classmethod
+    def build_hypothesis_ledger(cls, constructs):
+        """Freeze one result-independent AMJ hypothesis identity contract.
+
+        The system-generated quantitative design is observational and does not
+        justify result-driven directional hypotheses.  The three numbered
+        hypotheses therefore freeze only the construct pairing/association
+        identity.  The indirect association is an analysis target, not H4.
+        """
+        values = [re.sub(r"\s+", " ", str(item or "")).strip() for item in (constructs or [])]
+        if len(values) != 3 or any(not item for item in values):
+            return {}
+        predictor, mediator, outcome = values
+        items = [
+            {"hypothesis_id": "H1", "predictor": predictor, "outcome": outcome, "relation": "association"},
+            {"hypothesis_id": "H2", "predictor": predictor, "outcome": mediator, "relation": "association"},
+            {"hypothesis_id": "H3", "predictor": mediator, "outcome": outcome, "relation": "association"},
+        ]
+        return {
+            "version": cls.HYPOTHESIS_LEDGER_VERSION,
+            "items": items,
+            "indirect_analysis": {
+                "predictor": predictor,
+                "mediator": mediator,
+                "outcome": outcome,
+                "numbered_hypothesis": False,
+                "relation": "estimated_indirect_association",
+            },
+        }
+
+    @classmethod
+    def build_table_manifest_from_evidence_markdown(cls, evidence_markdown):
+        """Rebuild the synthetic table manifest from a durable pre-C5 evidence packet."""
+        value = str(evidence_markdown or "")
+        targets = {
+            1: "Descriptive Statistics and Measurement Assessment",
+            2: "Descriptive Statistics and Measurement Assessment",
+            3: "Hypothesis and Association Tests",
+            4: "Additional Analyses and Robustness Checks",
+        }
+        rows = []
+        pattern = re.compile(
+            r"(?ms)^###\s+Table\s+(?P<number>[1-4])\s*:\s*(?P<title>[^\n]+)\n(?P<body>.*?)(?=^###\s+|\Z)"
+        )
+        for match in pattern.finditer(value):
+            number = int(match.group("number"))
+            body = match.group("body").strip()
+            table_lines = []
+            note = ""
+            for line in body.splitlines():
+                stripped = line.strip()
+                if stripped.startswith("|"):
+                    table_lines.append(line.rstrip())
+                elif re.match(r"^\*?Note\s*:", stripped, flags=re.I):
+                    note = re.sub(r"^\*?Note\s*:\s*", "", stripped, flags=re.I).strip("* ")
+                elif table_lines:
+                    break
+            if len(table_lines) < 2:
+                continue
+            title = re.sub(r"^Frozen\s+", "", match.group("title").strip(), flags=re.I)
+            row = {
+                "number": number,
+                "title": title,
+                "target_subsection": targets[number],
+                "markdown": "\n".join(table_lines),
+            }
+            if note:
+                row["note"] = note
+            rows.append(row)
+        if {row["number"] for row in rows} != {1, 2, 3, 4}:
+            return {}
+        return {"version": cls.QUANTITATIVE_TABLE_MANIFEST_VERSION, "tables": sorted(rows, key=lambda row: row["number"])}
+
+    @classmethod
+    def build_table_manifest(cls, validity_df, fl_matrix_df, path_table, spearman):
+        """Return the exact reader-facing tables required by the synthetic AMJ path."""
+        rows = [
+            {
+                "number": 1,
+                "title": "Reliability and Convergent Validity",
+                "target_subsection": "Descriptive Statistics and Measurement Assessment",
+                "markdown": validity_df.to_markdown(index=False),
+            },
+            {
+                "number": 2,
+                "title": "Descriptive Statistics and Discriminant Validity (Fornell-Larcker Criterion)",
+                "target_subsection": "Descriptive Statistics and Measurement Assessment",
+                "note": "Diagonal elements in parentheses are the square roots of AVE.",
+                "markdown": fl_matrix_df.to_markdown(),
+            },
+            {
+                "number": 3,
+                "title": "Association Model Results",
+                "target_subsection": "Hypothesis and Association Tests",
+                "markdown": path_table.to_markdown(index=False),
+            },
+            {
+                "number": 4,
+                "title": "Rank-Order Robustness Correlations (Spearman)",
+                "target_subsection": "Additional Analyses and Robustness Checks",
+                "markdown": spearman.round(3).to_markdown(),
+            },
+        ]
+        return {"version": cls.QUANTITATIVE_TABLE_MANIFEST_VERSION, "tables": rows}
 
     @staticmethod
     def generate_psd_corr_matrix(n_constructs, max_corr=0.5):
@@ -989,6 +1096,10 @@ Keywords: {keywords}"""
                 "Frozen bootstrap and collinearity checks",
             ],
         }
+        metadata["hypothesis_ledger"] = self.build_hypothesis_ledger(construct_names)
+        metadata["quantitative_table_manifest"] = self.build_table_manifest(
+            validity_df, fl_matrix_df, path_table, spearman
+        )
         metadata["quantitative_evidence_ledger"] = self.build_quantitative_evidence_ledger(metadata, generated_prompt)
         return (
             df,