R14-TQ1-R4-R1: evidence-carrying semantic repair and durable claim recovery

core/generation_runner.py

@@ -871,9 +871,9 @@ class GenerationRunner:
         discovery cannot see them. No semantic synonym merging is performed.
         """
         items = []
-        seen = set()
+        index_by_key = {}
 
-        def add(value, kind):
+        def add(value, kind, policy='FLEXIBLE_TERM'):
             term = re.sub(r'\s+', ' ', str(value or '')).strip(' \t\r\n,;:.')
             if not term or len(term) < 3 or len(term) > 100:
                 return
@@ -881,32 +881,37 @@ class GenerationRunner:
             if not words or len(words) > 10:
                 return
             key = re.sub(r'[–—-]+', '-', term).casefold()
-            if key in seen:
+            normalized_policy = 'STRICT_CONSTRUCT' if str(policy or '').upper() == 'STRICT_CONSTRUCT' else 'FLEXIBLE_TERM'
+            if key in index_by_key:
+                row = items[index_by_key[key]]
+                if normalized_policy == 'STRICT_CONSTRUCT' and row.get('policy') != 'STRICT_CONSTRUCT':
+                    row['policy'] = 'STRICT_CONSTRUCT'
+                    row['kind'] = str(kind or row.get('kind') or 'concept')
                 return
-            seen.add(key)
-            items.append({'source': term, 'kind': str(kind or 'concept')})
+            index_by_key[key] = len(items)
+            items.append({'source': term, 'kind': str(kind or 'concept'), 'policy': normalized_policy})
 
         meta = dict(data_metadata or {})
         if str(mode or '') == 'amj':
             constructs = meta.get('constructs') if isinstance(meta.get('constructs'), dict) else {}
             for role in ('predictor', 'mediator', 'outcome'):
-                add(constructs.get(role), f'quantitative_{role}')
+                add(constructs.get(role), f'quantitative_{role}', 'STRICT_CONSTRUCT')
         else:
             for theme in (meta.get('themes') or []):
                 if not isinstance(theme, dict):
                     continue
-                add(theme.get('title'), 'qualitative_theme')
+                add(theme.get('title'), 'qualitative_theme', 'STRICT_CONSTRUCT')
                 for concept in (theme.get('concepts') or []):
-                    add(concept, 'qualitative_concept')
+                    add(concept, 'qualitative_concept', 'STRICT_CONSTRUCT')
             for row in (meta.get('evidence') or []):
                 if isinstance(row, dict):
-                    add(row.get('concept'), 'qualitative_concept')
+                    add(row.get('concept'), 'qualitative_concept', 'STRICT_CONSTRUCT')
 
         for task in (adaptive_structure or []):
             if not isinstance(task, dict):
                 continue
             for concept in (task.get('concepts') or []):
-                add(concept, 'outline_concept')
+                add(concept, 'outline_concept', 'FLEXIBLE_TERM')
 
         # User keywords remain planning seeds, not canonical identities. Use
         # them only as a fallback when frozen empirical/qualitative metadata and
@@ -915,7 +920,7 @@ class GenerationRunner:
             for seed in re.split(r'[,;\n|]+', str(keywords or '')):
                 seed = re.sub(r'\s+', ' ', seed).strip()
                 if 1 <= len(seed.split()) <= 6:
-                    add(seed, 'keyword_fallback')
+                    add(seed, 'keyword_fallback', 'FLEXIBLE_TERM')
 
         # Capture explicit lower-case hyphenated academic expressions only when
         # they recur or carry a research-concept cue. This closes the
@@ -932,9 +937,9 @@ class GenerationRunner:
         for term in hyphenated:
             parts = set(term.casefold().split('-'))
             if counts.get(term.casefold(), 0) >= 2 or parts.intersection(concept_cues):
-                add(term, 'logic_hyphenated_concept')
+                add(term, 'logic_hyphenated_concept', 'FLEXIBLE_TERM')
 
-        return {'version': 'concept-ledger-v1', 'items': items}
+        return {'version': 'concept-ledger-v2', 'items': items}
 
     @staticmethod
     def _canonical_concept_writer_directive(ledger: dict) -> str:
@@ -1270,15 +1275,22 @@ class GenerationRunner:
                 'glossary': translation_glossary, 'section_name': unit_name, 'max_attempts': 2,
             }
             try:
-                parameters = inspect.signature(translate_method).parameters.values()
+                parameters = list(inspect.signature(translate_method).parameters.values())
                 supports_stage_callback = any(
                     item.name == 'stage_callback' or item.kind == inspect.Parameter.VAR_KEYWORD
                     for item in parameters
                 )
+                supports_claim_states = any(
+                    item.name == 'claim_states' or item.kind == inspect.Parameter.VAR_KEYWORD
+                    for item in parameters
+                )
             except (TypeError, ValueError):
                 supports_stage_callback = False
+                supports_claim_states = False
             if supports_stage_callback:
                 call_kwargs['stage_callback'] = publish_translation_stage
+            if supports_claim_states and isinstance(existing.get('claim_states'), dict):
+                call_kwargs['claim_states'] = dict(existing.get('claim_states') or {})
             result = translate_method(translation_source, target_lang, **call_kwargs)
             translated_text = str(result.get('text') or '').strip()
             if pending_quote_tags or reused_quote_tags:
@@ -1314,6 +1326,8 @@ class GenerationRunner:
                 'fact_ledger_hash': fact_ledger_hash,
                 'translated_text': '',
                 'qa_status': 'repair_pending',
+                'claim_states': dict(getattr(agent, '_last_translation_claim_states', {}) or {}),
+                'claim_state_version': 'translation-claim-state-v1',
                 'error_code': error_code,
                 'attempts': int(existing.get('attempts') or 0) + 1,
                 'deferred_at': datetime.now(timezone.utc).isoformat(),
@@ -1344,6 +1358,8 @@ class GenerationRunner:
             'qa_status': str(result.get('qa_status') or 'passed'),
             'language_adapter': str(result.get('language_adapter') or ''),
             'diagnostics': list(result.get('diagnostics') or [])[-20:],
+            'claim_states': dict(result.get('claim_states') or {}),
+            'claim_state_version': str(result.get('claim_state_version') or 'translation-claim-state-v1'),
             'attempts': int(result.get('attempts') or 1),
             'completed_at': datetime.now(timezone.utc).isoformat(),
         }

core/pipeline_agent.py

@@ -4767,13 +4767,26 @@ Output only the repaired section."""
         mapping = dict(existing or {})
         rows = (dict(concept_ledger or {}).get('items') or []) if isinstance(concept_ledger, dict) else []
         sources = []
