R6A-R4-R4 four-language translatable-payload gate and async progress observability

core/generation_runner.py

@@ -183,6 +183,12 @@ class GenerationRunner:
         self._translation_worker_collected = 0
         self._translation_worker_pending = set()
         self._translation_worker_context = {}
+        # Display-only observability. These counters never decide task success,
+        # checkpoint authority, quota, or translation reuse.
+        self._translation_progress_sections = {}
+        self._translation_progress_queued = set()
+        self._translation_progress_inflight = set()
+        self._translation_progress_total = 0
         self._qualitative_quote_sources = {}
         if self.preview_token:
             preview_dir = Path(__file__).resolve().parents[1] / "static" / "previews"
@@ -711,6 +717,40 @@ class GenerationRunner:
             raise RuntimeError(f"Unresolved target-language qualitative quote tags: {unresolved[:8]}")
         return value, ledger
 
+    def _translation_progress_snapshot(self) -> dict[str, int]:
+        """Return non-authoritative logical target-unit progress for the UI."""
+        mapping = dict(getattr(self, '_translation_progress_sections', {}) or {})
+        context = getattr(self, '_translation_worker_context', {}) or {}
+        tasks = context.get('adaptive_structure') or [] if isinstance(context, dict) else []
+        logical_keys = []
+        if tasks:
+            logical_keys = ['__front_matter__'] + [
+                key for key, _title in self._unique_top_level_sections(tasks)
+            ]
+            logical_keys = list(dict.fromkeys(key for key in logical_keys if key))
+        total = int(getattr(self, '_translation_progress_total', 0) or len(logical_keys))
+        if total <= 0 and logical_keys:
+            total = len(logical_keys)
+        qa_passed = 0
+        repair_pending = 0
+        inflight = set(getattr(self, '_translation_progress_inflight', set()) or set())
+        for key in logical_keys:
+            record = mapping.get(key) or {}
+            status = str(record.get('qa_status') or '')
+            if key not in inflight and status in {'passed', 'passed_with_warning'} and record.get('translated_text'):
+                qa_passed += 1
+            elif status == 'repair_pending':
+                repair_pending += 1
+        queued = len(set(getattr(self, '_translation_progress_queued', set()) or set()).intersection(logical_keys))
+        in_flight = len(inflight.intersection(logical_keys))
+        return {
+            'qa_passed': min(qa_passed, total) if total else qa_passed,
+            'repair_pending': min(repair_pending, total) if total else repair_pending,
+            'queued': min(queued, total) if total else queued,
+            'in_flight': min(in_flight, total) if total else in_flight,
+            'total': max(0, total),
+        }
+
     def _write_preview_payload(self, *, force: bool = False) -> None:
         with self._preview_lock:
             if self.preview_path is None:
@@ -731,6 +771,7 @@ class GenerationRunner:
                 "target_english_commit_text": self._preview_visible_text(
                     self._preview_target_english_commit_text
                 ),
+                "target_progress": self._translation_progress_snapshot(),
                 "updated_at": datetime.now(timezone.utc).isoformat(),
             }
             temporary = self.preview_path.with_suffix(".json.tmp")
@@ -874,7 +915,18 @@ class GenerationRunner:
             'adaptive_structure': list(adaptive_structure or []),
             'target_lang': str(target_lang or ''),
         }
+        logical_keys = ['__front_matter__'] + [
+            key for key, _title in self._unique_top_level_sections(adaptive_structure)
+        ]
+        logical_keys = list(dict.fromkeys(key for key in logical_keys if key))
+        self._translation_progress_total = len(logical_keys)
+        self._translation_progress_sections = dict(translation_sections or {})
+        self._translation_progress_queued = {
+            key for key in logical_keys if key in self._translation_progress_sections
+        }
+        self._translation_progress_inflight = set()
         self._translation_worker = _OrderedTranslationWorker(self._run_translation_job)
+        self._write_preview_payload(force=True)
 
     def _run_translation_job(self, job: dict[str, Any]) -> dict[str, Any]:
         if self.translation_agent is None:
@@ -912,6 +964,14 @@ class GenerationRunner:
         source_hash = self._text_hash(source)
         fact_ledger_hash = self._translation_ledger_hash(agent, source)
         existing = dict((translation_sections or {}).get(unit_key) or {})
+        if not isinstance(getattr(self, '_translation_progress_queued', None), set):
+            self._translation_progress_queued = set()
+        if not isinstance(getattr(self, '_translation_progress_inflight', None), set):
+            self._translation_progress_inflight = set()
+        # Do not replace the worker-owned display snapshot with a possibly stale
+        # main-thread copy here; collection or the worker completion path updates
+        # _translation_progress_sections monotonically.
+        self._translation_progress_queued.add(str(unit_key))
         if (
             existing.get('qa_status') in {'passed', 'passed_with_warning'}
             and existing.get('source_hash') == source_hash
@@ -924,6 +984,7 @@ class GenerationRunner:
         pending_key = (str(unit_key), source_hash, fact_ledger_hash)
         if pending_key in self._translation_worker_pending:
             return False
+        self._translation_progress_inflight.add(str(unit_key))
         future = self._translation_worker.submit({
             'unit_key': str(unit_key),
             'unit_name': str(unit_name),
@@ -937,6 +998,7 @@ class GenerationRunner:
             'pending_key': pending_key,
             'future': future,
         })
+        self._write_preview_payload(force=True)
         return True
 
     def _collect_translation_results(self, translation_sections, translation_glossary, *, block=False):
@@ -950,6 +1012,7 @@ class GenerationRunner:
             result = future.result()
             sections = dict(result.get('translation_sections') or sections)
             glossary = dict(result.get('translation_glossary') or glossary)
+            self._translation_progress_sections = dict(sections)
             self._translation_worker_pending.discard(item['pending_key'])
             self._translation_worker_collected += 1
         return sections, glossary
@@ -973,6 +1036,12 @@ class GenerationRunner:
         contract_version = TRANSLATION_CONTRACT_VERSION
         fact_ledger_hash = self._translation_ledger_hash(agent, source)
         existing = dict((translation_sections or {}).get(unit_key) or {})
+        if not isinstance(getattr(self, '_translation_progress_queued', None), set):
+            self._translation_progress_queued = set()
+        if not isinstance(getattr(self, '_translation_progress_inflight', None), set):
+            self._translation_progress_inflight = set()
+        self._translation_progress_sections = dict(translation_sections or {})
+        self._translation_progress_queued.add(str(unit_key))
         if (
             existing.get('qa_status') in {'passed', 'passed_with_warning'}
             and existing.get('source_hash') == source_hash
@@ -981,7 +1050,9 @@ class GenerationRunner:
             and existing.get('fact_ledger_hash') == fact_ledger_hash
             and existing.get('translated_text')
         ):
+            self._translation_progress_inflight.discard(str(unit_key))
             return translation_sections, translation_glossary
+        self._translation_progress_inflight.add(str(unit_key))
         partial = self._assemble_partial_translation(translation_sections, adaptive_structure, include_front=include_front)
         self._publish_target_preview(
             partial, f'translating:{unit_name}', force=True, complete=False,
@@ -1089,6 +1160,8 @@ class GenerationRunner:
                 'Translation deferred until final gate | unit=%s code=%s error=%s',
                 unit_name, error_code, type(error).__name__,
             )
+            self._translation_progress_sections = dict(updated_sections)
+            self._translation_progress_inflight.discard(str(unit_key))
             self._publish_target_preview(
                 partial, f'translation_repair_pending:{unit_name}', force=True, complete=False,
                 english_commit_text=english_commit_text,
@@ -1109,6 +1182,8 @@ class GenerationRunner:
             'completed_at': datetime.now(timezone.utc).isoformat(),
         }
         partial = self._assemble_partial_translation(updated_sections, adaptive_structure, include_front=include_front)
+        self._translation_progress_sections = dict(updated_sections)
+        self._translation_progress_inflight.discard(str(unit_key))
         self._publish_target_preview(
             partial, f'translation_ready:{unit_name}', force=True, complete=False,
             english_commit_text=english_commit_text,

core/pipeline_agent.py

@@ -4873,6 +4873,32 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             parsed[match.group(1)] = translated
         return parsed
 
+    @classmethod
+    def _translation_translatable_payload_profile(cls, source):
+        """Classify source content left after Translation-IR immutable anchors.
+
+        Equality is only suspicious when the source still contains lexical
+        payload after server-owned immutable facts/names/IDs/formulas have been
+        replaced by SP-T anchors.  This classifier is source-side and therefore
+        shared by Chinese, Russian, and Kazakh without language-specific
+        exceptions.  English remains the frozen authority.
+        """
+        ir = cls._compile_translation_ir(source)
+        protected = str(ir.get('protected_source') or '')
+        residual = re.sub(r'\[\[SP-(?:C\d{6}|T\d{6})\]\]', ' ', protected)
+        # Markdown structure and punctuation carry no lexical translation
+        # obligation by themselves.  Unicode alphabetic runs are retained so
+        # mixed academic text is never treated as immutable merely because it
+        # is short or contains protected objects.
+        tokens = re.findall(r"[^\W\d_]+(?:['’\-][^\W\d_]+)*", residual, flags=re.UNICODE)
+        lexical = [token for token in tokens if any(char.isalpha() for char in token)]
+        return {
+            'has_translatable_payload': bool(lexical),
+            'payload_tokens': len(lexical),
+            'payload_chars': sum(sum(1 for char in token if char.isalpha()) for token in lexical),
+            'protected_anchors': len(ir.get('fact_ledger') or {}),
+        }
+
     def _validate_translation_block(self, source, translated, block_id, target_lang):
         """Hierarchical multilingual block assurance.
 
@@ -4883,10 +4909,22 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
         source_compact = re.sub(r'\s+', '', str(source or ''))
         target_compact = re.sub(r'\s+', '', str(translated or ''))
         profile = self._translation_language_profile(target_lang)
+        payload_profile = self._translation_translatable_payload_profile(source)
+        has_translatable_payload = bool(payload_profile.get('has_translatable_payload'))
         evidence = []
 
         if source_compact and source_compact.casefold() == target_compact.casefold():
-            evidence.append({'code': 'UNCHANGED_SOURCE', 'severity': 'repair'})
+            if has_translatable_payload:
+                evidence.append({
+                    'code': 'UNCHANGED_SOURCE', 'severity': 'repair',
+                    'payload_tokens': int(payload_profile.get('payload_tokens') or 0),
+                    'payload_chars': int(payload_profile.get('payload_chars') or 0),
+                })
+            else:
+                evidence.append({
+                    'code': 'UNCHANGED_IMMUTABLE_ONLY', 'severity': 'diagnostic',
+                    'protected_anchors': int(payload_profile.get('protected_anchors') or 0),
+                })
 
         if len(source_compact) >= 120:
             minimum = float(APP_CONFIG.get(profile['surface_min_key'], profile['surface_min_default']))
@@ -4897,7 +4935,10 @@ Translate headings and terminology naturally. Preserve standard variable abbrevi
             elif ratio > maximum:
                 evidence.append({'code': 'SURFACE_RATIO_HIGH', 'severity': 'diagnostic', 'ratio': round(ratio, 4), 'threshold': maximum})
 
-        if len(target_compact) >= 40:
+        # A block made entirely of immutable Translation-IR anchors is not
+        # expected to acquire Han/Cyrillic characters. Script presence is
+        # meaningful only when the English source has lexical payload.
+        if len(target_compact) >= 40 and has_translatable_payload:
             if profile['language_code'] == 'zh':
                 han = len(re.findall(r'[\u4e00-\u9fff]', str(translated or '')))
                 if han == 0:

smartpaper_restore_bridge/frontend/index.html

@@ -386,6 +386,14 @@
     } else if (stage === "translation_complete") {
       value = String(map.complete || value);
     }
+    // R6A-R4-R4: keep the approved localized status/layout and append only a
+    // language-neutral logical QA-passed/total fraction. Unapproved target
+    // prose remains hidden until the existing QA publication boundary passes.
+    const targetProgress = payload && payload.target_progress && typeof payload.target_progress === "object"
+      ? payload.target_progress : {};
+    const total = Math.max(0, asCount(targetProgress.total));
+    const qaPassed = Math.max(0, Math.min(total || Number.MAX_SAFE_INTEGER, asCount(targetProgress.qa_passed)));
+    if (total > 0) value = `${value} · ${qaPassed}/${total}`;
     if (!value) return;
     const hasTargetText = Boolean(String(payload && payload.target_text || "").trim()) || Boolean(channel.authoritative);
     if (hasTargetText) {