R6A-R4-R10 typed qualitative quote translation object and deterministic target slot injection

core/generation_runner.py

@@ -651,12 +651,11 @@ class GenerationRunner:
 
     @classmethod
     def _prepare_qualitative_translation_source(cls, source, quote_sources, quote_ledger):
-        """Use one task-level target rendering for every frozen qualitative quote.
+        """Replace every frozen qualitative quote with a typed semantic slot.
 
-        The first translated occurrence keeps the English quote in context and
-        appends its protected QUOTE tag so the target rendering can be captured.
-        Later occurrences (notably the Appendix) send only the protected tag and
-        are restored from the already QA-passed task-level quote ledger.
+        QUOTE identity is never inferred from translated narrative surface text.
+        Missing target objects are classified as pending and translated through
+        the dedicated typed-object path before the surrounding narrative runs.
         """
         value = str(source or "")
         pending = []
@@ -665,42 +664,21 @@ class GenerationRunner:
             wrapped = f'"{quote}"'
             tagged = f'{wrapped} [{tag}]'
             if tagged in value:
-                if tag in (quote_ledger or {}):
-                    value = value.replace(tagged, f"[{tag}]")
-                    reused.append(tag)
-                else:
-                    pending.append(tag)
+                value = value.replace(tagged, f"[{tag}]")
+                (reused if tag in (quote_ledger or {}) else pending).append(tag)
                 continue
             if wrapped not in value:
                 continue
-            if tag in (quote_ledger or {}):
-                value = value.replace(wrapped, f"[{tag}]")
-                reused.append(tag)
-            else:
-                value = value.replace(wrapped, tagged)
-                pending.append(tag)
+            value = value.replace(wrapped, f"[{tag}]")
+            (reused if tag in (quote_ledger or {}) else pending).append(tag)
         return value, pending, reused
 
     @classmethod
     def _finalize_qualitative_translation_quotes(cls, translated, pending, reused, quote_ledger):
         value = str(translated or "")
         ledger = dict(quote_ledger or {})
-        quote_pattern = r'(?P<quote>"[^"\n]{4,2500}"|“[^”\n]{4,2500}”|«[^»\n]{4,2500}»|„[^“\n]{4,2500}“)' 
-        for tag in pending or []:
-            pattern = re.compile(
-                quote_pattern + rf"\s*(?<!\w)\[?{re.escape(tag)}\]?(?!\w)",
-                flags=re.UNICODE,
-            )
-            matches = list(pattern.finditer(value))
-            if len(matches) != 1:
-                raise RuntimeError(
-                    f"Qualitative quote translation ledger capture failed: {tag} matches={len(matches)}"
-                )
-            rendered_quote = matches[0].group("quote")
-            ledger[tag] = rendered_quote
-            value = value[:matches[0].start()] + rendered_quote + value[matches[0].end():]
-
-        for tag in reused or []:
+        ordered_tags = list(dict.fromkeys(list(pending or []) + list(reused or [])))
+        for tag in ordered_tags:
             rendered_quote = str(ledger.get(tag) or "").strip()
             if not rendered_quote:
                 raise RuntimeError(f"Qualitative quote translation ledger missing target text: {tag}")
@@ -708,7 +686,7 @@ class GenerationRunner:
             count = len(re.findall(pattern, value))
             if count != 1:
                 raise RuntimeError(
-                    f"Qualitative quote translation ledger reuse failed: {tag} matches={count}"
+                    f"Qualitative quote deterministic target slot injection failed: {tag} matches={count}"
                 )
             value = re.sub(pattern, lambda _m, q=rendered_quote: q, value, count=1)
 
@@ -1104,6 +1082,46 @@ class GenerationRunner:
             )
 
         try:
+            if pending_quote_tags:
+                quote_translate_method = getattr(agent, 'translate_qualitative_quote_objects', None)
+                if not callable(quote_translate_method):
+                    raise RuntimeError('Typed qualitative quote translation entry point is unavailable.')
+                quote_objects = {
+                    tag: getattr(self, '_qualitative_quote_sources', {}).get(tag, '')
+                    for tag in pending_quote_tags
+                }
+                quote_result = quote_translate_method(
+                    quote_objects,
+                    target_lang,
+                    glossary=translation_glossary,
+                    max_attempts=2,
+                    stage_callback=publish_translation_stage,
+                )
+                returned_quotes = (quote_result or {}).get('translations')
+                if not isinstance(returned_quotes, dict) or set(returned_quotes) != set(pending_quote_tags):
+                    raise RuntimeError(
+                        'Typed qualitative quote translation identity mismatch: '
+                        f'expected={sorted(pending_quote_tags)} got={sorted(returned_quotes) if isinstance(returned_quotes, dict) else []}'
+                    )
+                for tag in pending_quote_tags:
+                    rendered_quote = str(returned_quotes.get(tag) or '').strip()
+                    if not rendered_quote:
+                        raise RuntimeError(f'Typed qualitative quote translation is empty: {tag}')
+                    quote_ledger[tag] = rendered_quote
+                returned_glossary = (quote_result or {}).get('glossary')
+                if isinstance(returned_glossary, dict):
+                    translation_glossary = dict(returned_glossary)
+                updated_sections['__quote_ledger__'] = {
+                    'task_id': getattr(getattr(self, "task_dir", None), "name", ""),
+                    'request_hash': current_request_hash,
+                    'writer_execution_epoch': int(getattr(self, "execution_epoch", 0) or 0),
+                    'target_lang': target_lang,
+                    'translation_contract_version': TRANSLATION_CONTRACT_VERSION,
+                    'source_quote_hashes': current_quote_hashes,
+                    'translations': dict(quote_ledger),
+                    'updated_at': datetime.now(timezone.utc).isoformat(),
+                }
+
             translate_method = agent.translate_frozen_section
             call_kwargs = {
                 'glossary': translation_glossary, 'section_name': unit_name, 'max_attempts': 2,

core/pipeline_agent.py

@@ -4695,6 +4695,82 @@ Output only the repaired section."""
             )
         return True
 
+    @classmethod
+    def _render_qualitative_quote_translation(cls, translated, target_lang):
+        """Return one reader-facing target quotation from a typed quote object.
+
+        Quote identity is carried by the caller's object key, never inferred
+        from where a QUOTE-ID happened to surface in translated narrative.
+        The model translates only the frozen quotation payload; the outer
+        quotation punctuation is rendered deterministically by language.
+        """
+        text = str(translated or '').strip()
+        if not text:
+            raise DeepSeekRequestError('Qualitative quote translation is empty.')
+        if re.search(r'\bQUOTE-\d+\b', text):
+            raise DeepSeekRequestError('Qualitative quote translation leaked an internal Quote ID.')
+
+        quote_pairs = (
+            ('"', '"'), ('“', '”'), ('«', '»'), ('„', '“'), ('「', '」'), ('『', '』'),
+        )
+        inner = text
+        for left, right in quote_pairs:
+            if inner.startswith(left) and inner.endswith(right) and len(inner) > len(left) + len(right):
+                inner = inner[len(left):-len(right)].strip()
+                break
+        if not inner:
+            raise DeepSeekRequestError('Qualitative quote translation contains no target text.')
+
+        profile = cls._translation_language_profile(target_lang)
+        if profile.get('language_code') == 'zh':
+            left, right = '“', '”'
+        elif profile.get('language_code') in {'ru', 'kk'}:
+            left, right = '«', '»'
+        else:
+            left, right = '“', '”'
+        return f'{left}{inner}{right}'
+
+    def translate_qualitative_quote_objects(
+        self, quote_objects, target_lang, glossary=None, max_attempts=2, stage_callback=None,
+    ):
+        """Translate frozen qualitative quotations as typed objects.
+
+        Each QUOTE-ID is an out-of-band object key.  The corresponding English
+        quotation is translated in isolation through the existing frozen-section
+        compiler and QA stack.  The returned mapping is therefore authoritative
+        without parsing target-language narrative surface order or punctuation.
+        """
+        if not isinstance(quote_objects, dict):
+            raise DeepSeekRequestError('Qualitative quote objects must be a mapping.')
+        frozen_glossary = dict(glossary or {})
+        translations = {}
+        for tag, source_quote in sorted(quote_objects.items()):
+            tag_value = str(tag or '').strip()
+            source_value = str(source_quote or '').strip()
+            if not re.fullmatch(r'QUOTE-\d+', tag_value):
+                raise DeepSeekRequestError(f'Invalid qualitative Quote ID: {tag_value!r}.')
+            if not source_value:
+                raise DeepSeekRequestError(f'Qualitative quote source is empty: {tag_value}.')
+            result = self.translate_frozen_section(
+                source_value,
+                target_lang,
+                glossary=frozen_glossary,
+                section_name=f'Qualitative Quote {tag_value}',
+                max_attempts=max_attempts,
+                stage_callback=stage_callback,
+            )
+            translated = str((result or {}).get('text') or '').strip()
+            translations[tag_value] = self._render_qualitative_quote_translation(translated, target_lang)
+            returned_glossary = (result or {}).get('glossary')
+            if isinstance(returned_glossary, dict):
+                frozen_glossary = dict(returned_glossary)
+        return {
+            'translations': translations,
+            'glossary': frozen_glossary,
+            'qa_status': 'passed',
+            'translation_contract_version': self.TRANSLATION_CONTRACT_VERSION,
+        }
+
     def validate_translation_semantic_equivalence(self, source, translated, target_lang, section_name=''):
         """Fail closed when translation changes claim meaning or evidence strength."""
         source_value = str(source or '').strip()