+        strict_sources = []
+        flexible_sources = []
         for row in rows:
             if not isinstance(row, dict):
                 continue
             term = re.sub(r'\s+', ' ', str(row.get('source') or '')).strip()
-            if term and term not in sources:
-                sources.append(term)
-        self._canonical_concept_sources = set(sources)
+            if not term or term in sources:
+                continue
+            sources.append(term)
+            policy = str(row.get('policy') or 'STRICT_CONSTRUCT').strip().upper()
+            if policy == 'FLEXIBLE_TERM':
+                flexible_sources.append(term)
+            else:
+                strict_sources.append(term)
+        # Only true research constructs receive deterministic canonical-identity
+        # enforcement.  Descriptive academic terms still receive a preferred
+        # glossary translation but may vary naturally with target-language
+        # grammar or equivalent wording; semantic QA remains authoritative.
+        self._canonical_concept_sources = set(strict_sources)
+        self._canonical_flexible_sources = set(flexible_sources)
         if not sources:
             return mapping
 
@@ -5002,6 +5015,14 @@ SECTION: {section_name}
             'UNCERTAINTY_DRIFT','TERMINOLOGY','GRAMMAR','LANGUAGE_CONTAMINATION',
             'CITATION_ATTACHMENT','ACRONYM_ATTACHMENT','SEMANTIC_DRIFT',
         }
