P3-A5-R2 adaptive recovery integrity correction (#154)

- P3-A5-R2 adaptive recovery integrity correction (49b0ef3ede32183047f4c44967bebd685ce88505)

core/generation_runner.py

@@ -3688,6 +3688,348 @@ class GenerationRunner:
         )
         return updated_sections, updated_glossary
 
+    @staticmethod
+    def _adaptive_recovery_language_code(target_lang: str) -> str:
+        value = str(target_lang or '').strip().casefold().replace('_', '-')
+        if value.startswith('zh') or 'chinese' in value or '中文' in value or '简体' in value:
+            return 'zh'
+        if value.startswith('ru') or 'russian' in value or 'рус' in value:
+            return 'ru'
+        if value.startswith('kk') or value.startswith('kz') or 'kazakh' in value or 'қазақ' in value:
+            return 'kk'
+        return 'other'
+
+    def _validate_adaptive_recovery_text(
+        self, agent, *, source_text: str, translated_text: str, target_lang: str,
+        section_name: str, glossary=None,
+    ) -> None:
+        """Fail closed on adaptive recovery text that bypassed strict compiler QA.
+
+        Adaptive fallbacks exist to preserve delivery after bounded strict-QA
+        exhaustion, not to waive language or semantic integrity. Short quotes and
+        Appendix cells are explicitly re-audited because the ordinary validators
+        intentionally treat very short strings as diagnostic-only.
+        """
+        source = str(source_text or '').strip()
+        translated = str(translated_text or '').strip()
+        if not source or not translated:
+            raise RuntimeError('adaptive_recovery_text_missing')
+
+        # Deterministic script signal for genuinely translatable English lexical
+        # material. Protected acronyms/numeric-only cells are allowed to remain
+        # unchanged and therefore do not require a target-script character.
+        words = re.findall(r'[A-Za-z]{2,}', source)
+        translatable_words = [word for word in words if not (word.isupper() and len(word) <= 6)]
+        protected_surface = bool(
+            source == translated
+            and re.fullmatch(r'[A-Za-z][A-Za-z0-9._+\-]{1,40}', source)
+            and (source.isupper() or any(char.isupper() for char in source[1:]))
+        )
+        if translatable_words and not protected_surface:
+            language_code = self._adaptive_recovery_language_code(target_lang)
+            if language_code == 'zh' and not re.search(r'[\u4e00-\u9fff]', translated):
+                raise RuntimeError('adaptive_recovery_target_language_signal_missing:zh')
+            if language_code in {'ru', 'kk'} and not re.search(
+                r'[А-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі]', translated
+            ):
+                raise RuntimeError(
+                    f'adaptive_recovery_target_language_signal_missing:{language_code}'
+                )
+
+        language_validator = getattr(agent, '_validate_translation_language_quality', None)
+        if callable(language_validator):
+            language_validator(translated, target_lang)
+
+        # The ordinary semantic-equivalence validator intentionally short-circuits
+        # short strings. For those exact objects, require a fresh provider-neutral
+        # binary audit so a non-empty but unrelated translation cannot be accepted.
+        short_object = len(source) < 120 or len(translated) < 80
+        if short_object:
+            call_model = getattr(agent, 'call_model', None)
+            extractor = getattr(agent, '_extract_json_from_text', None)
+            if not callable(call_model):
+                raise RuntimeError('adaptive_recovery_short_semantic_audit_unavailable')
+            prompt = (
+                'Audit semantic equivalence for one short academic translation object. '
+                'Return JSON only: {"ok":true|false,"issues":["short issue"]}. '
+                'Reject if the TARGET is not actually written in the requested target language except '
+                'for source-owned acronyms, proper names, identifiers, or numbers; also reject if it '
+                'omits/adds substantive meaning, changes actor/object/relation, negation, uncertainty, '
+                'causality strength, temporal scope, or quotation meaning. Do not rewrite.\n'
+                f'SECTION: {section_name}\nTARGET_LANGUAGE: {target_lang}\n'
+                f'<ENGLISH>{source}</ENGLISH>\n<TARGET>{translated}</TARGET>'
+            )
+            response = call_model(
+                'You are a conservative bilingual academic translation equivalence auditor.',
+                prompt, max_tokens=700, temp=0.0, model_role='reasoning', thinking=False,
+                allow_fallback=False, logical_node='ordinary_translation_qa',
+            )
+            if callable(extractor):
+                parsed = extractor(response)
+            else:
+                try:
+                    parsed = json.loads(str(response or ''))
+                except Exception:
+                    parsed = None
+            if not isinstance(parsed, dict) or parsed.get('ok') is not True:
+                issues = parsed.get('issues') if isinstance(parsed, dict) else []
+                detail = '; '.join(str(item) for item in (issues or [])[:4])
+                raise RuntimeError(
+                    'adaptive_recovery_short_semantic_equivalence_failed:' + (detail[:300] or 'unconfirmed')
+                )
+            return
+
+        semantic_validator = getattr(agent, 'validate_translation_semantic_equivalence', None)
+        if not callable(semantic_validator):
+            raise RuntimeError('adaptive_recovery_semantic_validator_unavailable')
+        semantic_validator(
+            source, translated, target_lang, section_name=section_name,
+        )
+
+    @staticmethod
+    def _rebind_target_qualitative_appendix_once(
+        manuscript: str, canonical_appendix: str, reference_heading: str,
+    ) -> str:
+        """Deterministically replace/append the final typed Appendix exactly once."""
+        value = str(manuscript or '').strip()
+        appendix = str(canonical_appendix or '').strip()
+        if not appendix:
+            raise RuntimeError('target_qualitative_appendix_rebind_source_missing')
+        heading_match = re.match(r'(?m)^#\s+([^\n]+)\s*$', appendix)
+        if not heading_match:
+            raise RuntimeError('target_qualitative_appendix_rebind_heading_missing')
+        appendix_heading = heading_match.group(1).strip()
+        ref_heading = str(reference_heading or '').strip()
+
+        reference_tail = ''
+        body = value
+        if ref_heading:
+            ref_match = re.search(rf'(?m)^#\s+{re.escape(ref_heading)}\s*$', value)
+            if ref_match:
+                body = value[:ref_match.start()].rstrip()
+                reference_tail = value[ref_match.start():].strip()
+
+        appendix_match = re.search(rf'(?m)^#\s+{re.escape(appendix_heading)}\s*$', body)
+        if appendix_match:
+            appendix_tail = body[appendix_match.end():]
+            if re.search(r'(?m)^#\s+[^#].*$', appendix_tail):
+                raise RuntimeError('target_qualitative_appendix_rebind_nonterminal_section')
+            body = body[:appendix_match.start()].rstrip()
+
+        pieces = [part for part in (body, appendix, reference_tail) if str(part or '').strip()]
+        return '\n\n'.join(str(part).strip() for part in pieces).strip()
+
+    def _recover_qualitative_quote_ledger(
+        self, agent, *, sections: dict, target_lang: str, translation_glossary: dict,
+        required_tags, stage_callback=None,
+    ):
+        """Recover only missing/stale typed qualitative Quote objects.
+
+        Quote identity remains server-owned.  Recovery is language/tier neutral:
+        required IDs come from the frozen Appendix manifest, source text comes
+        from the frozen evidence ledger, and only missing target objects are
+        translated.  A failed batch is isolated to individual Quote objects so
+        one brittle item cannot discard already recoverable evidence.
+        """
+        updated_sections = sections if isinstance(sections, dict) else {}
+        glossary = dict(translation_glossary or {})
+        quote_sources = dict(getattr(self, '_qualitative_quote_sources', {}) or {})
+        required = {
+            str(tag or '').strip() for tag in (required_tags or []) if str(tag or '').strip()
+        }
+        missing_sources = sorted(tag for tag in required if not str(quote_sources.get(tag) or '').strip())
+        if missing_sources:
+            raise RuntimeError(f'qualitative_quote_source_missing:{missing_sources[:8]}')
+
+        current_quote_hashes = {
+            tag: hashlib.sha256(re.sub(r"\s+", " ", quote).strip().encode("utf-8")).hexdigest()
+            for tag, quote in quote_sources.items()
+        }
+        current_request_hash = str(getattr(self, 'request_hash', '') or '')
+        current_task_id = str(getattr(getattr(self, 'task_dir', None), 'name', '') or '')
+        raw_record = updated_sections.get('__quote_ledger__')
+        record = dict(raw_record) if isinstance(raw_record, dict) else {}
+        identity_ok = self._qualitative_quote_ledger_identity_ok(
+            record,
+            task_id=current_task_id,
+            target_lang=target_lang,
+            request_hash=current_request_hash,
+            translation_contract_version=TRANSLATION_CONTRACT_VERSION,
+            source_quote_hashes=current_quote_hashes,
+        )
+        stored = record.get('translations')
+        quote_map = dict(stored) if identity_ok and isinstance(stored, dict) else {}
+        recovery_events = (
+            list(record.get('recovery_events') or [])
+            if identity_ok and isinstance(record.get('recovery_events'), list) else []
+        )
+
+        def persist():
+            updated_sections['__quote_ledger__'] = {
+                'task_id': current_task_id,
+                '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_map),
+                'recovery_events': list(recovery_events)[-40:],
+                'updated_at': datetime.now(timezone.utc).isoformat(),
+            }
+
+        missing = sorted(tag for tag in required if not str(quote_map.get(tag) or '').strip())
+        if not missing:
+            if required and (not identity_ok or raw_record is not updated_sections.get('__quote_ledger__')):
+                persist()
+            return updated_sections, glossary, quote_map, recovery_events
+
+        translate_quotes = getattr(agent, 'translate_qualitative_quote_objects', None)
+        if not callable(translate_quotes):
+            raise RuntimeError('Typed qualitative quote translation entry point is unavailable.')
+
+        def merge_result(result, expected_tags, route):
+            nonlocal glossary
+            returned = (result or {}).get('translations')
+            expected = set(expected_tags)
+            if not isinstance(returned, dict) or set(returned) != expected:
+                raise RuntimeError(
+                    'Typed qualitative quote translation identity mismatch: '
+                    f'expected={sorted(expected)} got={sorted(returned) if isinstance(returned, dict) else []}'
+                )
+            for tag in expected_tags:
+                rendered = str(returned.get(tag) or '').strip()
+                if not rendered:
+                    raise RuntimeError(f'Typed qualitative quote translation is empty: {tag}')
+                quote_map[tag] = rendered
+            returned_glossary = (result or {}).get('glossary')
+            if isinstance(returned_glossary, dict):
+                glossary = dict(returned_glossary)
+            recovery_events.append({
+                'route': str(route), 'quote_ids': sorted(expected),
+                'at': datetime.now(timezone.utc).isoformat(),
+            })
+            persist()
+
+        # Fast path: one typed batch.  If any object destabilizes the strict
+        # compiler, isolate only the missing objects instead of losing the batch.
+        try:
+            batch_result = translate_quotes(
+                {tag: quote_sources[tag] for tag in missing},
+                target_lang, glossary=glossary, max_attempts=2,
+                stage_callback=stage_callback,
+            )
+            merge_result(batch_result, missing, 'typed_batch')
+        except PipelineCancelledError:
+            raise
+        except Exception as batch_error:
+            logging.warning(
+                'Qualitative Quote Ledger batch recovery isolated | missing=%s error=%s',
+                len(missing), type(batch_error).__name__,
+            )
+            for tag in missing:
+                if str(quote_map.get(tag) or '').strip():
+                    continue
+                source_quote = str(quote_sources.get(tag) or '').strip()
+                try:
+                    one_result = translate_quotes(
+                        {tag: source_quote}, target_lang,
+                        glossary=glossary, max_attempts=2,
+                        stage_callback=stage_callback,
+                    )
+                    merge_result(one_result, [tag], 'typed_single')
+                    continue
+                except PipelineCancelledError:
+                    raise
+                except Exception as strict_error:
+                    fallback_method = getattr(agent, 'adaptive_translate_fallback', None)
+                    if not callable(fallback_method):
+                        raise strict_error
+                    fallback = fallback_method(
+                        source_quote, target_lang,
+                        section_name=f'Qualitative Quote {tag}', glossary=glossary,
+                    )
+                    adaptive_text = str((fallback or {}).get('text') or '').strip()
+                    if not adaptive_text:
+                        raise RuntimeError(f'Adaptive qualitative quote translation is empty: {tag}') from strict_error
+                    self._validate_adaptive_recovery_text(
+                        agent, source_text=source_quote, translated_text=adaptive_text,
+                        target_lang=target_lang, section_name=f'Qualitative Quote {tag}',
+                        glossary=glossary,
+                    )
+                    render_method = getattr(agent, '_render_qualitative_quote_translation', None)
+                    rendered = adaptive_text
+                    if callable(render_method):
+                        rendered = str(render_method(adaptive_text, target_lang) or '').strip()
+                    if not rendered:
+                        raise RuntimeError(f'Adaptive qualitative quote rendering is empty: {tag}') from strict_error
+                    quote_map[tag] = rendered
+                    recovery_events.append({
+                        'route': 'adaptive_single', 'quote_ids': [tag],
+                        'prior_error': type(strict_error).__name__,
+                        'at': datetime.now(timezone.utc).isoformat(),
+                    })
+                    persist()
+
+        unresolved = sorted(tag for tag in required if not str(quote_map.get(tag) or '').strip())
+        if unresolved:
+            raise RuntimeError(f'qualitative_appendix_quote_ledger_incomplete:{unresolved[:8]}')
+        return updated_sections, glossary, quote_map, recovery_events
+
+    def _adaptive_translate_qualitative_unit_with_quotes(
+        self, agent, *, source_text: str, target_lang: str, section_name: str,
+        translation_sections: dict, translation_glossary: dict,
+    ):
+        """Delivery-first fallback that preserves one shared Quote identity.
+
+        Generic whole-unit fallback is allowed for qualitative narrative only
+        after verbatim evidence is replaced by stable Quote slots.  Missing Quote
+        objects are recovered from frozen evidence, then injected back into the
+        fallback result.  This keeps Findings and Appendix on the same translated
+        quotation instead of letting two independent model calls paraphrase the
+        same verbatim evidence differently.
+        """
+        source = str(source_text or '').strip()
+        canonical_source, pending_tags, reused_tags = self._prepare_qualitative_translation_source(
+            source, getattr(self, '_qualitative_quote_sources', {}), {}
+        )
+        canonical_source = str(canonical_source or '').strip()
+        required_tags = list(dict.fromkeys(list(pending_tags or []) + list(reused_tags or [])))
+        sections = translation_sections if isinstance(translation_sections, dict) else {}
+        glossary = dict(translation_glossary or {})
+        quote_map = {}
+        recovery_events = []
+        if required_tags:
+            sections, glossary, quote_map, recovery_events = self._recover_qualitative_quote_ledger(
+                agent, sections=sections, target_lang=target_lang,
+                translation_glossary=glossary, required_tags=required_tags,
+            )
+
+        fallback_method = getattr(agent, 'adaptive_translate_fallback', None)
+        if not callable(fallback_method):
+            raise RuntimeError('Adaptive translation fallback entry point is unavailable.')
+        fallback = fallback_method(
+            canonical_source, target_lang, section_name=section_name, glossary=glossary
+        )
+        translated = str((fallback or {}).get('text') or '').strip()
+        if not translated:
+            raise RuntimeError('adaptive_qualitative_translation_output_missing')
+        if required_tags:
+            translated, quote_map = self._finalize_qualitative_translation_quotes(
+                translated, required_tags, [], quote_map
+            )
+        alignment_validator = getattr(agent, 'validate_translation_alignment', None)
+        if callable(alignment_validator):
+            alignment_validator(source, translated)
+        self._validate_adaptive_recovery_text(
+            agent, source_text=source, translated_text=translated,
+            target_lang=target_lang, section_name=section_name, glossary=glossary,
+        )
+        result = dict(fallback or {})
+        result['text'] = translated
+        if recovery_events:
+            result['qualitative_quote_recovery_events'] = list(recovery_events)[-40:]
+        return result, sections, glossary
+
     def _translate_qualitative_appendix_unit(
         self, agent, qual_engine, *, unit_key: str, source_text: str, target_lang: str,
         translation_sections: dict, translation_glossary: dict, data_metadata: dict,
@@ -3721,12 +4063,17 @@ class GenerationRunner:
             qual_engine.validate_bound_target_appendix(existing['translated_text'], manifest, target_lang)
             return sections, translation_glossary
 
-        quote_record = sections.get('__quote_ledger__')
-        quote_map = dict((quote_record or {}).get('translations') or {}) if isinstance(quote_record, dict) else {}
-        required_quotes = {str(row.get('quote_id') or '') for row in manifest.get('rows') or []}
-        if any(not str(quote_map.get(tag) or '').strip() for tag in required_quotes):
-            missing = sorted(tag for tag in required_quotes if not str(quote_map.get(tag) or '').strip())
-            raise RuntimeError(f'qualitative_appendix_quote_ledger_incomplete:{missing[:6]}')
+        required_quotes = {
+            str(row.get('quote_id') or '').strip()
+            for row in manifest.get('rows') or []
+            if str(row.get('quote_id') or '').strip()
+        }
+        sections, translation_glossary, quote_map, quote_recovery_events = (
+            self._recover_qualitative_quote_ledger(
+                agent, sections=sections, target_lang=target_lang,
+                translation_glossary=translation_glossary, required_tags=required_quotes,
+            )
+        )
 
         translate_cells = getattr(agent, 'translate_qualitative_appendix_cells', None)
         if not callable(translate_cells):
@@ -3735,14 +4082,82 @@ class GenerationRunner:
             prebound_cells, model_cells = qual_engine.appendix_translation_plan(
                 manifest, translation_glossary
             )
+            appendix_cell_recovery_events = []
             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 {})
