S0-R1-C5.5-R6-R4 Canonical Concept + Protected Anchor Translation Closure

core/generation_runner.py

@@ -2168,13 +2168,28 @@ class GenerationRunner:
         if not callable(translate_cells):
             raise RuntimeError('Typed qualitative Appendix cell translation entry point is unavailable.')
         try:
-            result = translate_cells(
-                qual_engine.appendix_translation_cells(manifest), target_lang,
-                glossary=translation_glossary, max_attempts=2,
-                claim_states=(existing.get('appendix_cell_claim_states') if isinstance(existing.get('appendix_cell_claim_states'), dict) else None),
+            prebound_cells, model_cells = qual_engine.appendix_translation_plan(
+                manifest, translation_glossary
             )
-            cell_translations = dict((result or {}).get('translations') or {})
-            updated_glossary = dict((result or {}).get('glossary') or translation_glossary or {})
+            if model_cells:
+                result = translate_cells(
+                    model_cells, target_lang,
+                    glossary=translation_glossary, max_attempts=2,
+                    claim_states=(existing.get('appendix_cell_claim_states') if isinstance(existing.get('appendix_cell_claim_states'), dict) else None),
+                )
+                model_translations = dict((result or {}).get('translations') or {})
+                updated_glossary = dict((result or {}).get('glossary') or translation_glossary or {})
+            else:
+                result = {
+                    'qa_status': 'passed', 'attempts': 1, 'claim_states': {},
+                    'appendix_cell_contract_version': 'typed-qualitative-appendix-cells-v2',
+                }
+                model_translations = {}
+                updated_glossary = dict(translation_glossary or {})
+            cell_translations = {**dict(prebound_cells), **dict(model_translations)}
+            expected_cell_ids = set(qual_engine.appendix_translation_cells(manifest))
+            if set(cell_translations) != expected_cell_ids:
+                raise RuntimeError('qualitative_appendix_target_cell_binding_incomplete')
             translated_text = qual_engine.bind_target_language_appendix(
                 manifest, target_lang, cell_translations, quote_map
             )
@@ -2222,6 +2237,7 @@ class GenerationRunner:
             'qualitative_appendix_manifest_hash': manifest_hash,
             'appendix_cell_contract_version': str((result or {}).get('appendix_cell_contract_version') or ''),
             'appendix_cell_claim_states': dict((result or {}).get('claim_states') or {}),
+            'appendix_prebound_label_count': len(prebound_cells),
             'updated_at': datetime.now(timezone.utc).isoformat(),
         }
         return sections, updated_glossary

core/pipeline_agent.py

@@ -74,7 +74,7 @@ class SSciPipelineAgent:
     MAX_TRANSLATION_CLAIM_REPAIR_ATTEMPTS = 2
     MAX_TRANSLATION_CLAIM_TOTAL_REPAIRS = 6
     PROTECTED_ENTITY_LEDGER_VERSION = 'protected-entity-ledger-v2'
-    CANONICAL_GLOSSARY_QUALITY_VERSION = 'canonical-glossary-quality-v1'
+    CANONICAL_GLOSSARY_QUALITY_VERSION = 'canonical-glossary-quality-v2'
     SEMANTIC_CLAIM_FRAME_VERSION = 'smartpaper-core-semantic-claim-frame-v1'
     CITATION_CLAIM_FRAME_VERSION = 'smartpaper-citation-claim-frame-v1'
 
@@ -6026,7 +6026,10 @@ TEXT:
         }
 
     @classmethod
-    def _canonical_glossary_violations(cls, source, target, target_lang, glossary=None, canonical_sources=None):
+    def _canonical_glossary_violations(
+        cls, source, target, target_lang, glossary=None, canonical_sources=None,
+        canonical_kinds=None, section_name='',
+    ):
         """Return canonical concept terms whose frozen target identity disappeared.
 
         Chinese uses compact exact identity. Russian/Kazakh use conservative
@@ -6039,16 +6042,37 @@ TEXT:
         language_code = profile.get('language_code')
         violations = []
         canonical = {str(item or '').strip().casefold() for item in (canonical_sources or []) if str(item or '').strip()}
+        kinds = {
+            str(key or '').strip().casefold(): str(value or '').strip().casefold()
+            for key, value in dict(canonical_kinds or {}).items() if str(key or '').strip()
+        }
+        label_context = 'qualitative appendix semantic cells' in str(section_name or '').casefold()
         for source_term, target_term in dict(glossary or {}).items():
             source_term = str(source_term or '').strip()
             target_term = str(target_term or '').strip()
             if canonical and source_term.casefold() not in canonical:
                 continue
+            # Qualitative theme/concept names are exact reader-facing labels in
+            # the deterministic Appendix, but ordinary prose may realize the
+            # same semantic identity with natural grammar.  Requiring an exact
+            # frozen phrase inside every ZH/RU/KK sentence created a false
+            # terminology loop (especially for morphologically rich RU/KK and
+            # compact Chinese paraphrase).  Bilingual semantic QA remains the
+            # prose authority; label identity stays exact in the Appendix binder.
+            concept_kind = kinds.get(source_term.casefold(), '')
+            if concept_kind in {'qualitative_theme', 'qualitative_concept'} and not label_context:
+                continue
             if not source_term or not target_term or source_term.casefold() not in source_value.casefold():
                 continue
             # Pure technical/acronym identities are protected elsewhere.
+            # Skip deterministic canonical checking only when the *entire*
+            # target is a pure technical/acronym identity.  A localized label
+            # containing an embedded acronym (e.g. ``AI推荐...``) is still a
+            # conceptual phrase and must not be mistaken for the bare ``AI``
+            # token merely because stripping non-ASCII characters leaves AI.
+            pure_ascii = bool(re.fullmatch(r'[A-Za-z0-9+.#-]+', target_term))
             compact = re.sub(r'[^A-Za-z0-9]', '', target_term)
-            if compact and ' ' not in target_term and (
+            if pure_ascii and compact and ' ' not in target_term and (
                 compact.isupper() or any(ch.isdigit() for ch in compact)
                 or (len(compact) > 2 and any(ch.isupper() for ch in compact[1:]))
             ):
@@ -6076,6 +6100,30 @@ TEXT:
                     violations.append(source_term)
         return list(dict.fromkeys(violations))[:8]
 
+    @classmethod
+    def _canonical_target_preserves_source_anchors(cls, source, target):
+        """Require exact-render source anchors to survive inside a canonical label.
+
+        Canonical labels and the protected-anchor compiler must never issue
+        contradictory instructions.  Dates/word-quantities are intentionally
+        excluded because their render policy is localized/canonicalized; exact
+        acronyms, names, hypotheses, IDs, formulas, statistics and numbers remain
+        byte-stable under the existing translation contract.
+        """
+        _protected, mapping = cls._protect_translation_tokens(str(source or ''))
+        target_value = str(target or '')
+        required = Counter()
+        for manifest in dict(mapping or {}).values():
+            if not isinstance(manifest, dict):
+                continue
+            if str(manifest.get('render_policy') or '') != 'exact':
+                continue
+            surface = str(manifest.get('source') or '')
+            if not surface or surface == '|':
+                continue
+            required[surface] += 1
+        return all(target_value.count(surface) >= count for surface, count in required.items())
+
     @classmethod
     def _compile_translation_ir(cls, source):
         """Compile a private, language-neutral translation intermediate representation."""
@@ -7049,6 +7097,7 @@ TEXT:
         sources = []
         strict_sources = []
         flexible_sources = []
+        source_kinds = {}
         for row in rows:
             if not isinstance(row, dict):
                 continue
@@ -7056,6 +7105,7 @@ TEXT:
             if not term or term in sources:
                 continue
             sources.append(term)
+            source_kinds[term] = str(row.get('kind') or 'concept').strip()
             policy = str(row.get('policy') or 'STRICT_CONSTRUCT').strip().upper()
             if policy == 'FLEXIBLE_TERM':
                 flexible_sources.append(term)
@@ -7067,6 +7117,7 @@ TEXT:
         # grammar or equivalent wording; semantic QA remains authoritative.
         self._canonical_concept_sources = set(strict_sources)
         self._canonical_flexible_sources = set(flexible_sources)
+        self._canonical_concept_kinds = dict(source_kinds)
         if not sources:
             return mapping
 
@@ -7155,6 +7206,7 @@ TEXT:
                 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. "
+                "Every protected exact source anchor inside a construct (for example AI, LLM, model/tool IDs, hypothesis IDs, formulas, statistics or exact numbers) must remain byte-for-byte in the target canonical label; never localize, delete, duplicate, or rename such an anchor. "
                 "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":"..."}]}'
@@ -7183,7 +7235,10 @@ TEXT:
             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)
+            suspicious = (
+                self._canonical_target_surface_suspicious(candidate, target_lang)
+                or not self._canonical_target_preserves_source_anchors(source, candidate)
+            )
             if accepted and not suspicious:
                 # An acceptance cannot silently remap the frozen candidate.
                 if reviewed_target and reviewed_target != candidate:
@@ -7195,7 +7250,10 @@ TEXT:
                 )
             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):
+            if (
+                self._canonical_target_surface_suspicious(reviewed_target, target_lang)
+                or not self._canonical_target_preserves_source_anchors(source, reviewed_target)
+            ):
                 raise DeepSeekRequestError(f'Canonical concept correction remains malformed: {source[:80]}')
             refined[source] = reviewed_target
             self._record_translation_assurance(
@@ -7523,6 +7581,8 @@ If ok=false, codes MUST contain only values from allowed_issue_codes. For SEMANT
                 canonical_violations = self._canonical_glossary_violations(
                     meta['english'], meta['target'], target_lang, relevant_glossary,
                     canonical_sources=getattr(self, '_canonical_concept_sources', set()),
+                    canonical_kinds=getattr(self, '_canonical_concept_kinds', {}),
+                    section_name=section_name,
                 )
                 if canonical_violations:
                     if 'TERMINOLOGY' not in codes:
@@ -8580,10 +8640,21 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
             section_name='Qualitative Appendix semantic cells', max_attempts=max_attempts,
             stage_callback=stage_callback, claim_states=claim_states,
         )
-        body_units = [
-            unit['text'] for unit in self._translation_body_units(result.get('text') or '')
-            if unit.get('kind') == 'body'
-        ]
+        body_map = dict((result or {}).get('body_translations') or {})
+        if body_map:
+            expected_block_ids = [f'{index:06d}' for index in range(1, len(ordered) + 1)]
+            if set(body_map) != set(expected_block_ids):
+                raise DeepSeekRequestError(
+                    'Qualitative Appendix compiler block identity mismatch after translation: '
+                    f'expected={expected_block_ids[:4]} got={sorted(body_map)[:4]}.'
+                )
+            body_units = [str(body_map[block_id] or '').strip() for block_id in expected_block_ids]
+        else:
+            # Backward-compatible fallback for test doubles/legacy callers.
+            body_units = [
+                unit['text'] for unit in self._translation_body_units(result.get('text') or '')
+                if unit.get('kind') == 'body'
+            ]
         if len(body_units) != len(ordered):
             raise DeepSeekRequestError(
                 'Qualitative Appendix cell identity mismatch after translation: '
@@ -8601,7 +8672,7 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
             **dict(result or {}),
             'translations': translations,
             'cell_ids': list(cell_objects.keys()),
-            'appendix_cell_contract_version': 'typed-qualitative-appendix-cells-v1',
+            'appendix_cell_contract_version': 'typed-qualitative-appendix-cells-v2',
         }
 
     def translate_frozen_section(self, section_text, target_lang, glossary=None, section_name='', max_attempts=2, stage_callback=None, claim_states=None):
@@ -9018,6 +9089,7 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                 # Rebuild after any localized repairs, then apply deterministic
                 # whole-unit integrity/language checks before publication.
                 output_units = []
+                body_translations = {}
                 body_index = 0
                 for unit in units:
                     if unit['kind'] == 'heading':
@@ -9027,7 +9099,10 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                         output_units.append('#' * int(unit['level']) + ' ' + translated_heading)
                     else:
                         body_index += 1
-                        output_units.append(self._strip_translation_claim_markers(translations[f'{body_index:06d}']))
+                        block_id = f'{body_index:06d}'
+                        clean_body = self._strip_translation_claim_markers(translations[block_id])
+                        body_translations[block_id] = clean_body
+                        output_units.append(clean_body)
                 output = '\n\n'.join(output_units).strip()
                 self.validate_translation_alignment(value, output)
                 self._validate_translation_language_quality(output, target_lang)
@@ -9045,6 +9120,10 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                     'diagnostics': diagnostics[-20:],
                     'claim_states': dict(self._last_translation_claim_states),
                     'claim_state_version': 'translation-claim-state-v2',
+                    # Internal typed consumers (notably the qualitative Appendix)
+                    # bind by compiler block identity rather than reparsing
+                    # paragraph breaks from reader-facing text.
+                    'body_translations': dict(body_translations),
                 }
             except PipelineCancelledError:
                 raise

core/qualitative_engine.py

@@ -13,9 +13,10 @@ from config import APP_CONFIG
 
 class QualitativeEngine:
     QUALITATIVE_APPENDIX_MANIFEST_VERSION = "qualitative-appendix-manifest-v1"
-    TARGET_APPENDIX_BINDING_VERSION = "target-qualitative-appendix-binding-v1"
+    TARGET_APPENDIX_BINDING_VERSION = "target-qualitative-appendix-binding-v2"
     SEMANTIC_CLAIM_FRAME_VERSION = "smartpaper-core-semantic-claim-frame-v1"
     APPENDIX_FIELDS = ("theme", "concept", "stage", "case_label", "evidence_role")
+    APPENDIX_CANONICAL_LABEL_FIELDS = ("theme", "concept")
     APPENDIX_HEADERS = (
         "Higher-order Theme", "Empirical Concept", "Process Stage",
         "Case/Actor", "Evidence Role", "Representative Quote",
@@ -412,6 +413,37 @@ class QualitativeEngine:
                 cells[f"{row_id}:{field}"] = value
         return cells
 
+    @classmethod
+    def appendix_translation_plan(cls, manifest, glossary=None):
+        """Split Appendix cells into server-bound labels and model-owned surfaces.
+
+        Theme/concept cells are frozen research labels.  When the task-level
+        canonical glossary already owns their target identity, the translation
+        model must not rewrite those labels or their internal anchors.  Remaining
+        ordinary cells (stage/case/evidence-role) continue through the typed
+        translation compiler and are rebound by server-side cell ID.
+        """
+        expected = cls.appendix_translation_cells(manifest)
+        glossary_value = dict(glossary or {})
+        folded_glossary = {
+            str(key or '').strip().casefold(): str(value or '').strip()
+            for key, value in glossary_value.items() if str(key or '').strip() and str(value or '').strip()
+        }
+        prebound = {}
+        model_cells = {}
+        for cell_id, source in expected.items():
+            field = str(cell_id or '').rsplit(':', 1)[-1]
+            canonical_target = str(glossary_value.get(source) or '').strip()
+            if not canonical_target:
+                canonical_target = folded_glossary.get(str(source or '').strip().casefold(), '')
+            if field in cls.APPENDIX_CANONICAL_LABEL_FIELDS and canonical_target:
+                prebound[cell_id] = canonical_target
+            else:
+                model_cells[cell_id] = source
+        if set(prebound) & set(model_cells) or set(prebound) | set(model_cells) != set(expected):
+            raise RuntimeError("qualitative_appendix_translation_plan_identity_mismatch")
+        return prebound, model_cells
+
     @staticmethod
     def _escape_appendix_markdown_cell(value):
         return re.sub(r"\s+", " ", str(value or "").strip()).replace("|", r"\|")