+        allowed_semantic_evidence = {
+            'ACTOR_CHANGED','RELATION_CHANGED','OBJECT_CHANGED','POLARITY_CHANGED',
+            'CAUSALITY_STRENGTH_CHANGED','MODALITY_UNCERTAINTY_CHANGED',
+            'CONDITION_SCOPE_CHANGED','COMPARISON_DIRECTION_CHANGED','TEMPORAL_SCOPE_CHANGED',
+            'CLAIM_MISSING','CLAIM_ADDED','MEANING_CHANGED',
+        }
+        if not isinstance(getattr(self, '_translation_claim_qa_evidence', None), dict):
+            self._translation_claim_qa_evidence = {}
         claim_rows = []
         glossary_value = dict(glossary or {})
         for block_id in expected_ids:
@@ -5040,8 +5061,8 @@ SECTION: {section_name}
 Review EACH claim independently. Do not rewrite and do not provide free-form explanations. Check semantic equivalence and language quality: omissions/additions, negation, uncertainty, association-versus-causation strength, terminology, grammar, language contamination, citation attachment, and acronym attachment.
 LANGUAGE ADAPTER: {profile['adapter_id']}; morphology={profile['morphology']}. {profile['term_policy']} {profile['contamination_policy']}
 For Russian, do not flag a terminology error merely because a translated term is inflected for grammatical case/number/gender. Russian LANGUAGE_CONTAMINATION includes sustained unprotected English prose and ordinary untranslated English lexemes; never use it for protected acronyms, AI/LLM/SEM, model or tool names, variable names, citations, canonical glossary terms, or source-matched proper names/titles. For Chinese, ordinary untranslated English lexical words are contamination while protected technical objects remain legal. For Kazakh, do not flag a terminology error merely because productive suffixes attach to the correct term stem. Kazakh LANGUAGE_CONTAMINATION includes ordinary untranslated English lexemes and clearly Russian lexical/syntactic intrusions not justified by shared international terminology; Cyrillic script or absence of Kazakh-specific letters alone is never sufficient evidence.