+                try:
+                    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 {})
+                    if set(model_translations) != set(model_cells):
+                        raise RuntimeError('qualitative_appendix_target_cell_translation_identity_mismatch')
+                    updated_glossary = dict((result or {}).get('glossary') or translation_glossary or {})
+                except PipelineCancelledError:
+                    raise
+                except Exception as batch_error:
+                    logging.warning(
+                        'Qualitative Appendix cell batch recovery isolated | cells=%s error=%s',
+                        len(model_cells), type(batch_error).__name__,
+                    )
+                    model_translations = {}
+                    updated_glossary = dict(translation_glossary or {})
+                    for cell_id, source_cell in model_cells.items():
+                        try:
+                            one_result = translate_cells(
+                                {cell_id: source_cell}, target_lang,
+                                glossary=updated_glossary, max_attempts=2,
+                            )
+                            one_map = dict((one_result or {}).get('translations') or {})
+                            if set(one_map) != {cell_id} or not str(one_map.get(cell_id) or '').strip():
+                                raise RuntimeError(
+                                    f'qualitative_appendix_target_cell_translation_identity_mismatch:{cell_id}'
+                                )
+                            model_translations[cell_id] = str(one_map[cell_id]).strip()
+                            one_glossary = (one_result or {}).get('glossary')
+                            if isinstance(one_glossary, dict):
+                                updated_glossary = dict(one_glossary)
+                            appendix_cell_recovery_events.append({
+                                'route': 'typed_single', 'cell_id': str(cell_id),
+                                'at': datetime.now(timezone.utc).isoformat(),
+                            })
+                        except PipelineCancelledError:
+                            raise
+                        except Exception as strict_error:
+                            fallback_method = getattr(agent, 'adaptive_translate_fallback', None)
+                            if not callable(fallback_method):
+                                raise strict_error
+                            fallback = fallback_method(
+                                str(source_cell or ''), target_lang,
+                                section_name=f'Qualitative Appendix Cell {cell_id}',
+                                glossary=updated_glossary,
+                            )
+                            rendered_cell = re.sub(
+                                r'\s+', ' ', str((fallback or {}).get('text') or '')
+                            ).strip()
+                            if not rendered_cell:
+                                raise RuntimeError(
+                                    f'qualitative_appendix_target_cell_translation_empty:{cell_id}'
+                                ) from strict_error
+                            self._validate_adaptive_recovery_text(
+                                agent, source_text=str(source_cell or ''),
+                                translated_text=rendered_cell, target_lang=target_lang,
+                                section_name=f'Qualitative Appendix Cell {cell_id}',
+                                glossary=updated_glossary,
+                            )
+                            model_translations[cell_id] = rendered_cell
+                            appendix_cell_recovery_events.append({
+                                'route': 'adaptive_single', 'cell_id': str(cell_id),
+                                'prior_error': type(strict_error).__name__,
+                                'at': datetime.now(timezone.utc).isoformat(),
+                            })
+                    result = {
+                        'qa_status': 'passed_with_warning',
+                        'attempts': int(existing.get('attempts') or 0) + 1,
+                        'claim_states': {},
+                        'appendix_cell_contract_version': 'typed-qualitative-appendix-cells-v2',
+                    }
             else:
                 result = {
                     'qa_status': 'passed', 'attempts': 1, 'claim_states': {},
@@ -3802,6 +4217,8 @@ class GenerationRunner:
             '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),
+            'quote_ledger_recovery_events': list(quote_recovery_events)[-40:],
+            'appendix_cell_recovery_events': list(appendix_cell_recovery_events)[-80:],
             'updated_at': datetime.now(timezone.utc).isoformat(),
         }
         return sections, updated_glossary