-Return valid JSON only in this exact form: {{"items":[{{"id":"SP-P000001-C000001","ok":true,"codes":[]}}]}}.
-If ok=false, codes MUST contain only values from allowed_issue_codes. Never return numbered prose, examples, quotes, or corrective wording. Return every supplied id exactly once."""
+Return valid JSON only in this exact form: {{"items":[{{"id":"SP-P000001-C000001","ok":true,"codes":[],"evidence_codes":[]}}]}}.
+If ok=false, codes MUST contain only values from allowed_issue_codes. For SEMANTIC_DRIFT, NEGATION_DRIFT, CAUSALITY_STRENGTH_DRIFT, UNCERTAINTY_DRIFT, OMITTED_CLAIM, or ADDED_CLAIM, evidence_codes should identify the affected semantic dimension using only: {sorted(allowed_semantic_evidence)}. Use the narrowest supported evidence; do not invent an error dimension. Never return numbered prose, examples, quotes, or corrective wording. Never return replacement text. Return every supplied id exactly once."""
             response = self.call_deepseek(
                 system_prompt, json.dumps(payload, ensure_ascii=False),
                 max_tokens=max(1200, min(3600, 650 + 260 * len(rows))),
@@ -5062,6 +5083,29 @@ If ok=false, codes MUST contain only values from allowed_issue_codes. Never retu
                 meta = row_lookup[row_id]
                 raw_codes = [str(code or '').strip().upper() for code in (returned_row.get('codes') or [])]
                 codes = [code for code in raw_codes if code in allowed_codes]
+                evidence_codes = [
+                    str(code or '').strip().upper() for code in (returned_row.get('evidence_codes') or [])
+                    if str(code or '').strip().upper() in allowed_semantic_evidence
+                ]
+                semantic_code_set = {
+                    'SEMANTIC_DRIFT','NEGATION_DRIFT','CAUSALITY_STRENGTH_DRIFT',
+                    'UNCERTAINTY_DRIFT','OMITTED_CLAIM','ADDED_CLAIM',
+                }
+                if any(code in semantic_code_set for code in codes):
+                    derived = {
+                        'NEGATION_DRIFT': 'POLARITY_CHANGED',
+                        'CAUSALITY_STRENGTH_DRIFT': 'CAUSALITY_STRENGTH_CHANGED',
+                        'UNCERTAINTY_DRIFT': 'MODALITY_UNCERTAINTY_CHANGED',
+                        'OMITTED_CLAIM': 'CLAIM_MISSING',
+                        'ADDED_CLAIM': 'CLAIM_ADDED',
+                    }
+                    for issue_code in codes:
+                        fallback = derived.get(issue_code)
+                        if fallback and fallback not in evidence_codes:
+                            evidence_codes.append(fallback)
+                    self._translation_claim_qa_evidence[(meta['block_id'], meta['claim_id'])] = {
+                        'semantic_delta_codes': evidence_codes[:6],
+                    }
                 model_contamination = 'LANGUAGE_CONTAMINATION' in codes
                 lexical = self._translation_lexical_purity_evidence(
                     meta['english'], meta['target'], target_lang, relevant_glossary
@@ -5473,12 +5517,19 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         ]
         if lexical_tokens:
             evidence_payload['detected_unprotected_english_tokens'] = lexical_tokens[:8]
+        semantic_delta_codes = [
+            str(code or '').strip().upper() for code in (supplied_evidence.get('semantic_delta_codes') or [])
+            if str(code or '').strip()
+        ]
+        if semantic_delta_codes:
+            evidence_payload['semantic_delta_codes'] = semantic_delta_codes[:6]
         effective_prompt = system_prompt + (
             '\nCLAIM QUALITY RETRY. Repair only this one claim. The only permitted repair signals are the following bounded issue codes: '
             + json.dumps(codes[:5], ensure_ascii=False)
             + '. Use the bounded detector evidence below only to repair the supplied claim; do not invent additional defects. '
               'The current target claim is context, not authority: the frozen English source remains authoritative. '
               'If detected_unprotected_english_tokens is non-empty, every listed token is deterministic lexical-purity evidence and MUST be translated or removed from ordinary target-language prose; do not preserve it unless it is represented by a protected SP-T anchor. '
+              'If semantic_delta_codes is non-empty, repair ONLY those evidenced semantic dimensions against the frozen English source and preserve all unaffected actor, relation, object, polarity, modality, condition, comparison, temporal scope, and evidentiary strength. Do not freely paraphrase a claim that only needs a bounded semantic correction. '
               'Preserve every SP-P, SP-C, and SP-T marker byte-for-byte. Do not add claims or factual content. Return only the marked translated claim.'
             + '\n<CLAIM_REPAIR_EVIDENCE>\n'
             + json.dumps(evidence_payload, ensure_ascii=False)
@@ -5553,6 +5604,65 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         _order, segments = self._parse_translation_claim_segments(restored, [claim_id])
         return segments[claim_id]
 
+    @classmethod
+    def _snapshot_translation_claim_states(cls, source_blocks, translations, pending_issues=None, repair_evidence=None):
+        """Freeze claim-level target text so a deferred section never needs wholesale retranslation."""
+        pending = dict(pending_issues or {})
+        evidence_map = dict(repair_evidence or {})
+        states = {}
+        for block_id, source_block in dict(source_blocks or {}).items():
+            if block_id not in dict(translations or {}):
+                continue
+            source_claims = cls._translation_claim_units(source_block)
+            claim_order = [item['claim_id'] for item in source_claims]
+            try:
+                _order, target_claims = cls._parse_translation_claim_segments(translations[block_id], claim_order)
+            except Exception:
+                continue
+            for item in source_claims:
+                claim_id = item['claim_id']
+                target_text = str(target_claims.get(claim_id) or '').strip()
+                if not target_text:
+                    continue
+                codes = list((pending.get(block_id) or {}).get(claim_id) or [])
+                states[f'SP-P{block_id}-{claim_id}'] = {
+                    'block_id': str(block_id),
+                    'claim_id': str(claim_id),
+                    'source_hash': hashlib.sha256(str(item['text']).encode('utf-8')).hexdigest(),
+                    'target_text': target_text,
+                    'qa_status': 'repair_pending' if codes else 'passed',
+                    'issue_codes': [str(code or '').strip().upper() for code in codes if str(code or '').strip()][:5],
+                    'repair_evidence': dict(evidence_map.get((block_id, claim_id)) or {}),
+                }
+        return states
+
+    @classmethod
+    def _restore_translation_claim_states(cls, source_blocks, claim_states):
+        """Reassemble marked blocks only when every frozen source claim has a hash-matched durable target."""
+        states = dict(claim_states or {})
+        if not states:
+            return {}, {}, False
+        translations = {}
+        pending = {}
+        for block_id, source_block in dict(source_blocks or {}).items():
+            source_claims = cls._translation_claim_units(source_block)
+            order = [item['claim_id'] for item in source_claims]
+            segments = {}
+            for item in source_claims:
+                claim_id = item['claim_id']
+                key = f'SP-P{block_id}-{claim_id}'
+                row = states.get(key)
+                expected_hash = hashlib.sha256(str(item['text']).encode('utf-8')).hexdigest()
+                if not isinstance(row, dict) or row.get('source_hash') != expected_hash or not str(row.get('target_text') or '').strip():
+                    return {}, {}, False
+                segments[claim_id] = str(row.get('target_text') or '').strip()
+                if str(row.get('qa_status') or '') == 'repair_pending':
+                    codes = [str(code or '').strip().upper() for code in (row.get('issue_codes') or []) if str(code or '').strip()]
+                    if codes:
+                        pending.setdefault(str(block_id), {})[str(claim_id)] = codes[:5]
+            translations[str(block_id)] = cls._compose_translation_claim_segments(order, segments)
+        return translations, pending, True
+
     def _adjudicate_persistent_translation_issues(
         self, sources, translations, issues, target_lang, glossary=None, section_name=''
     ):
@@ -5740,7 +5850,7 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                 )
         return {}
 
-    def translate_frozen_section(self, section_text, target_lang, glossary=None, section_name='', max_attempts=2, stage_callback=None):
+    def translate_frozen_section(self, section_text, target_lang, glossary=None, section_name='', max_attempts=2, stage_callback=None, claim_states=None):
         """Translate one frozen English unit and publish only after all QA passes."""
         value = str(section_text or '').strip()
         if not value:
@@ -5769,8 +5879,15 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
             + '\nUse this frozen terminology glossary consistently. Never remap an existing term:\n'
             + json.dumps(frozen_glossary, ensure_ascii=False)
         )
+        source_blocks = {
+            block_id: source
+            for batch in batches
+            for block_id, source in batch
+        }
         attempts = max(1, int(max_attempts or 1))
         last_error = None
+        self._last_translation_claim_states = {}
+        self._translation_claim_qa_evidence = {}
 
         def emit_stage(stage):
             if not callable(stage_callback):
@@ -5788,17 +5905,33 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
             emit_stage('translating')
             translations = {}
             try:
-                for batch in batches:
-                    translations.update(self._translate_marked_batch_with_recovery(batch, system_prompt, target_lang))
-                source_blocks = {
-                    block_id: source
-                    for batch in batches
-                    for block_id, source in batch
-                }
-                emit_stage('qa')
-                quality_issues = self._audit_translation_blocks_with_retry(
-                    source_blocks, translations, target_lang, frozen_glossary,
-                    section_name=section_name or 'frozen section', max_attempts=2,
+                translations, quality_issues, resumed_claims = self._restore_translation_claim_states(
+                    source_blocks, claim_states
+                )
+                if resumed_claims:
+                    for row in dict(claim_states or {}).values():
+                        if not isinstance(row, dict):
+                            continue
+                        block_key = str(row.get('block_id') or '')
+                        claim_key = str(row.get('claim_id') or '')
+                        evidence = dict(row.get('repair_evidence') or {})
+                        if block_key and claim_key and evidence:
+                            self._translation_claim_qa_evidence[(block_key, claim_key)] = evidence
+                    logging.info(
+                        'Translation claim-level durable resume | section=%s claims=%s pending=%s',
+                        section_name or 'section', len(dict(claim_states or {})),
+                        sum(len(v or {}) for v in quality_issues.values()),
+                    )
+                else:
+                    for batch in batches:
+                        translations.update(self._translate_marked_batch_with_recovery(batch, system_prompt, target_lang))
+                    emit_stage('qa')
+                    quality_issues = self._audit_translation_blocks_with_retry(
+                        source_blocks, translations, target_lang, frozen_glossary,
+                        section_name=section_name or 'frozen section', max_attempts=2,
+                    )
+                self._last_translation_claim_states = self._snapshot_translation_claim_states(
+                    source_blocks, translations, quality_issues, self._translation_claim_qa_evidence
                 )
                 local_round = 0
                 while quality_issues and local_round < 2:
@@ -5829,7 +5962,15 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                         claim_order = list(source_claims)
                         _order, segments = self._parse_translation_claim_segments(translations[block_id], claim_order)
                         current_target_claim = segments[claim_id]
-                        repair_evidence = {}
+                        repair_evidence = dict(
+                            self._translation_claim_qa_evidence.get((block_id, claim_id)) or {}
+                        )
+                        if repair_evidence.get('semantic_delta_codes'):
+                            self._record_translation_assurance(
+                                'SEMANTIC_REPAIR_EVIDENCE', target_lang=target_lang,
+                                block_id=f'SP-P{block_id}-{claim_id}', severity='repair', decision='REPAIR_LOCAL',
+                                detail={'deltas': '|'.join(repair_evidence.get('semantic_delta_codes') or [])[:180]},
+                            )
                         if 'LANGUAGE_CONTAMINATION' in {str(code or '').strip().upper() for code in (codes or [])}:
                             lexical = self._translation_lexical_purity_evidence(
                                 source_claims[claim_id], current_target_claim, target_lang, frozen_glossary
@@ -5871,12 +6012,18 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                         {block_id: translations[block_id] for block_id in sorted(touched_blocks)},
                         target_lang, frozen_glossary, section_name=section_name or 'frozen section', max_attempts=2,
                     )
+                    self._last_translation_claim_states = self._snapshot_translation_claim_states(
+                        source_blocks, translations, quality_issues, self._translation_claim_qa_evidence
+                    )
                 if quality_issues:
                     emit_stage('qa')
                     quality_issues = self._adjudicate_persistent_translation_issues(
                         source_blocks, translations, quality_issues, target_lang, frozen_glossary,
                         section_name=section_name or 'frozen section',
                     )
+                self._last_translation_claim_states = self._snapshot_translation_claim_states(
+                    source_blocks, translations, quality_issues, self._translation_claim_qa_evidence
+                )
                 if quality_issues:
                     detail = '; '.join(
                         f'SP-P{block_id}-{claim_id}: {",".join(codes)}'
@@ -5913,6 +6060,8 @@ Decision must be CONFIRMED_BLOCKING or NOT_CONFIRMED. evidence_code must be one
                     'fact_ledger_hash': self.translation_fact_ledger_hash(value),
                     'language_adapter': profile['adapter_id'],
                     'diagnostics': diagnostics[-20:],
+                    'claim_states': dict(self._last_translation_claim_states),
+                    'claim_state_version': 'translation-claim-state-v1',
                 }
             except PipelineCancelledError:
                 raise