@@ -7182,12 +7599,31 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                         f'translation_{unit_key}_repair_pending', state, pct,
                         display_node=f'translation_repair_pending:{index}:{len(final_units)}',
                     )
+                    if mode == "asq" and str(unit_key).strip().casefold() == "appendix":
+                        # The Appendix is a manifest-owned typed artifact.  Generic
+                        # whole-unit translation must never bypass its row/cell/Quote
+                        # identity contract.  _translate_qualitative_appendix_unit
+                        # already exhausted bounded typed recovery above.
+                        raise RuntimeError(
+                            'qualitative_appendix_typed_recovery_exhausted:'
+                            + str(state.get('translation_blocking_error_code') or 'TRANSLATION_PENDING')
+                        )
                     # A2: strict claim QA is diagnostic/recovery authority, not a
                     # manuscript kill switch. Escalate the same frozen source to
                     # a whole-unit provider-neutral fallback and keep going.
-                    fallback = translation_agent.adaptive_translate_fallback(
-                        translation_source, target_lang, section_name=unit_name, glossary=translation_glossary
-                    )
+                    if mode == "asq":
+                        fallback, translation_sections, translation_glossary = (
+                            self._adaptive_translate_qualitative_unit_with_quotes(
+                                translation_agent, source_text=translation_source,
+                                target_lang=target_lang, section_name=unit_name,
+                                translation_sections=translation_sections,
+                                translation_glossary=translation_glossary,
+                            )
+                        )
+                    else:
+                        fallback = translation_agent.adaptive_translate_fallback(
+                            translation_source, target_lang, section_name=unit_name, glossary=translation_glossary
+                        )
                     current_translation.update({
                         'translated_text': str(fallback.get('text') or '').strip(),
                         'qa_status': str(fallback.get('qa_status') or 'passed_with_warning'),
@@ -7258,6 +7694,33 @@ The previous attempt for this same manuscript unit was rejected by validation. R
                         quality_notes, [{'type':'target_qualitative_appendix_review','reason':str(exc)[:500]}],
                         section_name='Appendix', stage='translation_final_review'
                     )
+                    # One deterministic self-heal is allowed when the typed
+                    # Appendix record itself is valid. No model is called and no
+                    # row/cell/Quote identity can change during this rebind.
+                    appendix_record = dict(translation_sections.get('appendix') or {})
+                    canonical_appendix = _normalize_target_manuscript_typography(
+                        str(appendix_record.get('translated_text') or '').strip(), target_lang
+                    )
+                    try:
+                        qual_engine.validate_bound_target_appendix(
+                            canonical_appendix, appendix_manifest, target_lang
+                        )
+                        full_draft_translated = self._rebind_target_qualitative_appendix_once(
+                            full_draft_translated, canonical_appendix,
+                            agent._localized_reference_heading(target_lang),
+                        )
+                        qual_engine.validate_bound_target_appendix(
+                            full_draft_translated, appendix_manifest, target_lang
+                        )
+                        self._record_adaptive_completion_residual(
+                            quality_notes,
+                            [{'type':'target_qualitative_appendix_deterministic_rebind','reason':type(exc).__name__}],
+                            section_name='Appendix', stage='translation_final_recovery'
+                        )
+                    except Exception as recovery_exc:
+                        # A structurally invalid/missing typed Appendix is a real
+                        # completion failure, not a cosmetic translation warning.
+                        raise RuntimeError('target_qualitative_appendix_final_validation_failed') from recovery_exc
             try:
                 agent.validate_translation_alignment(full_draft_en, full_draft_translated)
             except Exception as exc: