Spaces:
Running
Running
| # -*- coding: utf-8 -*- | |
| """R3-Free headless SmartPaper generation runner. | |
| The runner contains no Streamlit calls. It is safe to execute in a subprocess | |
| whose lifetime is independent from Streamlit ScriptRunner reruns and browser | |
| connections. Checkpoints are task-scoped, immutable-by-sequence JSON snapshots. | |
| """ | |
| from __future__ import annotations | |
| import hashlib | |
| import inspect | |
| import json | |
| import logging | |
| import math | |
| import os | |
| import queue | |
| import re | |
| import threading | |
| import time | |
| from concurrent.futures import Future | |
| from datetime import datetime, timezone, timedelta | |
| from pathlib import Path | |
| from typing import Callable, Any | |
| from config import APP_CONFIG, AMJ_PROMPTS, ASQ_PROMPTS | |
| from core.checkpoint_store import TaskCheckpointStore | |
| from core.meta_extractor import extract_golden_metadata | |
| from core.pipeline_agent import SSciPipelineAgent, PipelineCancelledError | |
| from core.production_guard import build_validated_package, replace_abstract | |
| from core.qualitative_engine import get_qualitative_engine | |
| from core.quantitative_engine import get_quantitative_engine | |
| from core.rag_engine import initialize_rag | |
| from utils.file_parser import extract_text_from_files | |
| ProgressCallback = Callable[..., None] | |
| CITATION_SEMANTIC_SURVIVAL_RATIO = 0.78 | |
| CITATION_ROLE_RULES = { | |
| "introduction": (3, 180), | |
| "theory": (3, 180), | |
| "theoretical_framing": (3, 180), | |
| "methodology": (2, 280), | |
| "methods": (2, 280), | |
| "discussion": (2, 220), | |
| } | |
| TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v8' | |
| FINAL_CITATION_LEDGER_VERSION = 'resolved-final-ledger-v1' | |
| ENGLISH_SECTION_SEAL_VERSION = 'english-section-integrity-seal-v1' | |
| def _clean_trailing_references(text, keywords=("reference", "references", "reference list", "参考文献")): | |
| lines = str(text or "").split("\n") | |
| for i in range(len(lines) - 1, max(-1, len(lines) - 100), -1): | |
| if any(re.match(rf"^##?\s*{kw}", lines[i], re.IGNORECASE) for kw in keywords): | |
| return "\n".join(lines[:i]).strip() | |
| return str(text or "").strip() | |
| def _clean_markdown_headings(text): | |
| return re.sub(r"(?m)^#+\s+(.*)$", r"**\1**", str(text or "")) | |
| def _canonical_request_hash(request_data: dict[str, Any]) -> str: | |
| material = dict(request_data or {}) | |
| material.pop("request_hash", None) | |
| encoded = json.dumps(material, sort_keys=True, ensure_ascii=False, separators=(",", ":")).encode("utf-8") | |
| return hashlib.sha256(encoded).hexdigest() | |
| def _normalize_target_manuscript_typography(text: str, target_lang: str) -> str: | |
| """Apply only deterministic language-local typography after semantic QA. | |
| R14-TQ2-PQ1-R1 fixes English-style spaces around Chinese full-width | |
| punctuation without touching Russian/Kazakh spacing, references, semantic | |
| content, or code-fenced material. This runs only after the target manuscript | |
| has passed translation QA; it is presentation normalization, not translation. | |
| """ | |
| value = str(text or "") | |
| language = str(target_lang or "") | |
| if "chinese" not in language.casefold() and "中文" not in language: | |
| return value | |
| value = value.replace("\r\n", "\n").replace("\r", "\n") | |
| # The English References tail is immutable data, not target-language prose. | |
| # Chinese punctuation normalization therefore stops before References. | |
| reference_match = re.search(r"(?m)^#\s+参考文献\s*$", value) | |
| frozen_reference_tail = "" | |
| if reference_match: | |
| frozen_reference_tail = value[reference_match.start():] | |
| value = value[:reference_match.start()] | |
| lines = value.split("\n") | |
| normalized = [] | |
| in_fence = False | |
| for line in lines: | |
| stripped = line.lstrip() | |
| if stripped.startswith("```"): | |
| in_fence = not in_fence | |
| normalized.append(line) | |
| continue | |
| if in_fence: | |
| normalized.append(line) | |
| continue | |
| current = re.sub(r"[ \t]+([，。！？；：])", r"\1", line) | |
| current = re.sub(r"([，。！？；：])[ \t]+(?=\S)", r"\1", current) | |
| normalized.append(current) | |
| normalized_body = "\n".join(normalized) | |
| return normalized_body + frozen_reference_tail | |
| class _OrderedTranslationWorker: | |
| """One daemon worker owned by the headless Executor process. | |
| It intentionally has no checkpoint authority. Results are returned through | |
| Future objects and are checkpointed only by GenerationRunner's main thread. | |
| """ | |
| def __init__(self, handler): | |
| self._handler = handler | |
| self._queue = queue.Queue() | |
| self._stop = threading.Event() | |
| # Lifecycle closure and task cancellation are deliberately distinct. | |
| # A graceful drain/close before the strict final translation gate must | |
| # not poison the translation agent with E_TASK_CANCELLED. | |
| self._cancel = threading.Event() | |
| self._thread = threading.Thread( | |
| target=self._run, | |
| name="smartpaper-translation-worker", | |
| daemon=True, | |
| ) | |
| self._thread.start() | |
| def stop_event(self): | |
| return self._stop | |
| def cancel_event(self): | |
| return self._cancel | |
| def submit(self, job: dict[str, Any]) -> Future: | |
| future = Future() | |
| self._queue.put((future, dict(job))) | |
| return future | |
| def _run(self): | |
| while True: | |
| item = self._queue.get() | |
| try: | |
| if item is None: | |
| return | |
| future, job = item | |
| if not future.set_running_or_notify_cancel(): | |
| continue | |
| try: | |
| result = self._handler(job) | |
| except BaseException as error: | |
| future.set_exception(error) | |
| else: | |
| future.set_result(result) | |
| finally: | |
| self._queue.task_done() | |
| def close(self, *, wait: bool = False) -> None: | |
| if self._stop.is_set(): | |
| if wait and self._thread.is_alive(): | |
| self._thread.join(timeout=2.0) | |
| return | |
| # wait=True is the normal final-gate path after all queued work has | |
| # been collected. Only an abortive/non-waiting shutdown signals the | |
| # translation agent to cancel an in-flight model call. | |
| if not wait: | |
| self._cancel.set() | |
| self._stop.set() | |
| while True: | |
| try: | |
| item = self._queue.get_nowait() | |
| except queue.Empty: | |
| break | |
| try: | |
| if item is not None: | |
| future, _job = item | |
| future.cancel() | |
| finally: | |
| self._queue.task_done() | |
| self._queue.put(None) | |
| if wait and self._thread.is_alive(): | |
| self._thread.join(timeout=2.0) | |
| class GenerationRunner: | |
| def __init__( | |
| self, | |
| *, | |
| task_dir: str | os.PathLike, | |
| request_data: dict[str, Any], | |
| execution_epoch: int, | |
| executor_instance_id: str = "", | |
| confirmed_checkpoint_seq: int | None = None, | |
| confirmed_checkpoint_stage: str = "", | |
| progress_callback: ProgressCallback | None = None, | |
| cancel_checker: Callable[[], bool] | None = None, | |
| ): | |
| self.task_dir = Path(task_dir) | |
| self.request = dict(request_data) | |
| self.execution_epoch = int(execution_epoch) | |
| self.executor_instance_id = str(executor_instance_id or "") | |
| self.confirmed_checkpoint_seq = ( | |
| None if confirmed_checkpoint_seq is None else max(0, int(confirmed_checkpoint_seq)) | |
| ) | |
| self.confirmed_checkpoint_stage = str(confirmed_checkpoint_stage or "")[:120] | |
| self.progress_callback = progress_callback or (lambda **_: None) | |
| self.cancel_checker = cancel_checker or (lambda: False) | |
| expected_hash = str(self.request.get("request_hash") or "") | |
| actual_hash = _canonical_request_hash(self.request) | |
| if not expected_hash or expected_hash != actual_hash: | |
| raise RuntimeError("request_hash_mismatch") | |
| self.request_hash = expected_hash | |
| self.checkpoints = TaskCheckpointStore( | |
| self.task_dir, | |
| self.request_hash, | |
| self.execution_epoch, | |
| task_id=str(self.request.get("task_id") or "") if self.executor_instance_id else "", | |
| executor_instance_id=self.executor_instance_id, | |
| confirmed_checkpoint_seq=self.confirmed_checkpoint_seq, | |
| confirmed_checkpoint_stage=self.confirmed_checkpoint_stage, | |
| ) | |
| self.output_dir = self.task_dir / "outputs" | |
| self.output_dir.mkdir(parents=True, exist_ok=True) | |
| self.agent = None | |
| self.translation_agent = None | |
| self._preview_lock = threading.RLock() | |
| preview_token = str(self.request.get("preview_token") or "").strip() | |
| self.preview_token = preview_token if re.fullmatch(r"[A-Za-z0-9_-]{32,128}", preview_token) else "" | |
| self.preview_path = None | |
| self._last_preview_write = 0.0 | |
| self._preview_english_text = "" | |
| self._preview_english_stage = "" | |
| self._preview_english_complete = False | |
| self._preview_target_text = "" | |
| self._preview_target_stage = "translation_waiting" | |
| self._preview_target_complete = False | |
| self._preview_target_english_commit_text = "" | |
| self._translation_worker = None | |
| self._translation_worker_sections = {} | |
| self._translation_worker_glossary = {} | |
| self._translation_worker_jobs = [] | |
| 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 = {} | |
| # R14-TQ2-PQ1-R3-R1: evidence coverage is a quality signal, not a | |
| # manuscript-completeness authority. It is persisted in checkpoints and | |
| # exposed to the browser only as a stable status code. | |
| self._preview_evidence_status = "normal" | |
| self._evidence_coverage_events = [] | |
| if self.preview_token: | |
| preview_dir = Path(__file__).resolve().parents[1] / "static" / "previews" | |
| preview_dir.mkdir(parents=True, exist_ok=True) | |
| self.preview_path = preview_dir / f"{self.preview_token}.json" | |
| # Best-effort cleanup of stale display-only previews; task/checkpoint | |
| # authority is unaffected by this cache. | |
| cutoff = time.time() - 24 * 3600 | |
| for old_preview in preview_dir.glob("*.json"): | |
| try: | |
| if old_preview.stat().st_mtime < cutoff: | |
| old_preview.unlink() | |
| except OSError: | |
| pass | |
| def _compute_request_counts(self) -> tuple[int, int, int]: | |
| """Return UI metrics: DeepSeek requests and eligible literature results.""" | |
| agent = self.agent | |
| if agent is None: | |
| return 0, 0, 0 | |
| metrics = getattr(agent, "source_metrics", {}) or {} | |
| s2_metric = metrics.get("s2", {}) if isinstance(metrics, dict) else {} | |
| oa_metric = metrics.get("oa", {}) if isinstance(metrics, dict) else {} | |
| translator_calls = int(getattr(getattr(self, "translation_agent", None), "ds_calls", 0) or 0) | |
| return ( | |
| int(getattr(agent, "ds_calls", 0) or 0) + translator_calls, | |
| int(s2_metric.get("eligible_result_count", 0) or 0), | |
| int(oa_metric.get("eligible_result_count", 0) or 0), | |
| ) | |
| def _progress(self, pct: int, node: str, *, checkpoint_seq=None, checkpoint_stage=None) -> None: | |
| ds_calls, s2_calls, oa_calls = self._compute_request_counts() | |
| self.progress_callback( | |
| progress_pct=max(0, min(100, int(pct))), | |
| current_node=str(node or "")[:240], | |
| checkpoint_seq=checkpoint_seq, | |
| checkpoint_stage=checkpoint_stage, | |
| ds_calls=ds_calls, | |
| s2_calls=s2_calls, | |
| oa_calls=oa_calls, | |
| ) | |
| def _checkpoint(self, stage: str, state: dict[str, Any], pct: int, *, display_node: str | None = None) -> None: | |
| seq, normalized_stage = self.checkpoints.save(stage, state) | |
| self._progress( | |
| pct, display_node or normalized_stage, | |
| checkpoint_seq=seq, checkpoint_stage=normalized_stage, | |
| ) | |
| def _preview_visible_text(value: str) -> str: | |
| text = str(value or "") | |
| # Never expose hidden reasoning. Raw CITE placeholders are also hidden | |
| # from the user because they have not passed semantic citation QA yet; | |
| # the authoritative manuscript buffer remains untouched. | |
| text = re.sub(r"(?is)<think>.*?(?:</think>|$)", "", text) | |
| text = re.sub(r"\s*\[?CITE-\d+\]?\s*", " ", text, flags=re.I) | |
| text = re.sub(r"[ \t]{2,}", " ", text) | |
| return text.strip() | |
| def _build_qualitative_quote_ownership(tasks, data_metadata): | |
| """Freeze one and only one Findings owner for every qualitative quote.""" | |
| evidence_by_tag = { | |
| str(item.get("tag") or "").strip(): dict(item) | |
| for item in (data_metadata or {}).get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("tag") or "").strip() | |
| } | |
| owners = {} | |
| for task in tasks or []: | |
| if str(task.get("section_id") or "") != "findings": | |
| continue | |
| owner = str(task.get("display_name") or "").strip() | |
| for item in task.get("theme_evidence") or []: | |
| if not isinstance(item, dict): | |
| continue | |
| tag = str(item.get("tag") or "").strip() | |
| if not tag: | |
| continue | |
| if tag not in evidence_by_tag: | |
| raise RuntimeError(f"Qualitative quotation ownership references unknown tag: {tag}") | |
| if tag in owners: | |
| raise RuntimeError( | |
| f"Qualitative quotation has multiple Findings owners: {tag} -> {owners[tag]!r}, {owner!r}" | |
| ) | |
| owners[tag] = owner | |
| missing = sorted(set(evidence_by_tag) - set(owners)) | |
| if missing: | |
| raise RuntimeError(f"Qualitative quotations lack a Findings owner: {missing[:8]}") | |
| return owners | |
| def _qualitative_task_key(task): | |
| task = task or {} | |
| explicit = str(task.get("task_key") or "").strip() | |
| if explicit: | |
| return explicit | |
| section_id = str(task.get("section_id") or task.get("section") or "section").strip() | |
| stable_unit = str(task.get("theme_id") or task.get("subsection_id") or section_id).strip() | |
| part_index = int(task.get("part_index") or 0) + 1 | |
| return f"{section_id}:{stable_unit}:part:{part_index:02d}" | |
| def _build_qualitative_quote_invocation_plan(cls, tasks, data_metadata): | |
| """Freeze one deterministic body invocation for every retained quote. | |
| Logical ownership is a stable Theme object. Runtime Part chunks only | |
| select where the already-owned quote is rendered; they never become the | |
| quote owner and can never change cardinality. | |
| """ | |
| metadata = data_metadata or {} | |
| evidence_by_tag = { | |
| str(item.get("tag") or "").strip(): dict(item) | |
| for item in metadata.get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("tag") or "").strip() | |
| } | |
| theme_id_by_title = {} | |
| for index, theme in enumerate(metadata.get("themes") or [], start=1): | |
| if not isinstance(theme, dict): | |
| continue | |
| title = re.sub(r"\s+", " ", str(theme.get("title") or "")).strip().casefold() | |
| if title: | |
| theme_id_by_title[title] = str(theme.get("theme_id") or f"THEME-{index:02d}").strip() | |
| plan = {} | |
| task_slots = {} | |
| for task in tasks or []: | |
| if str(task.get("section_id") or "") != "findings": | |
| continue | |
| task_key = cls._qualitative_task_key(task) | |
| task_theme_id = str(task.get("theme_id") or "").strip() | |
| task_theme_title = str(task.get("theme_title") or task.get("subsection") or "").strip() | |
| for item in task.get("theme_evidence") or []: | |
| if not isinstance(item, dict): | |
| continue | |
| tag = str(item.get("tag") or "").strip() | |
| if not tag: | |
| continue | |
| evidence = evidence_by_tag.get(tag) | |
| if evidence is None: | |
| raise RuntimeError(f"Qualitative quote plan references unknown evidence: {tag}") | |
| if tag in plan: | |
| raise RuntimeError(f"Qualitative quote plan has duplicate render assignment: {tag}") | |
| owner_theme_id = str( | |
| task_theme_id | |
| or evidence.get("theme_id") | |
| or theme_id_by_title.get( | |
| re.sub(r"\s+", " ", str(evidence.get("theme") or task_theme_title)).strip().casefold() | |
| ) | |
| or "" | |
| ).strip() | |
| if not owner_theme_id: | |
| raise RuntimeError(f"Qualitative quote lacks a stable theme owner: {tag}") | |
| quote = str(evidence.get("quote") or "").strip() | |
| if not quote: | |
| raise RuntimeError(f"Qualitative quote is empty: {tag}") | |
| quote_hash = str(evidence.get("quote_hash") or "").strip() or hashlib.sha256( | |
| re.sub(r"\s+", " ", quote).strip().encode("utf-8") | |
| ).hexdigest() | |
| slot_ordinal = task_slots.get(task_key, 0) + 1 | |
| task_slots[task_key] = slot_ordinal | |
| plan[tag] = { | |
| "quote_id": tag, | |
| "quote_hash": quote_hash, | |
| "owner_theme_id": owner_theme_id, | |
| "owner_theme_title": str(evidence.get("theme") or task_theme_title).strip(), | |
| "render_task_key": task_key, | |
| "render_part_index": int(task.get("part_index") or 0), | |
| "render_part_count": int(task.get("part_count") or 1), | |
| "slot_ordinal": slot_ordinal, | |
| "cardinality": 1, | |
| } | |
| missing = sorted(set(evidence_by_tag) - set(plan)) | |
| extra = sorted(set(plan) - set(evidence_by_tag)) | |
| if missing or extra: | |
| raise RuntimeError( | |
| f"Qualitative quote invocation plan mismatch: missing={missing[:8]} extra={extra[:8]}" | |
| ) | |
| cls._validate_qualitative_quote_invocation_plan(plan, tasks, metadata) | |
| return plan | |
| def _validate_qualitative_quote_invocation_plan(cls, plan, tasks, data_metadata): | |
| metadata = data_metadata or {} | |
| evidence_by_tag = { | |
| str(item.get("tag") or "").strip(): dict(item) | |
| for item in metadata.get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("tag") or "").strip() | |
| } | |
| task_by_key = {cls._qualitative_task_key(task): task for task in tasks or []} | |
| if set(plan or {}) != set(evidence_by_tag): | |
| raise RuntimeError("Qualitative quote invocation plan does not match the frozen evidence ledger.") | |
| for tag, record in (plan or {}).items(): | |
| if int(record.get("cardinality") or 0) != 1: | |
| raise RuntimeError(f"Qualitative quote cardinality must equal one: {tag}") | |
| owner_theme_id = str(record.get("owner_theme_id") or "").strip() | |
| if not owner_theme_id or "part " in owner_theme_id.casefold(): | |
| raise RuntimeError(f"Qualitative quote owner is not a stable Theme object: {tag}") | |
| task_key = str(record.get("render_task_key") or "").strip() | |
| task = task_by_key.get(task_key) | |
| if not task or str(task.get("section_id") or "") != "findings": | |
| raise RuntimeError(f"Qualitative quote render task is invalid: {tag} -> {task_key!r}") | |
| task_theme_id = str(task.get("theme_id") or "").strip() | |
| if task_theme_id and task_theme_id != owner_theme_id: | |
| raise RuntimeError(f"Qualitative quote render task crossed theme ownership: {tag}") | |
| evidence = evidence_by_tag[tag] | |
| quote = str(evidence.get("quote") or "").strip() | |
| expected_hash = str(evidence.get("quote_hash") or "").strip() or hashlib.sha256( | |
| re.sub(r"\s+", " ", quote).strip().encode("utf-8") | |
| ).hexdigest() | |
| if str(record.get("quote_hash") or "") != expected_hash: | |
| raise RuntimeError(f"Qualitative quote hash drifted after planning: {tag}") | |
| return True | |
| def _strip_model_controlled_qualitative_quotes(text, data_metadata): | |
| """Remove model-emitted quote controls/literals before Executor rendering. | |
| This is a containment boundary, not the quote control plane: quotation | |
| presence and cardinality come exclusively from the frozen invocation | |
| plan. The model's attempted tags/literals therefore have zero authority. | |
| """ | |
| value = str(text or "") | |
| value = re.sub(r"(?<!\w)\[?QUOTE-\d+\]?(?!\w)", "", value, flags=re.I) | |
| quotes = sorted( | |
| { | |
| str(item.get("quote") or "").strip() | |
| for item in (data_metadata or {}).get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("quote") or "").strip() | |
| }, | |
| key=len, | |
| reverse=True, | |
| ) | |
| for quote in quotes: | |
| for wrapped in (f'"{quote}"', f'“{quote}”', f'«{quote}»', quote): | |
| value = value.replace(wrapped, "") | |
| value = re.sub(r"[ \t]+([,.;:!?])", r"\1", value) | |
| value = re.sub(r"([,;:])\s*([.!?])", r"\2", value) | |
| value = re.sub(r"[ \t]{2,}", " ", value) | |
| value = re.sub(r"\n[ \t]+", "\n", value) | |
| return value.strip() | |
| def _render_planned_qualitative_quotes(cls, text, *, task, data_metadata, quote_invocation_plan): | |
| """Deterministically insert frozen quotations into one rendered task.""" | |
| value = cls._strip_model_controlled_qualitative_quotes(text, data_metadata) | |
| task_key = cls._qualitative_task_key(task) | |
| scheduled = [ | |
| dict(record) | |
| for record in (quote_invocation_plan or {}).values() | |
| if str(record.get("render_task_key") or "") == task_key | |
| ] | |
| scheduled.sort(key=lambda record: (int(record.get("slot_ordinal") or 0), str(record.get("quote_id") or ""))) | |
| if not scheduled: | |
| return value | |
| evidence_by_tag = { | |
| str(item.get("tag") or "").strip(): dict(item) | |
| for item in (data_metadata or {}).get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("tag") or "").strip() | |
| } | |
| blocks = [block.strip() for block in re.split(r"\n\s*\n", value) if block.strip()] | |
| anchor_indices = [ | |
| index for index, block in enumerate(blocks) | |
| if not re.match(r"^(?:#{1,6}\s|\||[-*+]\s|>)", block) | |
| ] | |
| if not anchor_indices: | |
| anchor_indices = [len(blocks) - 1] if blocks else [] | |
| insertions = {} | |
| for index, record in enumerate(scheduled): | |
| tag = str(record.get("quote_id") or "").strip() | |
| quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip() | |
| if not quote: | |
| raise RuntimeError(f"Planned qualitative quotation is unavailable at render time: {tag}") | |
| if anchor_indices: | |
| anchor_pos = min( | |
| len(anchor_indices) - 1, | |
| max(0, int(((index + 1) * len(anchor_indices)) / (len(scheduled) + 1))), | |
| ) | |
| block_index = anchor_indices[anchor_pos] | |
| insertions.setdefault(block_index, []).append(f'"{quote}"') | |
| else: | |
| blocks.append(f'"{quote}"') | |
| rendered_blocks = [] | |
| for index, block in enumerate(blocks): | |
| rendered_blocks.append(block) | |
| rendered_blocks.extend(insertions.get(index, [])) | |
| rendered = "\n\n".join(rendered_blocks).strip() | |
| for record in scheduled: | |
| tag = str(record.get("quote_id") or "").strip() | |
| quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip() | |
| if rendered.count(quote) != 1: | |
| raise RuntimeError(f"Qualitative deterministic quote rendering failed: {tag}") | |
| return rendered | |
| def _validate_qualitative_quote_rendering(text, quote_invocation_plan, data_metadata): | |
| body_text = re.split( | |
| r"(?m)^#\s+Appendix(?:[:\s].*)?$", str(text or ""), maxsplit=1 | |
| )[0] | |
| unresolved = sorted(set(re.findall(r"\bQUOTE-\d+\b", body_text, flags=re.I))) | |
| if unresolved: | |
| raise RuntimeError(f"Unresolved qualitative quote controls in body: {unresolved[:8]}") | |
| normalized_body = re.sub(r"\s+", " ", body_text).casefold() | |
| evidence_by_tag = { | |
| str(item.get("tag") or "").strip(): dict(item) | |
| for item in (data_metadata or {}).get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("tag") or "").strip() | |
| } | |
| for tag, record in (quote_invocation_plan or {}).items(): | |
| quote = re.sub(r"\s+", " ", str((evidence_by_tag.get(tag) or {}).get("quote") or "")).strip().casefold() | |
| expected = int(record.get("cardinality") or 0) | |
| actual = normalized_body.count(quote) if quote else 0 | |
| if actual != expected: | |
| raise RuntimeError( | |
| f"Qualitative quote invocation invariant failed: {tag} actual={actual} planned={expected}" | |
| ) | |
| return True | |
| def _redact_qualitative_quotes(text, data_metadata): | |
| """Hide prior verbatim quote text from cross-unit continuity context.""" | |
| value = str(text or "") | |
| quotes = sorted( | |
| { | |
| str(item.get("quote") or "").strip() | |
| for item in (data_metadata or {}).get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("quote") or "").strip() | |
| }, | |
| key=len, | |
| reverse=True, | |
| ) | |
| for quote in quotes: | |
| value = value.replace(f'"{quote}"', "[prior qualitative quotation omitted]") | |
| value = value.replace(quote, "[prior qualitative quotation omitted]") | |
| return value | |
| def _bind_owned_qualitative_quotes( | |
| cls, | |
| text, | |
| *, | |
| display_name, | |
| data_metadata, | |
| quote_ownership, | |
| prior_manuscript="", | |
| ): | |
| """Bind only the current Findings unit's owned quotation tags. | |
| Other manuscript units are not allowed to contain frozen quote text or | |
| quote tags. A direct verbatim reproduction of an owned quote is | |
| normalized back to its tag before deterministic binding, preserving the | |
| frozen wording without letting model prose create a second copy. | |
| """ | |
| value = str(text or "") | |
| evidence_by_tag = { | |
| str(item.get("tag") or "").strip(): dict(item) | |
| for item in (data_metadata or {}).get("evidence", []) | |
| if isinstance(item, dict) and str(item.get("tag") or "").strip() | |
| } | |
| allowed = { | |
| tag for tag, owner in (quote_ownership or {}).items() | |
| if str(owner) == str(display_name) | |
| } | |
| raw_tags = set(re.findall(r"\bQUOTE-\d+\b", value)) | |
| unauthorized_tags = sorted(raw_tags - allowed) | |
| if unauthorized_tags: | |
| raise RuntimeError( | |
| f"Qualitative quotation tag used outside owning Findings unit {display_name!r}: {unauthorized_tags[:8]}" | |
| ) | |
| # Quote text itself is also fenced. Owned verbatim text is converted | |
| # back to a tag and then bound once; unowned verbatim text fails early. | |
| for tag, item in evidence_by_tag.items(): | |
| quote = str(item.get("quote") or "").strip() | |
| if not quote or quote not in value: | |
| continue | |
| if tag not in allowed: | |
| raise RuntimeError( | |
| f"Qualitative quotation text used outside owning Findings unit: {tag} in {display_name!r}" | |
| ) | |
| value = value.replace(f'"{quote}"', tag) | |
| value = value.replace(quote, tag) | |
| # A quote already present in frozen prior prose cannot be rebound later, | |
| # even after resume. This makes ownership idempotent without new state. | |
| normalized_prior = re.sub(r"\s+", " ", str(prior_manuscript or "")).casefold() | |
| for tag in sorted(allowed): | |
| quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip() | |
| if quote and re.sub(r"\s+", " ", quote).casefold() in normalized_prior: | |
| raise RuntimeError(f"Qualitative quotation already used before owning unit: {tag}") | |
| for tag in sorted(allowed): | |
| tag_pattern = rf"(?<!\w)\[?{re.escape(tag)}\]?(?!\w)" | |
| count = len(re.findall(tag_pattern, value)) | |
| if count > 1: | |
| raise RuntimeError( | |
| f"Qualitative quotation repeated within owning Findings unit: {tag} count={count}" | |
| ) | |
| if count == 1: | |
| quote = str((evidence_by_tag.get(tag) or {}).get("quote") or "").strip() | |
| value = re.sub( | |
| tag_pattern, | |
| lambda _match, frozen_quote=quote: f'"{frozen_quote}"', | |
| value, | |
| count=1, | |
| ) | |
| # No unresolved quote tag may escape into citation QA or final output. | |
| unresolved = sorted(set(re.findall(r"\bQUOTE-\d+\b", value))) | |
| if unresolved: | |
| raise RuntimeError(f"Unresolved qualitative quotation tags: {unresolved[:8]}") | |
| return value | |
| def _qualitative_quote_source_map(data_metadata): | |
| return { | |
| str(item.get("tag") or "").strip(): str(item.get("quote") or "").strip() | |
| for item in (data_metadata or {}).get("evidence", []) | |
| if isinstance(item, dict) | |
| and str(item.get("tag") or "").strip() | |
| and str(item.get("quote") or "").strip() | |
| } | |
| def _qualitative_quote_ledger_identity_ok( | |
| record, *, target_lang: str, request_hash: str, source_quote_hashes: dict, | |
| task_id: str = "", translation_contract_version: str = TRANSLATION_CONTRACT_VERSION, | |
| ) -> bool: | |
| """Type-safe identity gate for task-level target-language Quote reuse. | |
| Missing scalar identity fields remain backward-compatible with pre-R4 | |
| checkpoints. Present fields must be strings and match exactly. Quote | |
| hashes must be a mapping and match the current frozen English evidence. | |
| execution_epoch is deliberately not part of logical identity so a newer | |
| epoch can reuse the same QA-passed target rendering for the same task. | |
| """ | |
| if not isinstance(record, dict) or not isinstance(source_quote_hashes, dict): | |
| return False | |
| def scalar_matches(stored, current: str) -> bool: | |
| if stored is None or stored == "": | |
| return True | |
| return isinstance(stored, str) and stored == str(current or "") | |
| stored_hashes = record.get("source_quote_hashes") | |
| if stored_hashes is None or stored_hashes == "": | |
| hashes_match = True | |
| else: | |
| hashes_match = isinstance(stored_hashes, dict) and stored_hashes == source_quote_hashes | |
| return ( | |
| scalar_matches(record.get("task_id"), task_id) | |
| and scalar_matches(record.get("target_lang"), target_lang) | |
| and scalar_matches(record.get("request_hash"), request_hash) | |
| and scalar_matches(record.get("translation_contract_version"), translation_contract_version) | |
| and hashes_match | |
| ) | |
| def _prepare_qualitative_translation_source(cls, source, quote_sources, quote_ledger): | |
| """Replace every frozen qualitative quote with a typed semantic slot. | |
| 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 = [] | |
| reused = [] | |
| for tag, quote in sorted((quote_sources or {}).items()): | |
| wrapped = f'"{quote}"' | |
| tagged = f'{wrapped} [{tag}]' | |
| if tagged in value: | |
| value = value.replace(tagged, f"[{tag}]") | |
| (reused if tag in (quote_ledger or {}) else pending).append(tag) | |
| continue | |
| if wrapped not in value: | |
| continue | |
| value = value.replace(wrapped, f"[{tag}]") | |
| (reused if tag in (quote_ledger or {}) else pending).append(tag) | |
| return value, pending, reused | |
| def _finalize_qualitative_translation_quotes(cls, translated, pending, reused, quote_ledger): | |
| value = str(translated or "") | |
| ledger = dict(quote_ledger 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}") | |
| pattern = rf"(?<!\w)\[?{re.escape(tag)}\]?(?!\w)" | |
| count = len(re.findall(pattern, value)) | |
| if count != 1: | |
| raise RuntimeError( | |
| f"Qualitative quote deterministic target slot injection failed: {tag} matches={count}" | |
| ) | |
| value = re.sub(pattern, lambda _m, q=rendered_quote: q, value, count=1) | |
| unresolved = sorted(set(re.findall(r"\bQUOTE-\d+\b", value))) | |
| if unresolved: | |
| 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: | |
| return | |
| now = time.monotonic() | |
| if not force and now - self._last_preview_write < 0.45: | |
| return | |
| payload = { | |
| "task_id": str(self.request.get("task_id") or ""), | |
| "request_hash": self.request_hash, | |
| "execution_epoch": self.execution_epoch, | |
| "stage": str(self._preview_english_stage or "")[:240], | |
| "text": self._preview_visible_text(self._preview_english_text), | |
| "complete": bool(self._preview_english_complete), | |
| "target_stage": str(self._preview_target_stage or "")[:240], | |
| "target_text": self._preview_visible_text(self._preview_target_text), | |
| "target_complete": bool(self._preview_target_complete), | |
| "target_english_commit_text": self._preview_visible_text( | |
| self._preview_target_english_commit_text | |
| ), | |
| "target_progress": self._translation_progress_snapshot(), | |
| "evidence_status": str(self._preview_evidence_status or "normal")[:40], | |
| "updated_at": datetime.now(timezone.utc).isoformat(), | |
| } | |
| temporary = self.preview_path.with_suffix(".json.tmp") | |
| temporary.write_text(json.dumps(payload, ensure_ascii=False), encoding="utf-8") | |
| os.replace(temporary, self.preview_path) | |
| self._last_preview_write = now | |
| def _publish_preview(self, text: str, stage: str, *, force: bool = False, complete: bool = False) -> None: | |
| with self._preview_lock: | |
| self._preview_english_text = str(text or "") | |
| self._preview_english_stage = str(stage or "") | |
| self._preview_english_complete = bool(complete) | |
| self._write_preview_payload(force=force) | |
| def _publish_target_preview( | |
| self, text: str, stage: str, *, force: bool = True, complete: bool = False, | |
| english_commit_text: str | None = None, | |
| ) -> None: | |
| with self._preview_lock: | |
| self._preview_target_text = str(text or "") | |
| self._preview_target_stage = str(stage or "") | |
| self._preview_target_complete = bool(complete) | |
| if english_commit_text is not None: | |
| self._preview_target_english_commit_text = str(english_commit_text or "") | |
| self._write_preview_payload(force=force) | |
| def _stream_text(self, generator, *, preview_prefix: str | None = None, preview_stage: str = "") -> str: | |
| chunks = [] | |
| for chunk in generator: | |
| value = str(chunk or "") | |
| if not value: | |
| continue | |
| chunks.append(value) | |
| if preview_prefix is not None: | |
| self._publish_preview(preview_prefix + "".join(chunks), preview_stage, force=False) | |
| text = "".join(chunks) | |
| if preview_prefix is not None: | |
| self._publish_preview(preview_prefix + text, preview_stage, force=True) | |
| return text | |
| def _text_hash(value: str) -> str: | |
| normalized = re.sub(r'\s+', ' ', str(value or '')).strip() | |
| return hashlib.sha256(normalized.encode('utf-8')).hexdigest() | |
| def _unique_top_level_sections(tasks): | |
| output = [] | |
| seen = set() | |
| for task in tasks or []: | |
| key = str(task.get('section_id') or task.get('section') or '').strip() | |
| title = str(task.get('section') or '').strip() | |
| if key and title and key not in seen: | |
| seen.add(key); output.append((key, title)) | |
| return output | |
| def _extract_top_level_section(manuscript: str, title: str) -> str: | |
| value = str(manuscript or '') | |
| match = re.search( | |
| rf'(?ms)^#\s+{re.escape(str(title))}\s*$\n(.*?)(?=^#\s+[^#].*?$|\Z)', value | |
| ) | |
| if not match: | |
| return '' | |
| return f"# {title}\n" + match.group(1).strip() | |
| def _english_prefix_through_top_level(manuscript: str, title: str) -> str: | |
| value = str(manuscript or '') | |
| match = re.search( | |
| rf'(?ms)^#\s+{re.escape(str(title))}\s*$\n.*?(?=^#\s+[^#].*?$|\Z)', value | |
| ) | |
| return value[:match.end()].strip() if match else value.strip() | |
| def _extract_front_matter(manuscript: str, first_section_title: str) -> str: | |
| value = str(manuscript or '') | |
| match = re.search(rf'(?m)^#\s+{re.escape(str(first_section_title))}\s*$', value) | |
| return value[:match.start()].strip() if match else '' | |
| def _extract_body_from_first_section(manuscript: str, first_section_title: str) -> str: | |
| value = str(manuscript or '') | |
| match = re.search(rf'(?m)^#\s+{re.escape(str(first_section_title))}\s*$', value) | |
| return value[match.start():].strip() if match else value.strip() | |
| def _repair_validate_study_fact_contract( | |
| mode: str, agent, quant_engine, text: str, data_metadata: dict, *, section_name: str | |
| ): | |
| value = str(text or '').strip() | |
| diagnostics = [] | |
| if str(mode or '') == 'asq': | |
| value, diagnostics = agent.repair_qualitative_study_fact_contract( | |
| value, data_metadata, section_name=section_name | |
| ) | |
| agent.validate_qualitative_study_fact_contract( | |
| value, data_metadata, section_name=section_name | |
| ) | |
| elif str(mode or '') == 'amj': | |
| value, diagnostics = quant_engine.repair_quantitative_study_fact_contract( | |
| value, data_metadata, section_name=section_name | |
| ) | |
| quant_engine.validate_quantitative_study_fact_contract( | |
| value, data_metadata, section_name=section_name | |
| ) | |
| return value, list(diagnostics or []) | |
| def _section_integrity_hash(value: str) -> str: | |
| exact = str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip() | |
| return hashlib.sha256(exact.encode("utf-8")).hexdigest() | |
| def _build_section_integrity_seal(cls, section_key: str, section_title: str, section_text: str) -> dict: | |
| return { | |
| "version": ENGLISH_SECTION_SEAL_VERSION, | |
| "section_key": str(section_key or ""), | |
| "section_title": str(section_title or ""), | |
| "sha256": cls._section_integrity_hash(section_text), | |
| } | |
| def _replace_top_level_section(manuscript: str, title: str, replacement: str) -> str: | |
| value = str(manuscript or "") | |
| pattern = re.compile( | |
| rf'(?ms)^#\s+{re.escape(str(title))}\s*$\n.*?(?=^#\s+[^#].*?$|\Z)' | |
| ) | |
| matches = list(pattern.finditer(value)) | |
| if len(matches) != 1: | |
| raise RuntimeError(f"top_level_section_replace_count_invalid:{title}:{len(matches)}") | |
| match = matches[0] | |
| rendered = str(replacement or "").strip() | |
| return (value[:match.start()] + rendered + "\n\n" + value[match.end():].lstrip("\n")).strip() + "\n\n" | |
| def _validate_section_integrity_seal(cls, seal: dict, section_text: str, section_title: str) -> None: | |
| record = dict(seal or {}) | |
| if record.get("version") != ENGLISH_SECTION_SEAL_VERSION: | |
| raise RuntimeError(f"english_section_integrity_seal_version_invalid:{section_title}") | |
| expected = str(record.get("sha256") or "") | |
| actual = cls._section_integrity_hash(section_text) | |
| if not expected or expected != actual: | |
| raise RuntimeError(f"english_section_integrity_seal_mismatch:{section_title}") | |
| def _replace_keywords(manuscript: str, keywords) -> str: | |
| value = str(manuscript or '') | |
| if isinstance(keywords, (list, tuple)): | |
| rendered = ', '.join(str(item).strip() for item in keywords if str(item).strip()) | |
| else: | |
| rendered = str(keywords or '').strip() | |
| pattern = re.compile(r'(?ms)(^##\s+Keywords\s*$\n)(.*?)(?=^#\s+[^#].*?$|\Z)') | |
| matches = list(pattern.finditer(value)) | |
| if len(matches) != 1: | |
| raise RuntimeError(f'keywords_heading_count_invalid:{len(matches)}') | |
| match = matches[0] | |
| return value[:match.start()] + match.group(1) + rendered + '\n\n' + value[match.end():].lstrip('\n') | |
| def _section_is_complete(tasks, completed_display_names, section_key: str) -> bool: | |
| relevant = [ | |
| str(task.get('display_name') or '') for task in (tasks or []) | |
| if str(task.get('section_id') or task.get('section') or '') == str(section_key) | |
| ] | |
| done = set(str(item) for item in (completed_display_names or [])) | |
| return bool(relevant) and all(name in done for name in relevant) | |
| def _glossary_version(glossary) -> str: | |
| encoded = json.dumps(dict(glossary or {}), ensure_ascii=False, sort_keys=True).encode('utf-8') | |
| return hashlib.sha256(encoded).hexdigest()[:16] | |
| def _build_canonical_concept_ledger( | |
| mode: str, data_metadata: dict, *, keywords='', adaptive_structure=None, research_logic_contract='' | |
| ) -> dict: | |
| """Freeze task-level research concept identity before body translation. | |
| This ledger is intentionally conservative: quantitative construct names | |
| come directly from frozen metadata; qualitative concept/theme labels come | |
| from the Evidence Ledger/outline. Lower-case hyphenated concepts used in | |
| the frozen research logic are added because the legacy Title-Case glossary | |
| discovery cannot see them. No semantic synonym merging is performed. | |
| """ | |
| items = [] | |
| index_by_key = {} | |
| 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 | |
| words = re.findall(r"[A-Za-zÀ-ÖØ-öø-ÿА-Яа-яЁёӘәҒғҚқҢңӨөҰұҮүҺһІі][\w’'\-]*", term, flags=re.UNICODE) | |
| if not words or len(words) > 10: | |
| return | |
| key = re.sub(r'[–—-]+', '-', term).casefold() | |
| 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 | |
| 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}', 'STRICT_CONSTRUCT') | |
| else: | |
| for theme in (meta.get('themes') or []): | |
| if not isinstance(theme, dict): | |
| continue | |
| add(theme.get('title'), 'qualitative_theme', 'STRICT_CONSTRUCT') | |
| for concept in (theme.get('concepts') or []): | |
| add(concept, 'qualitative_concept', 'STRICT_CONSTRUCT') | |
| for row in (meta.get('evidence') or []): | |
| if isinstance(row, dict): | |
| 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', 'FLEXIBLE_TERM') | |
| # User keywords remain planning seeds, not canonical identities. Use | |
| # them only as a fallback when frozen empirical/qualitative metadata and | |
| # outline concepts provide no concept labels at all. | |
| if not items: | |
| 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', 'FLEXIBLE_TERM') | |
| # Capture explicit lower-case hyphenated academic expressions only when | |
| # they recur or carry a research-concept cue. This closes the | |
| # context-awareness/context-aware blind spot without freezing generic | |
| # methodological adjectives such as cross-sectional by default. | |
| hyphenated = re.findall(r'(?<![A-Za-z])(?:[a-z][a-z]{2,}(?:-[a-z][a-z]{2,})+)(?![A-Za-z])', str(research_logic_contract or '')) | |
| concept_cues = { | |
| 'awareness','trust','governance','literacy','fidelity','performance','context','capability', | |
| 'calibration','memory','acceptance','effectiveness','efficiency','alignment','recommendation', | |
| } | |
| counts = {} | |
| for term in hyphenated: | |
| counts[term.casefold()] = counts.get(term.casefold(), 0) + 1 | |
| 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', 'FLEXIBLE_TERM') | |
| return {'version': 'concept-ledger-v2', 'items': items} | |
| def _canonical_concept_writer_directive(ledger: dict) -> str: | |
| terms = [str(item.get('source') or '').strip() for item in (dict(ledger or {}).get('items') or []) if isinstance(item, dict)] | |
| terms = [term for term in terms if term] | |
| if not terms: | |
| return '' | |
| rendered = '; '.join(terms[:40]) | |
| return ( | |
| 'CANONICAL CONCEPT IDENTITY: The following research labels are frozen task-level concepts: ' | |
| + rendered | |
| + '. When referring to the same construct or concept, preserve its canonical English label; ' | |
| 'do not silently replace it with a near-synonym that changes concept identity. Ordinary grammatical ' | |
| 'articles, possessives, and surrounding syntax remain free. Distinct listed labels are not synonyms unless the evidence explicitly says so.' | |
| ) | |
| def _translation_ledger_hash(agent, source: str) -> str: | |
| method = getattr(agent, 'translation_fact_ledger_hash', None) | |
| if callable(method): | |
| return str(method(source) or '') | |
| # Test doubles / legacy adapters still get deterministic invalidation. | |
| return hashlib.sha256(str(source or '').encode('utf-8')).hexdigest() | |
| def _translation_identity_bundle(self, agent, source: str) -> dict[str, str]: | |
| """Return one canonical identity domain for translation reuse. | |
| The frozen English source remains the manuscript authority, while the | |
| Translation Compiler may consume a canonicalized representation (for | |
| example qualitative quote literals become QUOTE semantic slots). Cache | |
| identity must therefore persist both hashes and derive the private Fact | |
| Ledger only from the exact compiled source sent to the translator. | |
| """ | |
| raw_source = str(source or '').strip() | |
| compiled_source, _pending, _reused = self._prepare_qualitative_translation_source( | |
| raw_source, getattr(self, '_qualitative_quote_sources', {}), {} | |
| ) | |
| compiled_source = str(compiled_source or '').strip() | |
| return { | |
| 'source_hash': self._text_hash(raw_source), | |
| 'compiled_source': compiled_source, | |
| 'compiled_source_hash': self._text_hash(compiled_source), | |
| 'fact_ledger_hash': self._translation_ledger_hash(agent, compiled_source), | |
| } | |
| def _assemble_partial_translation(translation_sections, adaptive_structure, *, include_front=False) -> str: | |
| mapping = translation_sections if isinstance(translation_sections, dict) else {} | |
| parts = [] | |
| if include_front: | |
| record = mapping.get('__front_matter__') or {} | |
| if record.get('qa_status') in {'passed', 'passed_with_warning'} and record.get('translated_text'): | |
| parts.append(str(record['translated_text']).strip()) | |
| for key, _title in GenerationRunner._unique_top_level_sections(adaptive_structure): | |
| record = mapping.get(key) or {} | |
| if record.get('qa_status') in {'passed', 'passed_with_warning'} and record.get('translated_text'): | |
| parts.append(str(record['translated_text']).strip()) | |
| return '\n\n'.join(part for part in parts if part).strip() | |
| def _start_translation_worker( | |
| self, agent, *, translation_sections, translation_glossary, | |
| adaptive_structure, target_lang, | |
| ) -> None: | |
| if self._translation_worker is not None: | |
| return | |
| self.translation_agent = agent | |
| self._translation_worker_sections = dict(translation_sections or {}) | |
| self._translation_worker_glossary = dict(translation_glossary or {}) | |
| self._translation_worker_context = { | |
| '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: | |
| raise RuntimeError('translation_worker_agent_missing') | |
| context = dict(self._translation_worker_context or {}) | |
| sections, glossary = self._translate_frozen_unit( | |
| self.translation_agent, | |
| unit_key=str(job.get('unit_key') or ''), | |
| unit_name=str(job.get('unit_name') or ''), | |
| source_text=str(job.get('source_text') or ''), | |
| target_lang=str(context.get('target_lang') or ''), | |
| translation_sections=self._translation_worker_sections, | |
| translation_glossary=self._translation_worker_glossary, | |
| adaptive_structure=context.get('adaptive_structure') or [], | |
| include_front=True, | |
| defer_failures=True, | |
| english_commit_text=str(job.get('english_commit_text') or ''), | |
| ) | |
| self._translation_worker_sections = dict(sections or {}) | |
| self._translation_worker_glossary = dict(glossary or {}) | |
| return { | |
| 'unit_key': str(job.get('unit_key') or ''), | |
| 'source_hash': str(job.get('source_hash') or ''), | |
| 'translation_sections': dict(self._translation_worker_sections), | |
| 'translation_glossary': dict(self._translation_worker_glossary), | |
| } | |
| def _enqueue_translation_unit( | |
| self, agent, *, unit_key: str, unit_name: str, source_text: str, | |
| translation_sections: dict, english_commit_text: str, | |
| ) -> bool: | |
| source = str(source_text or '').strip() | |
| if not source or self._translation_worker is None: | |
| return False | |
| identity = self._translation_identity_bundle(agent, source) | |
| source_hash = identity['source_hash'] | |
| compiled_source_hash = identity['compiled_source_hash'] | |
| fact_ledger_hash = identity['fact_ledger_hash'] | |
| entity_ledger_method = getattr(agent, 'translation_protected_entity_ledger_hash', None) | |
| protected_entity_ledger_hash = str( | |
| entity_ledger_method(identity['compiled_source'], self._translation_worker_glossary) | |
| if callable(entity_ledger_method) else '' | |
| ) | |
| protected_entity_ledger_version = str( | |
| getattr(agent, 'PROTECTED_ENTITY_LEDGER_VERSION', 'protected-entity-ledger-v1') | |
| or 'protected-entity-ledger-v1' | |
| ) | |
| 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 | |
| and existing.get('compiled_source_hash') == compiled_source_hash | |
| and existing.get('target_lang') == str(self._translation_worker_context.get('target_lang') or '') | |
| and existing.get('translation_contract_version') == TRANSLATION_CONTRACT_VERSION | |
| and existing.get('fact_ledger_hash') == fact_ledger_hash | |
| and existing.get('protected_entity_ledger_version') == protected_entity_ledger_version | |
| and existing.get('protected_entity_ledger_hash') == protected_entity_ledger_hash | |
| and existing.get('translated_text') | |
| ): | |
| return False | |
| if existing and existing.get('translated_text'): | |
| reasons = [] | |
| if existing.get('source_hash') != source_hash: reasons.append('english_source_hash') | |
| if existing.get('compiled_source_hash') != compiled_source_hash: reasons.append('compiled_source_hash') | |
| if existing.get('fact_ledger_hash') != fact_ledger_hash: reasons.append('fact_ledger_hash') | |
| if existing.get('protected_entity_ledger_version') != protected_entity_ledger_version: reasons.append('protected_entity_ledger_version') | |
| if existing.get('protected_entity_ledger_hash') != protected_entity_ledger_hash: reasons.append('protected_entity_ledger_hash') | |
| if existing.get('translation_contract_version') != TRANSLATION_CONTRACT_VERSION: reasons.append('contract_version') | |
| if existing.get('target_lang') != str(self._translation_worker_context.get('target_lang') or ''): reasons.append('target_lang') | |
| if existing.get('qa_status') not in {'passed', 'passed_with_warning'}: reasons.append('qa_status') | |
| logging.info('Translation reuse miss | unit=%s phase=enqueue reasons=%s', unit_name, ','.join(reasons) or 'unknown') | |
| pending_key = (str(unit_key), source_hash, compiled_source_hash, fact_ledger_hash, protected_entity_ledger_version, protected_entity_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), | |
| 'source_text': source, | |
| 'source_hash': source_hash, | |
| 'compiled_source_hash': compiled_source_hash, | |
| 'fact_ledger_hash': fact_ledger_hash, | |
| 'english_commit_text': str(english_commit_text or ''), | |
| }) | |
| self._translation_worker_pending.add(pending_key) | |
| self._translation_worker_jobs.append({ | |
| '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): | |
| sections = dict(translation_sections or {}) | |
| glossary = dict(translation_glossary or {}) | |
| while self._translation_worker_collected < len(self._translation_worker_jobs): | |
| item = self._translation_worker_jobs[self._translation_worker_collected] | |
| future = item['future'] | |
| if not block and not future.done(): | |
| break | |
| 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 | |
| def _shutdown_translation_worker(self, *, wait=False) -> None: | |
| worker = self._translation_worker | |
| if worker is None: | |
| return | |
| worker.close(wait=bool(wait)) | |
| self._translation_worker = None | |
| def _translate_frozen_unit( | |
| self, agent, *, unit_key: str, unit_name: str, source_text: str, target_lang: str, | |
| translation_sections: dict, translation_glossary: dict, adaptive_structure, include_front=False, | |
| defer_failures=True, english_commit_text: str | None = None, final_gate: bool = False, | |
| ): | |
| source = str(source_text or '').strip() | |
| if not source: | |
| return translation_sections, translation_glossary | |
| identity = self._translation_identity_bundle(agent, source) | |
| source_hash = identity['source_hash'] | |
| compiled_source_hash = identity['compiled_source_hash'] | |
| canonical_translation_source = identity['compiled_source'] | |
| contract_version = TRANSLATION_CONTRACT_VERSION | |
| fact_ledger_hash = identity['fact_ledger_hash'] | |
| entity_ledger_method = getattr(agent, 'translation_protected_entity_ledger_hash', None) | |
| protected_entity_ledger_hash = str( | |
| entity_ledger_method(canonical_translation_source, translation_glossary) | |
| if callable(entity_ledger_method) else '' | |
| ) | |
| protected_entity_ledger_version = str( | |
| getattr(agent, 'PROTECTED_ENTITY_LEDGER_VERSION', 'protected-entity-ledger-v1') or 'protected-entity-ledger-v1' | |
| ) | |
| 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 | |
| and existing.get('compiled_source_hash') == compiled_source_hash | |
| and existing.get('target_lang') == target_lang | |
| and existing.get('translation_contract_version') == contract_version | |
| and existing.get('fact_ledger_hash') == fact_ledger_hash | |
| and existing.get('protected_entity_ledger_version') == protected_entity_ledger_version | |
| and existing.get('protected_entity_ledger_hash') == protected_entity_ledger_hash | |
| and existing.get('translated_text') | |
| ): | |
| self._translation_progress_inflight.discard(str(unit_key)) | |
| return translation_sections, translation_glossary | |
| if existing and existing.get('translated_text'): | |
| reasons = [] | |
| if existing.get('source_hash') != source_hash: reasons.append('english_source_hash') | |
| if existing.get('compiled_source_hash') != compiled_source_hash: reasons.append('compiled_source_hash') | |
| if existing.get('fact_ledger_hash') != fact_ledger_hash: reasons.append('fact_ledger_hash') | |
| if existing.get('protected_entity_ledger_version') != protected_entity_ledger_version: reasons.append('protected_entity_ledger_version') | |
| if existing.get('protected_entity_ledger_hash') != protected_entity_ledger_hash: reasons.append('protected_entity_ledger_hash') | |
| if existing.get('translation_contract_version') != contract_version: reasons.append('contract_version') | |
| if existing.get('target_lang') != target_lang: reasons.append('target_lang') | |
| if existing.get('qa_status') not in {'passed', 'passed_with_warning'}: reasons.append('qa_status') | |
| logging.info('Translation reuse miss | unit=%s phase=final_or_direct reasons=%s', unit_name, ','.join(reasons) or 'unknown') | |
| self._translation_progress_inflight.add(str(unit_key)) | |
| partial = self._assemble_partial_translation(translation_sections, adaptive_structure, include_front=include_front) | |
| initial_target_stage = f'translation_final_gate:{unit_name}' if final_gate else f'translating:{unit_name}' | |
| self._publish_target_preview( | |
| partial, initial_target_stage, force=True, complete=False, | |
| english_commit_text=english_commit_text, | |
| ) | |
| updated_sections = dict(translation_sections or {}) | |
| raw_quote_ledger_record = updated_sections.get("__quote_ledger__") | |
| if raw_quote_ledger_record is not None and not isinstance(raw_quote_ledger_record, dict): | |
| logging.warning( | |
| "Qualitative quote translation ledger ignored malformed record type=%s", | |
| type(raw_quote_ledger_record).__name__, | |
| ) | |
| quote_ledger_record = ( | |
| dict(raw_quote_ledger_record) if isinstance(raw_quote_ledger_record, dict) else {} | |
| ) | |
| current_quote_hashes = { | |
| tag: hashlib.sha256(re.sub(r"\s+", " ", quote).strip().encode("utf-8")).hexdigest() | |
| for tag, quote in getattr(self, "_qualitative_quote_sources", {}).items() | |
| } | |
| current_request_hash = str(getattr(self, "request_hash", "") or "") | |
| current_task_id = str(getattr(getattr(self, "task_dir", None), "name", "") or "") | |
| ledger_identity_ok = self._qualitative_quote_ledger_identity_ok( | |
| quote_ledger_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_translations = quote_ledger_record.get("translations") | |
| quote_ledger = ( | |
| dict(stored_translations) | |
| if ledger_identity_ok and isinstance(stored_translations, dict) | |
| else {} | |
| ) | |
| translation_source, pending_quote_tags, reused_quote_tags = self._prepare_qualitative_translation_source( | |
| source, getattr(self, "_qualitative_quote_sources", {}), quote_ledger | |
| ) | |
| translation_source = str(translation_source or '').strip() | |
| if translation_source != canonical_translation_source: | |
| raise RuntimeError('Canonical translation source changed with Quote Ledger state.') | |
| def publish_translation_stage(stage): | |
| value = str(stage or '').strip().casefold() | |
| if value == 'qa': | |
| preview_stage = f'translation_qa:{unit_name}' | |
| elif value == 'repairing': | |
| preview_stage = f'translation_repairing:{unit_name}' | |
| elif value == 'adjudicating': | |
| preview_stage = f'translation_adjudicating:{unit_name}' | |
| elif final_gate: | |
| preview_stage = f'translation_final_gate:{unit_name}' | |
| else: | |
| preview_stage = f'translating:{unit_name}' | |
| self._publish_target_preview( | |
| partial, preview_stage, force=True, complete=False, | |
| english_commit_text=english_commit_text, | |
| ) | |
| 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, | |
| } | |
| try: | |
| 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: | |
| translated_text, quote_ledger = self._finalize_qualitative_translation_quotes( | |
| translated_text, pending_quote_tags, reused_quote_tags, quote_ledger | |
| ) | |
| agent.validate_translation_alignment(source, translated_text) | |
| result = dict(result) | |
| result['text'] = translated_text | |
| 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(), | |
| } | |
| except PipelineCancelledError: | |
| raise | |
| except Exception as error: | |
| # Persist repair state for both intermediate and strict final gates. | |
| # The caller owns checkpoint authority and decides whether to stop. | |
| code_method = getattr(agent, 'translation_error_code', None) | |
| error_code = str(code_method(error) if callable(code_method) else 'TRANSLATION_PENDING') | |
| updated_sections[unit_key] = { | |
| 'source_hash': source_hash, | |
| 'compiled_source_hash': compiled_source_hash, | |
| 'target_lang': target_lang, | |
| 'glossary_version': self._glossary_version(translation_glossary), | |
| 'translation_contract_version': contract_version, | |
| 'fact_ledger_hash': fact_ledger_hash, | |
| 'protected_entity_ledger_version': protected_entity_ledger_version, | |
| 'protected_entity_ledger_hash': protected_entity_ledger_hash, | |
| 'translated_text': '', | |
| 'qa_status': 'repair_pending', | |
| 'claim_states': dict(getattr(agent, '_last_translation_claim_states', {}) or {}), | |
| 'claim_state_version': 'translation-claim-state-v2', | |
| 'error_code': error_code, | |
| 'attempts': int(existing.get('attempts') or 0) + 1, | |
| 'deferred_at': datetime.now(timezone.utc).isoformat(), | |
| } | |
| logging.warning( | |
| 'Translation repair pending | unit=%s final_gate=%s code=%s error=%s', | |
| unit_name, bool(final_gate), 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, | |
| ) | |
| return updated_sections, translation_glossary | |
| updated_glossary = dict(result.get('glossary') or translation_glossary or {}) | |
| result_fact_ledger_hash = str(result.get('fact_ledger_hash') or fact_ledger_hash) | |
| if result_fact_ledger_hash != fact_ledger_hash: | |
| raise RuntimeError('Translation result Fact Ledger identity differs from canonical compiled source.') | |
| updated_sections[unit_key] = { | |
| 'source_hash': source_hash, | |
| 'compiled_source_hash': compiled_source_hash, | |
| 'target_lang': target_lang, | |
| 'glossary_version': self._glossary_version(updated_glossary), | |
| 'translation_contract_version': str(result.get('translation_contract_version') or contract_version), | |
| 'fact_ledger_hash': result_fact_ledger_hash, | |
| 'protected_entity_ledger_version': str(result.get('protected_entity_ledger_version') or protected_entity_ledger_version), | |
| 'protected_entity_ledger_hash': str(result.get('protected_entity_ledger_hash') or protected_entity_ledger_hash), | |
| 'translated_text': str(result.get('text') or '').strip(), | |
| '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-v2'), | |
| 'attempts': int(result.get('attempts') or 1), | |
| '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, | |
| ) | |
| return updated_sections, updated_glossary | |
| def _bounded_target(base_target: int, accumulated_deviation: float, remaining_including_current: int, safe_chunk: int) -> int: | |
| base = max(80, int(base_target)) | |
| correction = -float(accumulated_deviation) / max(1, int(remaining_including_current)) | |
| lower = max(80, int(round(base * 0.75))) | |
| upper = min(int(safe_chunk * 1.5), max(lower, int(round(base * 1.25)))) | |
| return max(lower, min(upper, int(round(base + correction)))) | |
| def _english_count_for_target(manuscript: str) -> int: | |
| body = re.split(r"(?m)^#\s+References\s*$", str(manuscript or ""), maxsplit=1)[0] | |
| body = re.split(r"(?m)^#\s+Appendix: Data Structure and Evidence\s*$", body, maxsplit=1)[0] | |
| # Markdown punctuation is ignored; contractions/hyphenated terms stay one token. | |
| return len(re.findall(r"\b[A-Za-z0-9]+(?:[-'][A-Za-z0-9]+)*\b", body)) | |
| def _validate_target_length(self, manuscript: str, target_words: int) -> dict[str, Any]: | |
| """Return a nonfatal total-length quality diagnostic. | |
| The user-selected target remains a planning input, not a completion gate. | |
| Completeness is validated separately by structure, evidence, citation, | |
| translation, DOCX, and ZIP checks. | |
| """ | |
| quality = APP_CONFIG.get("manuscript_validation", {}) | |
| minimum = int(target_words * float(quality.get("minimum_total_ratio", 0.97))) | |
| maximum = int(target_words * float(quality.get("maximum_total_ratio", 1.03))) | |
| short_notice = float(quality.get("total_short_notice_ratio", 0.70)) | |
| topic_review = float(quality.get("total_topic_review_ratio", 0.50)) | |
| actual = self._english_count_for_target(manuscript) | |
| ratio = (actual / max(1, int(target_words))) | |
| status = "normal" | |
| recommendation = "" | |
| if ratio < topic_review: | |
| status = "very_short" | |
| recommendation = ( | |
| "First verify that generation completed normally and no section/evidence is missing. " | |
| "If the manuscript is complete, consider broadening the research scope or selecting a new research topic." | |
| ) | |
| elif ratio < short_notice: | |
| status = "short" | |
| recommendation = ( | |
| "If the manuscript is structurally complete, consider broadening the theoretical scope, context, " | |
| "research question, or research topic." | |
| ) | |
| elif actual < minimum: | |
| status = "slightly_short" | |
| elif actual > maximum: | |
| status = "long" | |
| if status != "normal": | |
| logging.warning( | |
| "Manuscript length diagnostic | actual=%s | target=%s | ratio=%.3f | status=%s | nonfatal=true", | |
| actual, target_words, ratio, status, | |
| ) | |
| return { | |
| "scope": "manuscript", "name": "English manuscript", "actual_words": actual, | |
| "target_words": int(target_words), "ratio": round(ratio, 4), "status": status, | |
| "recommendation": recommendation, "nonfatal": True, | |
| } | |
| def _upsert_quality_diagnostic(items: list[dict[str, Any]], diagnostic: dict[str, Any]) -> None: | |
| key = (str(diagnostic.get("scope") or ""), str(diagnostic.get("name") or "")) | |
| items[:] = [ | |
| item for item in items | |
| if (str(item.get("scope") or ""), str(item.get("name") or "")) != key | |
| ] | |
| items.append(dict(diagnostic)) | |
| def _section_word_diagnostic(self, name: str, actual_words: int, target_words: int) -> dict[str, Any]: | |
| validation = APP_CONFIG.get("manuscript_validation", {}) | |
| budget = APP_CONFIG.get("word_budget", {}) | |
| target = max(1, int(target_words)) | |
| actual = max(0, int(actual_words)) | |
| ratio = actual / target | |
| short_notice = float(validation.get("section_short_notice_ratio", 0.60)) | |
| topic_review = float(validation.get("section_topic_review_ratio", 0.40)) | |
| preferred_min = float(budget.get("section_min_ratio", 0.92)) | |
| preferred_max = float(budget.get("section_max_ratio", 1.08)) | |
| status = "normal" | |
| recommendation = "" | |
| if ratio < topic_review: | |
| status = "very_short" | |
| recommendation = ( | |
| "First verify that this section completed normally and that evidence was available. " | |
| "If it is complete, consider broadening the research scope or selecting a new research topic." | |
| ) | |
| elif ratio < short_notice: | |
| status = "short" | |
| recommendation = ( | |
| "If the section is complete, consider broadening the theoretical scope, context, research question, " | |
| "or research topic." | |
| ) | |
| elif ratio < preferred_min: | |
| status = "slightly_short" | |
| elif ratio > preferred_max: | |
| status = "long" | |
| if status != "normal": | |
| logging.warning( | |
| "Section length diagnostic | section=%s | actual=%s | target=%s | ratio=%.3f | status=%s | nonfatal=true", | |
| name, actual, target, ratio, status, | |
| ) | |
| return { | |
| "scope": "section", "name": str(name), "actual_words": actual, "target_words": target, | |
| "ratio": round(ratio, 4), "status": status, "recommendation": recommendation, "nonfatal": True, | |
| } | |
| def _citation_eligible_section_ids(mode: str) -> set[str]: | |
| return ( | |
| {"introduction", "theoretical_framing", "methods", "discussion"} | |
| if str(mode).lower() == "asq" | |
| else {"introduction", "theory", "methodology", "discussion"} | |
| ) | |
| def _adaptive_citation_requirements(cls, tasks, mode: str): | |
| """Derive evidence demand and semantic-survival reserve from manuscript units.""" | |
| allowed = cls._citation_eligible_section_ids(mode) | |
| minimums = {} | |
| reserve_targets = {} | |
| for task in tasks or []: | |
| section_id = str(task.get("section_id") or "") | |
| display_name = str(task.get("display_name") or "").strip() | |
| if section_id not in allowed or not display_name: | |
| continue | |
| base, words_per_source = CITATION_ROLE_RULES.get(section_id, (2, 240)) | |
| target_words = max(1, int(task.get("target_words") or 1)) | |
| minimum = max(base, int(math.ceil(target_words / max(1, words_per_source)))) | |
| reserve = max( | |
| minimum, | |
| int(math.ceil(minimum / max(0.01, CITATION_SEMANTIC_SURVIVAL_RATIO))), | |
| ) | |
| minimums[display_name] = minimum | |
| reserve_targets[display_name] = reserve | |
| return minimums, reserve_targets | |
| def _top_level_citation_requirements(cls, tasks, mode: str): | |
| """Aggregate citation authority by academic section, never by technical chunk. | |
| ``Introduction (Part 1/2)`` and ``Introduction (Part 2/2)`` are generation | |
| chunks, not separate scholarly evidence authorities. Planning may remain | |
| unit-local, but fatal literature sufficiency is evaluated on the canonical | |
| top-level academic section. | |
| """ | |
| allowed = cls._citation_eligible_section_ids(mode) | |
| words_by_section = {} | |
| for task in tasks or []: | |
| section_id = str(task.get('section_id') or '') | |
| if section_id not in allowed: | |
| continue | |
| words_by_section[section_id] = words_by_section.get(section_id, 0) + max( | |
| 1, int(task.get('target_words') or 1) | |
| ) | |
| minimums = {} | |
| reserve_targets = {} | |
| for section_id, target_words in words_by_section.items(): | |
| base, words_per_source = CITATION_ROLE_RULES.get(section_id, (2, 240)) | |
| minimum = max(base, int(math.ceil(target_words / max(1, words_per_source)))) | |
| reserve = max( | |
| minimum, | |
| int(math.ceil(minimum / max(0.01, CITATION_SEMANTIC_SURVIVAL_RATIO))), | |
| ) | |
| minimums[section_id] = minimum | |
| reserve_targets[section_id] = reserve | |
| return minimums, reserve_targets | |
| def _citation_unit_section_map(tasks): | |
| return { | |
| str(task.get('display_name') or ''): str(task.get('section_id') or '') | |
| for task in (tasks or []) if str(task.get('display_name') or '').strip() | |
| } | |
| def _top_level_unique_tag_counts(citation_plan, unit_section_map): | |
| tags_by_section = {} | |
| for unit_name, items in dict(citation_plan or {}).items(): | |
| section_id = str(unit_section_map.get(str(unit_name)) or '') | |
| if not section_id: | |
| continue | |
| bucket = tags_by_section.setdefault(section_id, set()) | |
| for item in items or []: | |
| tag = str((item or {}).get('tag') or '') | |
| if tag: | |
| bucket.add(tag) | |
| return {section_id: len(tags) for section_id, tags in tags_by_section.items()} | |
| def _adaptive_literature_candidate_target(target_words: int, agent) -> int: | |
| """Adaptive retrieval aspiration; never a final reference-count quota.""" | |
| dynamic = max(24, int(math.ceil(max(1, int(target_words or 1)) / 90.0))) | |
| configured_cap = max(1, int(getattr(agent, "lit_target_papers", dynamic) or dynamic)) | |
| return min(configured_cap, dynamic) | |
| def _citation_task_context(task, research_logic_contract="", synthesis_result="") -> str: | |
| parts = [ | |
| str(task.get("display_name") or ""), | |
| str(task.get("required_content") or ""), | |
| " ".join(str(value) for value in (task.get("concepts") or []) if str(value).strip()), | |
| str(research_logic_contract or "")[:1600], | |
| str(synthesis_result or "")[:1600], | |
| ] | |
| return re.sub(r"\s+", " ", " ".join(part for part in parts if part)).strip() | |
| def _literature_insufficient_error(shortfalls) -> RuntimeError: | |
| details = "; ".join( | |
| f"{name}={actual}/{required}" | |
| for name, actual, required in (shortfalls or [])[:8] | |
| ) | |
| return RuntimeError( | |
| "E_LITERATURE_INSUFFICIENT: high-relevance claim-specific literature is insufficient " | |
| f"for the planned manuscript units ({details}). Please revise the research topic or " | |
| "keywords and try again; unrelated references will not be used to satisfy a quota." | |
| ) | |
| def _evidence_warning_detail() -> str: | |
| return ( | |
| "Some externally oriented claims have limited direct high-relevance literature coverage. " | |
| "This may reflect a newer or specialized topic, dispersed literature, search/keyword limits, " | |
| "or limited prior evidence; this status does not determine the novelty, importance, or overall " | |
| "research value of the topic. Unrelated sources were not used to satisfy a citation count." | |
| ) | |
| def _record_evidence_coverage_event(self, state, shortfalls, *, phase: str, preview_status: str) -> None: | |
| """Persist a neutral evidence-coverage signal without stopping generation. | |
| The event records only deterministic coverage counts. It never infers why | |
| literature is sparse and never evaluates topic value. Duplicate events | |
| are collapsed across resume/replay. | |
| """ | |
| rows = [] | |
| for name, actual, required in shortfalls or []: | |
| rows.append({ | |
| "section": str(name or ""), | |
| "actual": max(0, int(actual or 0)), | |
| "required": max(0, int(required or 0)), | |
| }) | |
| if not rows: | |
| return | |
| signature = (str(phase or ""), tuple((r["section"], r["actual"], r["required"]) for r in rows)) | |
| existing = list(state.get("evidence_coverage_events") or self._evidence_coverage_events or []) | |
| seen = { | |
| (str(item.get("phase") or ""), tuple( | |
| (str(row.get("section") or ""), int(row.get("actual") or 0), int(row.get("required") or 0)) | |
| for row in (item.get("shortfalls") or []) if isinstance(row, dict) | |
| )) | |
| for item in existing if isinstance(item, dict) | |
| } | |
| if signature not in seen: | |
| existing.append({ | |
| "phase": str(phase or ""), | |
| "shortfalls": rows, | |
| "observed_at": datetime.now(timezone.utc).isoformat(), | |
| }) | |
| self._evidence_coverage_events = existing | |
| state["evidence_coverage_events"] = existing | |
| state["evidence_risk_detected"] = True | |
| self._preview_evidence_status = str(preview_status or "limited_risk") | |
| self._write_preview_payload(force=True) | |
| logging.warning( | |
| "Evidence coverage warning | phase=%s | shortfalls=%s | nonfatal=true", | |
| phase, rows, | |
| ) | |
| def _evidence_coverage_summary(*, status, early_risk, final_reference_count, minimums, validated_by_top_level, claims_by_top_level, events, method_authority_count=0, method_external_claims=0): | |
| sections = [] | |
| for section_id, planned_minimum in dict(minimums or {}).items(): | |
| validated = len(set((validated_by_top_level or {}).get(section_id) or [])) | |
| external_claims = max(0, int((claims_by_top_level or {}).get(section_id, 0) or 0)) | |
| required = min(int(planned_minimum), external_claims) if external_claims else 0 | |
| sections.append({ | |
| "section": str(section_id), | |
| "validated_direct_citations": validated, | |
| "evidence_requirement": required, | |
| "external_claims": external_claims, | |
| "planning_minimum": int(planned_minimum), | |
| }) | |
| return { | |
| "status": str(status or "adequate"), | |
| "early_risk_detected": bool(early_risk), | |
| "final_reference_count": max(0, int(final_reference_count or 0)), | |
| "sections": sections, | |
| "methodology_authority": { | |
| "validated_method_authority_citations": max(0, int(method_authority_count or 0)), | |
| "methodology_external_claims": max(0, int(method_external_claims or 0)), | |
| }, | |
| "events": list(events or []), | |
| } | |
| def _paragraph_token_set(value): | |
| text = re.sub(r'\[CITE-\d+\]', ' ', str(value or ''), flags=re.I) | |
| return set(re.findall(r'[A-Za-z][A-Za-z0-9_-]{2,}', text.lower())) | |
| def _remove_severe_cross_unit_duplicates(cls, candidate, prior_text): | |
| """Drop only very high-confidence cross-unit duplicate paragraphs.""" | |
| previous = [p.strip() for p in re.split(r'\n\s*\n', str(prior_text or '')) if len(p.split()) >= 55 and not p.lstrip().startswith('#')] | |
| kept, diagnostics = [], [] | |
| for paragraph in re.split(r'(\n\s*\n)', str(candidate or '')): | |
| if re.fullmatch(r'\n\s*\n', paragraph or ''): | |
| kept.append(paragraph); continue | |
| value = str(paragraph or '').strip() | |
| if len(value.split()) < 55 or value.startswith('#'): | |
| kept.append(paragraph); continue | |
| tokens = cls._paragraph_token_set(value) | |
| citations = set(re.findall(r'\[CITE-\d+\]', value, flags=re.I)) | |
| numerics = set(re.findall(r'(?<![A-Za-z])[-+]?\d+(?:\.\d+)?%?(?![A-Za-z])', value)) | |
| duplicate = None | |
| for old in previous: | |
| old_tokens = cls._paragraph_token_set(old) | |
| union = tokens | old_tokens | |
| similarity = len(tokens & old_tokens) / max(1, len(union)) | |
| if similarity < 0.90: | |
| continue | |
| old_citations = set(re.findall(r'\[CITE-\d+\]', old, flags=re.I)) | |
| old_numerics = set(re.findall(r'(?<![A-Za-z])[-+]?\d+(?:\.\d+)?%?(?![A-Za-z])', old)) | |
| # Never delete a paragraph carrying different evidence/facts. | |
| if citations == old_citations and numerics == old_numerics: | |
| duplicate = (similarity, old[:180]) | |
| break | |
| if duplicate: | |
| diagnostics.append({"similarity": round(duplicate[0], 4), "removed_preview": value[:180]}) | |
| continue | |
| kept.append(paragraph) | |
| previous.append(value) | |
| return ''.join(kept).strip(), diagnostics | |
| def _resolve_paths(self, items) -> list[str]: | |
| output = [] | |
| for item in items or []: | |
| path = (self.task_dir / str(item)).resolve() | |
| if self.task_dir.resolve() not in path.parents: | |
| raise RuntimeError("input_path_escape") | |
| if path.is_file(): | |
| output.append(str(path)) | |
| return output | |
| def _run_impl(self) -> dict[str, Any]: | |
| req = self.request | |
| mode = str(req.get("mode") or "").lower() | |
| if mode not in {"amj", "asq"}: | |
| raise RuntimeError("invalid_generation_mode") | |
| target_words = int(req.get("target_words") or 0) | |
| if not APP_CONFIG["target_words_min"] <= target_words <= APP_CONFIG["target_words_max"]: | |
| raise RuntimeError("target_words_out_of_range") | |
| ds_key = str(os.environ.get("DEEPSEEK_API_KEY") or "").strip() | |
| s2_key = str(os.environ.get("S2_API_KEY") or "").strip() | |
| if not ds_key: | |
| raise RuntimeError("DEEPSEEK_API_KEY_missing") | |
| if mode == "amj": | |
| APP_CONFIG.update(AMJ_PROMPTS) | |
| else: | |
| APP_CONFIG.update(ASQ_PROMPTS) | |
| agent = SSciPipelineAgent(ds_key, s2_key) | |
| self.agent = agent | |
| agent.set_cancel_checker(self.cancel_checker) | |
| # DeepSeek/search activity marks the executor as genuinely alive; the | |
| # heartbeat thread decides separately when to renew D1's lease. | |
| agent.set_execution_progress_hook(lambda: self._progress_activity()) | |
| safe_chunk_words = int(agent.chunk_words_default) | |
| quant_engine = get_quantitative_engine() | |
| qual_engine = get_qualitative_engine(agent) | |
| if self.confirmed_checkpoint_seq is None: | |
| # Compatibility path for isolated unit helpers that do not represent | |
| # a claimed production executor. Production always supplies Worker | |
| # checkpoint metadata through task_executor. | |
| state = self.checkpoints.load_latest() or {} | |
| else: | |
| state = self.checkpoints.load_confirmed() or {} | |
| quality_diagnostics = list(state.get("quality_diagnostics") or []) | |
| self._evidence_coverage_events = list(state.get("evidence_coverage_events") or []) | |
| if str(state.get("evidence_status") or "") == "limited": | |
| self._preview_evidence_status = "limited" | |
| elif bool(state.get("evidence_risk_detected")): | |
| self._preview_evidence_status = "limited_risk" | |
| if state: | |
| self._preview_english_text = str(state.get('full_draft_en') or '') | |
| self._preview_target_text = str(state.get('full_draft_translated') or '') | |
| logging.info( | |
| "[SP_EXECUTOR] resume checkpoint seq=%s stage=%s", | |
| state.get("checkpoint_seq"), state.get("checkpoint_stage"), | |
| ) | |
| topic = str(req.get("topic") or "").strip() | |
| keywords = str(req.get("keywords") or "").strip() | |
| target_lang = str(req.get("target_lang") or "").strip() | |
| output_suffix = str(req.get("output_suffix") or "Target").strip() | |
| interface_language = str(req.get("interface_language") or "zh").strip() | |
| reference_paths = self._resolve_paths(req.get("reference_files") or []) | |
| data_paths = self._resolve_paths(req.get("data_files") or []) | |
| interview_paths = self._resolve_paths(req.get("interview_files") or []) | |
| base_evidence_contract = str(APP_CONFIG.get("evidence_integrity_contract") or "") | |
| # Restore agent state before continuing any dependent step. | |
| if isinstance(state.get("agent_state"), dict): | |
| agent.load_state(state["agent_state"]) | |
| if state.get("english_finalized") and state.get("final_citation_ledger_version") != FINAL_CITATION_LEDGER_VERSION: | |
| # R4-R3: an R4-R2 checkpoint may contain a correct resolved body but | |
| # an empty References tail caused by rescanning already-resolved CITE | |
| # tags. Re-enter only finalization/translation/package gates; completed | |
| # English units and the same task/reservation remain authoritative. | |
| logging.warning("Final citation ledger contract advanced; re-running finalization from frozen completed units.") | |
| state["english_finalized"] = False | |
| state["translation_complete"] = False | |
| state["package_validated"] = False | |
| if state.get("english_finalized") and state.get("english_section_seal_version") != ENGLISH_SECTION_SEAL_VERSION: | |
| # R2 compatibility: upgrade old completed English sections locally. | |
| # No writing-model call, new task, or new quota reservation is allowed. | |
| logging.warning("English Section Integrity Seal contract advanced; re-validating frozen completed sections locally.") | |
| state["english_finalized"] = False | |
| state["translation_complete"] = False | |
| state["package_validated"] = False | |
| user_lit_context = str(state.get("user_lit_context") or "") | |
| user_data_md = str(state.get("user_data_md") or "") | |
| data_metadata = dict(state.get("data_metadata") or {}) | |
| if not state.get("evidence_ready"): | |
| self._progress(3, "inputs_and_evidence") | |
| if reference_paths: | |
| user_lit_context = extract_text_from_files(reference_paths[:20]) | |
| all_paper_texts = [extract_text_from_files([path]) for path in reference_paths[:20]] | |
| agent.global_memory["core_golden_metadata"] = extract_golden_metadata(agent, all_paper_texts) | |
| if mode == "amj": | |
| data_file = data_paths[0] if data_paths else None | |
| _, user_data_md, _, data_metadata = quant_engine.resolve_empirical_data( | |
| data_file, interface_language, topic=topic, keywords=keywords, agent=agent | |
| ) | |
| else: | |
| raw_interviews = extract_text_from_files(interview_paths[:1]) if interview_paths else "" | |
| prewrite_started = time.monotonic() | |
| prewrite_calls = agent.ds_calls | |
| user_data_md, _, data_metadata = qual_engine.resolve_qualitative_data(topic, keywords, raw_interviews) | |
| agent.record_prewrite_stage("qualitative_packet", prewrite_started, prewrite_calls) | |
| state.update( | |
| { | |
| "topic": topic, | |
| "keywords": keywords, | |
| "target_lang": target_lang, | |
| "target_words": target_words, | |
| "mode": mode, | |
| "interface_language": interface_language, | |
| "user_lit_context": user_lit_context, | |
| "user_data_md": user_data_md, | |
| "data_metadata": data_metadata, | |
| "agent_state": agent.export_state(), | |
| "evidence_ready": True, | |
| } | |
| ) | |
| self._checkpoint("evidence_ready", state, 6) | |
| if mode == "amj": | |
| current_quant_ledger = dict(data_metadata.get("quantitative_evidence_ledger") or {}) | |
| expected_quant_version = str(getattr(quant_engine, "QUANTITATIVE_EVIDENCE_LEDGER_VERSION", "")) | |
| if current_quant_ledger.get("version") != expected_quant_version: | |
| data_metadata["quantitative_evidence_ledger"] = quant_engine.build_quantitative_evidence_ledger( | |
| data_metadata, user_data_md | |
| ) | |
| state["data_metadata"] = data_metadata | |
| logging.warning( | |
| "Quantitative Evidence Ledger contract advanced; rebuilt locally version=%s", | |
| expected_quant_version, | |
| ) | |
| self._checkpoint("quantitative_evidence_ledger_upgraded", state, 6) | |
| self._qualitative_quote_sources = ( | |
| self._qualitative_quote_source_map(data_metadata) if mode == "asq" else {} | |
| ) | |
| # ASQ planning never needs access to verbatim quotations. Keep the | |
| # research-logic/front-matter stage on a quote-free internal-evidence | |
| # summary; verbatim text is disclosed later only to its owning Findings | |
| # unit (or the appendix). | |
| planning_data_md = ( | |
| qual_engine.build_scoped_evidence_md(data_metadata, include_themes=True) | |
| if mode == "asq" | |
| else user_data_md | |
| ) | |
| refs_text = str(state.get("refs_text") or "") | |
| if not refs_text: | |
| self._progress(8, "S2 + OpenAlex") | |
| literature_candidate_target = self._adaptive_literature_candidate_target(target_words, agent) | |
| refs_text = agent.fetch_references_parallel( | |
| topic, | |
| keywords, | |
| min_required=literature_candidate_target, | |
| progress_callback=lambda: self._progress(9, "S2 + OpenAlex"), | |
| ) | |
| if not agent.citation_map: | |
| self._record_evidence_coverage_event( | |
| state, [("retrieved literature pool", 0, 1)], | |
| phase="retrieval_pool_empty", preview_status="limited_risk", | |
| ) | |
| logging.warning( | |
| "No directly validated literature was retrieved; continuing in evidence-limited mode without padding references." | |
| ) | |
| state["literature_candidate_target"] = literature_candidate_target | |
| if user_lit_context: | |
| core_refs = agent.parse_user_core_literature(user_lit_context) | |
| if core_refs: | |
| refs_text = ( | |
| "【USER PROVIDED CORE LITERATURE】\n" + core_refs | |
| + "\n\n【RETRIEVED LITERATURE】\n" + refs_text | |
| ) | |
| state.update({"refs_text": refs_text, "agent_state": agent.export_state()}) | |
| self._checkpoint("literature_ready", state, 11) | |
| rag_engine = initialize_rag([value for value in (user_lit_context, refs_text) if value]) | |
| synthesis_result = str(state.get("synthesis_result") or "") | |
| if not synthesis_result: | |
| self._progress(12, "literature_synthesis") | |
| prewrite_started = time.monotonic() | |
| prewrite_calls = agent.ds_calls | |
| if not str(refs_text or "").strip() and not str(user_lit_context or "").strip(): | |
| synthesis_result = ( | |
| "EVIDENCE-LIMITED CONTEXT: no directly validated external literature is currently available. " | |
| "Continue with the frozen study evidence, cautious conceptual reasoning, and general scholarly knowledge, " | |
| "but do not invent citations or bibliographic metadata and do not present general model knowledge as retrieved evidence." | |
| ) | |
| else: | |
| synthesis_result = self._stream_text( | |
| agent.stream_synthesize_literature(topic, refs_text, user_core_literature=user_lit_context) | |
| ) | |
| agent.record_prewrite_stage("literature_synthesis", prewrite_started, prewrite_calls) | |
| synthesis_result = re.sub(r"<think>.*?</think>", "", synthesis_result, flags=re.DOTALL).strip() | |
| state.update({"synthesis_result": synthesis_result, "agent_state": agent.export_state()}) | |
| self._checkpoint("synthesis_ready", state, 13) | |
| adaptive_structure = state.get("adaptive_structure") | |
| if not isinstance(adaptive_structure, list) or not adaptive_structure: | |
| self._progress(14, "research_logic") | |
| prewrite_started = time.monotonic() | |
| prewrite_calls = agent.ds_calls | |
| research_logic_response = self._stream_text( | |
| agent.stream_generate_research_logic( | |
| topic, keywords, synthesis_result, refs_text, | |
| data_evidence=planning_data_md, mode=mode, | |
| ) | |
| ) | |
| agent.record_prewrite_stage("research_logic", prewrite_started, prewrite_calls) | |
| research_logic_contract = re.sub( | |
| r"<think>.*?</think>", "", research_logic_response, flags=re.DOTALL | |
| ).strip() | |
| if not research_logic_contract: | |
| raise RuntimeError("research_logic_contract_missing") | |
| agent.global_memory["research_logic_contract"] = research_logic_contract | |
| self._progress(14, "working_front_matter") | |
| working_front = agent.generate_working_front_matter( | |
| topic, keywords, research_logic_contract, synthesis_result, | |
| data_evidence=planning_data_md, mode=mode, | |
| ) | |
| working_abstract = str(working_front.get("abstract") or "").strip() | |
| working_keywords = list(working_front.get("keywords") or []) | |
| if not working_abstract or not working_keywords: | |
| raise RuntimeError("working_front_matter_missing") | |
| working_abstract, front_fact_rows = self._repair_validate_study_fact_contract( | |
| mode, agent, quant_engine, working_abstract, data_metadata, | |
| section_name="Working Abstract", | |
| ) | |
| if not working_abstract: | |
| raise RuntimeError("working_front_matter_empty_after_study_fact_contract") | |
| if front_fact_rows: | |
| logging.warning( | |
| "Working Abstract study-fact repair | mode=%s count=%s types=%s", | |
| mode, len(front_fact_rows), | |
| sorted({str(row.get("type") or "") for row in front_fact_rows}), | |
| ) | |
| quality_diagnostics.append({ | |
| "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract", | |
| "status":"deterministic_repair", "count":len(front_fact_rows), | |
| "examples":front_fact_rows[:3], | |
| }) | |
| agent.global_memory["working_abstract"] = working_abstract | |
| self._progress(14, "outline_planning") | |
| plan = agent.generate_outline(topic, keywords, mode=mode, data_metadata=data_metadata) | |
| en_topic = str(plan.get("en_topic", topic)) | |
| structure = plan.get("structure", []) | |
| full_draft_en = ( | |
| f"# {en_topic}\n\n## Abstract\n{working_abstract}\n\n" | |
| f"## Keywords\n{', '.join(working_keywords)}\n\n" | |
| ) | |
| # Publish the Working Abstract/Keywords immediately, before JIT and | |
| # citation planning, so the user sees the manuscript anchor before | |
| # Introduction while those longer planning stages continue. | |
| self._publish_preview(full_draft_en, "working_front_matter", force=True) | |
| front_budget = max(180, int(APP_CONFIG.get("word_budget", {}).get("front_matter_words", 220))) | |
| body_target_words = max(1000, target_words - front_budget) | |
| adaptive_structure = agent.build_generation_tasks(structure, body_target_words, safe_chunk_words) | |
| for task in adaptive_structure: | |
| task["is_first"] = bool(task.get("section_first")) | |
| quote_invocation_plan = ( | |
| self._build_qualitative_quote_invocation_plan(adaptive_structure, data_metadata) | |
| if mode == "asq" else {} | |
| ) | |
| jit_texts = agent.prefetch_jit_literature(topic, keywords, adaptive_structure) | |
| if jit_texts: | |
| rag_engine._initialize_chunks([jit_texts]) | |
| self._progress(15, "citation_planning") | |
| citation_minimums, citation_reserve_targets = self._adaptive_citation_requirements( | |
| adaptive_structure, mode | |
| ) | |
| citation_top_level_minimums, citation_top_level_reserve_targets = self._top_level_citation_requirements( | |
| adaptive_structure, mode | |
| ) | |
| citation_unit_section_map = self._citation_unit_section_map(adaptive_structure) | |
| citation_plan_target = sum(citation_top_level_reserve_targets.values()) | |
| citation_plan = agent.build_claim_evidence_plan( | |
| adaptive_structure, topic, synthesis_result, mode=mode, | |
| target_unique=max(1, citation_plan_target), | |
| ) | |
| # Early neutral risk signal: evaluate the high-relevance plan before | |
| # reserve filling/JIT can make a raw candidate count look healthier | |
| # than the claim-specific plan actually is. This is advisory only. | |
| initial_top_level_counts = self._top_level_unique_tag_counts( | |
| citation_plan, citation_unit_section_map | |
| ) | |
| initial_shortfalls = [ | |
| (section_id, int(initial_top_level_counts.get(section_id, 0)), required) | |
| for section_id, required in citation_top_level_minimums.items() | |
| if int(initial_top_level_counts.get(section_id, 0)) < int(required) | |
| ] | |
| if initial_shortfalls: | |
| self._record_evidence_coverage_event( | |
| state, initial_shortfalls, phase="initial_claim_specific_plan", | |
| preview_status="limited_risk", | |
| ) | |
| # Fill unit-specific reserve gaps before expensive drafting. Existing | |
| # retrieved evidence is preferred; bounded contextual JIT is last resort. | |
| planned_tags = { | |
| str(item.get("tag") or "") | |
| for items in citation_plan.values() | |
| for item in items | |
| if item.get("tag") | |
| } | |
| task_by_name = { | |
| str(task.get("display_name") or ""): task | |
| for task in adaptive_structure | |
| if str(task.get("display_name") or "") | |
| } | |
| for display_name, reserve_target in citation_reserve_targets.items(): | |
| planned_items = citation_plan.setdefault(display_name, []) | |
| shortfall = max(0, reserve_target - len(planned_items)) | |
| if shortfall <= 0: | |
| continue | |
| section_context = self._citation_task_context( | |
| task_by_name.get(display_name, {}), research_logic_contract, synthesis_result | |
| ) | |
| recovery_items = agent.select_contextual_citation_recovery_items( | |
| [], section_name=display_name, section_context=section_context, | |
| needed=shortfall, excluded_tags=planned_tags, | |
| ) | |
| for item in recovery_items: | |
| tag = str(item.get("tag") or "") | |
| if tag and tag not in planned_tags: | |
| planned_items.append(item) | |
| planned_tags.add(tag) | |
| # Do not trigger external retrieval merely because a technical | |
| # generation chunk is below a word-derived target. Fatal evidence | |
| # authority belongs to the canonical top-level academic section. | |
| top_level_counts = self._top_level_unique_tag_counts(citation_plan, citation_unit_section_map) | |
| for section_id, required in citation_top_level_minimums.items(): | |
| deficit = max(0, int(required) - int(top_level_counts.get(section_id, 0))) | |
| if deficit <= 0: | |
| continue | |
| section_units = [ | |
| name for name, mapped in citation_unit_section_map.items() if mapped == section_id | |
| ] | |
| if not section_units: | |
| continue | |
| primary_unit = section_units[0] | |
| section_context = " ".join( | |
| self._citation_task_context(task_by_name.get(name, {}), research_logic_contract, synthesis_result) | |
| for name in section_units | |
| ) | |
| jit_items = agent.fetch_contextual_jit_recovery_items( | |
| topic=topic, section_name=section_id, section_context=section_context, | |
| needed=deficit, excluded_tags=planned_tags, | |
| ) | |
| planned_items = citation_plan.setdefault(primary_unit, []) | |
| for item in jit_items: | |
| tag = str(item.get("tag") or "") | |
| if tag and tag not in planned_tags: | |
| planned_items.append(item) | |
| planned_tags.add(tag) | |
| top_level_counts = self._top_level_unique_tag_counts(citation_plan, citation_unit_section_map) | |
| shortfalls = [ | |
| (section_id, int(top_level_counts.get(section_id, 0)), required) | |
| for section_id, required in citation_top_level_minimums.items() | |
| if int(top_level_counts.get(section_id, 0)) < int(required) | |
| ] | |
| if shortfalls: | |
| self._record_evidence_coverage_event( | |
| state, shortfalls, phase="bounded_retrieval_complete", | |
| preview_status="limited_risk", | |
| ) | |
| logging.warning( | |
| "Top-level citation feasibility limited: minimum_total=%s reserve_target=%s planned_unique=%s sections=%s nonfatal=true", | |
| sum(citation_top_level_minimums.values()), citation_plan_target, len(planned_tags), len(citation_top_level_minimums), | |
| ) | |
| else: | |
| logging.info( | |
| "Top-level citation feasibility passed: minimum_total=%s reserve_target=%s planned_unique=%s sections=%s", | |
| sum(citation_top_level_minimums.values()), citation_plan_target, len(planned_tags), len(citation_top_level_minimums), | |
| ) | |
| logging.info( | |
| "Top-level citation planning complete: minimum_total=%s reserve_target=%s planned_unique=%s sections=%s", | |
| sum(citation_top_level_minimums.values()), citation_plan_target, len(planned_tags), len(citation_top_level_minimums), | |
| ) | |
| canonical_concept_ledger = self._build_canonical_concept_ledger( | |
| mode, data_metadata, keywords=keywords, adaptive_structure=adaptive_structure, | |
| research_logic_contract=research_logic_contract, | |
| ) | |
| state.update( | |
| { | |
| "adaptive_structure": adaptive_structure, | |
| "quote_invocation_plan": quote_invocation_plan, | |
| "citation_plan": citation_plan, | |
| "citation_minimums": citation_minimums, | |
| "citation_reserve_targets": citation_reserve_targets, | |
| "citation_top_level_minimums": citation_top_level_minimums, | |
| "citation_top_level_reserve_targets": citation_top_level_reserve_targets, | |
| "citation_unit_section_map": citation_unit_section_map, | |
| "citation_validated_tags_by_unit": {}, | |
| "citation_required_claim_counts_by_unit": {}, | |
| "section_integrity_seals": {}, | |
| "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION, | |
| "research_logic_contract": research_logic_contract, | |
| "working_abstract": working_abstract, | |
| "working_keywords": working_keywords, | |
| "front_matter_phase": "working", | |
| "body_target_words": body_target_words, | |
| "full_draft_en": full_draft_en, | |
| "full_draft_translated": "", | |
| "translation_sections": {}, | |
| "translation_glossary": {}, | |
| "canonical_concept_ledger": canonical_concept_ledger, | |
| "completed_sections": [], | |
| "context_tail": "", | |
| "accumulated_deviation": 0.0, | |
| "agent_state": agent.export_state(), | |
| "outline_ready": True, | |
| } | |
| ) | |
| self._publish_preview(full_draft_en, "outline_ready", force=True) | |
| self._checkpoint("outline_ready", state, 15) | |
| else: | |
| full_draft_en = str(state.get("full_draft_en") or "") | |
| # JIT text itself is already represented in agent citation state; | |
| # refs + user literature rebuild a safe RAG baseline on resume. | |
| completed_sections = list(state.get("completed_sections") or []) | |
| context_tail = str(state.get("context_tail") or "") | |
| accumulated_deviation = float(state.get("accumulated_deviation") or 0.0) | |
| research_logic_contract = str( | |
| state.get("research_logic_contract") or agent.global_memory.get("research_logic_contract") or "" | |
| ) | |
| citation_plan = state.get("citation_plan") if isinstance(state.get("citation_plan"), dict) else {} | |
| citation_minimums = dict(state.get("citation_minimums") or {}) | |
| citation_reserve_targets = dict(state.get("citation_reserve_targets") or {}) | |
| citation_top_level_minimums = dict(state.get("citation_top_level_minimums") or {}) | |
| citation_top_level_reserve_targets = dict(state.get("citation_top_level_reserve_targets") or {}) | |
| citation_unit_section_map = dict(state.get("citation_unit_section_map") or {}) | |
| citation_required_claim_counts_by_unit = { | |
| str(name): int(count or 0) | |
| for name, count in dict(state.get("citation_required_claim_counts_by_unit") or {}).items() | |
| } | |
| citation_validated_tags_by_unit = { | |
| str(name): list(tags or []) | |
| for name, tags in dict(state.get("citation_validated_tags_by_unit") or {}).items() | |
| } | |
| section_integrity_seals = { | |
| str(key): dict(record or {}) | |
| for key, record in dict(state.get("section_integrity_seals") or {}).items() | |
| if isinstance(record, dict) | |
| } | |
| if not citation_minimums: | |
| citation_minimums, citation_reserve_targets = self._adaptive_citation_requirements( | |
| adaptive_structure, mode | |
| ) | |
| if not citation_top_level_minimums: | |
| citation_top_level_minimums, citation_top_level_reserve_targets = self._top_level_citation_requirements( | |
| adaptive_structure, mode | |
| ) | |
| if not citation_unit_section_map: | |
| citation_unit_section_map = self._citation_unit_section_map(adaptive_structure) | |
| translation_sections = dict(state.get('translation_sections') or {}) | |
| translation_glossary = dict(state.get('translation_glossary') or {}) | |
| canonical_concept_ledger = state.get('canonical_concept_ledger') | |
| if not isinstance(canonical_concept_ledger, dict) or not canonical_concept_ledger.get('items'): | |
| canonical_concept_ledger = self._build_canonical_concept_ledger( | |
| mode, data_metadata, keywords=keywords, adaptive_structure=adaptive_structure, | |
| research_logic_contract=research_logic_contract, | |
| ) | |
| state['canonical_concept_ledger'] = canonical_concept_ledger | |
| working_abstract = str(state.get("working_abstract") or "").strip() | |
| working_keywords = list(state.get("working_keywords") or []) | |
| if working_abstract and not state.get("english_finalized"): | |
| checked_working_abstract, resumed_front_rows = self._repair_validate_study_fact_contract( | |
| mode, agent, quant_engine, working_abstract, data_metadata, | |
| section_name="Working Abstract", | |
| ) | |
| if not checked_working_abstract: | |
| raise RuntimeError("working_front_matter_empty_after_study_fact_contract") | |
| if checked_working_abstract != working_abstract or resumed_front_rows: | |
| working_abstract = checked_working_abstract | |
| state["working_abstract"] = working_abstract | |
| if str(state.get("full_draft_en") or "").strip(): | |
| state["full_draft_en"] = replace_abstract(str(state.get("full_draft_en") or ""), working_abstract) | |
| full_draft_en = str(state["full_draft_en"] or "") | |
| quality_diagnostics.append({ | |
| "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract", | |
| "status":"resume_deterministic_repair", "count":len(resumed_front_rows), | |
| "examples":resumed_front_rows[:3], | |
| }) | |
| state["quality_diagnostics"] = quality_diagnostics | |
| self._checkpoint("working_front_matter_fact_reconciled", state, 15) | |
| if working_abstract: | |
| agent.global_memory["working_abstract"] = working_abstract | |
| if self._preview_target_text == '': | |
| self._preview_target_text = str(state.get('full_draft_translated') or '') | |
| # P0-R1.1-R2 compatibility migration: old checkpoints predate Section | |
| # Integrity Seals. Reconcile only already-completed English sections with | |
| # deterministic local contracts, seal them, and invalidate target cache | |
| # only when the exact English section changes. Never call the writer. | |
| seal_migration = False | |
| seal_text_changed = False | |
| for section_key, section_title in self._unique_top_level_sections(adaptive_structure): | |
| if not self._section_is_complete(adaptive_structure, completed_sections, section_key): | |
| continue | |
| section_text = self._extract_top_level_section(full_draft_en, section_title) | |
| if not section_text: | |
| continue | |
| existing_seal = dict(section_integrity_seals.get(section_key) or {}) | |
| if ( | |
| existing_seal.get("version") == ENGLISH_SECTION_SEAL_VERSION | |
| and existing_seal.get("sha256") == self._section_integrity_hash(section_text) | |
| ): | |
| continue | |
| migrated_text, enum_rows = agent.repair_and_validate_enumeration_contract( | |
| section_text, section_name=f"Resume sealed section: {section_title}" | |
| ) | |
| migrated_text, fact_rows = self._repair_validate_study_fact_contract( | |
| mode, agent, quant_engine, migrated_text, data_metadata, | |
| section_name=f"Resume sealed section: {section_title}", | |
| ) | |
| if migrated_text.strip() != section_text.strip(): | |
| full_draft_en = self._replace_top_level_section( | |
| full_draft_en, section_title, migrated_text | |
| ) | |
| translation_sections.pop(section_key, None) | |
| seal_text_changed = True | |
| quality_diagnostics.append({ | |
| "scope":"english_section_seal_migration", "name":section_title, | |
| "status":"deterministic_repair", "count":len(enum_rows) + len(fact_rows), | |
| "examples":(list(enum_rows or []) + list(fact_rows or []))[:3], | |
| }) | |
| section_text = self._extract_top_level_section(full_draft_en, section_title) | |
| section_integrity_seals[section_key] = self._build_section_integrity_seal( | |
| section_key, section_title, section_text | |
| ) | |
| seal_migration = True | |
| if seal_migration: | |
| # Rebuild mutable manuscript memory only from sealed English text so | |
| # rejected/raw claims cannot propagate into later sections. | |
| for memory_key in ("hypotheses", "variables", "model_summary", "method_summary"): | |
| agent.global_memory[memory_key] = "" | |
| last_sealed = "" | |
| for section_key, section_title in self._unique_top_level_sections(adaptive_structure): | |
| if not self._section_is_complete(adaptive_structure, completed_sections, section_key): | |
| continue | |
| sealed_text = self._extract_top_level_section(full_draft_en, section_title) | |
| if sealed_text: | |
| agent._extract_memory_from_text(sealed_text, section_title) | |
| last_sealed = sealed_text | |
| if last_sealed: | |
| context_tail = " ".join(last_sealed.split()[-200:]) | |
| state.update({ | |
| "full_draft_en": full_draft_en, | |
| "translation_sections": translation_sections, | |
| "section_integrity_seals": section_integrity_seals, | |
| "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION, | |
| "context_tail": context_tail, | |
| "quality_diagnostics": quality_diagnostics, | |
| "agent_state": agent.export_state(), | |
| }) | |
| if seal_text_changed: | |
| state["full_draft_translated"] = self._assemble_partial_translation( | |
| translation_sections, adaptive_structure, include_front=True | |
| ) | |
| self._checkpoint("english_section_seal_migrated", state, 15) | |
| # Target-language work remains Executor-owned but no longer blocks the | |
| # English writer. A dedicated single translation agent/worker keeps | |
| # glossary order deterministic and avoids sharing mutable writer state. | |
| translation_agent = SSciPipelineAgent(ds_key, s2_key) | |
| translation_agent.set_execution_progress_hook(lambda: self._progress_activity()) | |
| translation_agent.set_cancel_checker(lambda: self.cancel_checker()) | |
| seeded_glossary = translation_agent.seed_canonical_concept_glossary( | |
| canonical_concept_ledger, target_lang, translation_glossary | |
| ) | |
| if seeded_glossary != translation_glossary: | |
| translation_glossary = dict(seeded_glossary) | |
| state['translation_glossary'] = translation_glossary | |
| state['canonical_concept_ledger'] = canonical_concept_ledger | |
| self._checkpoint('canonical_concept_ledger_ready', state, 15) | |
| current_entity_ledger_version = str( | |
| getattr(translation_agent, 'PROTECTED_ENTITY_LEDGER_VERSION', 'protected-entity-ledger-v1') | |
| or 'protected-entity-ledger-v1' | |
| ) | |
| stale_entity_units = [ | |
| key for key, record in translation_sections.items() | |
| if isinstance(record, dict) | |
| and record.get('translated_text') | |
| and record.get('qa_status') in {'passed', 'passed_with_warning'} | |
| and record.get('protected_entity_ledger_version') != current_entity_ledger_version | |
| ] | |
| if stale_entity_units: | |
| logging.warning( | |
| 'Protected Entity Ledger contract advanced; invalidating target-language cache only | units=%s version=%s', | |
| stale_entity_units[:8], current_entity_ledger_version, | |
| ) | |
| state['translation_complete'] = False | |
| state['package_validated'] = False | |
| state['full_draft_translated'] = '' | |
| self._preview_target_text = '' | |
| self._checkpoint('protected_entity_ledger_advanced', state, 15) | |
| self._start_translation_worker( | |
| translation_agent, | |
| translation_sections=translation_sections, | |
| translation_glossary=translation_glossary, | |
| adaptive_structure=adaptive_structure, | |
| target_lang=target_lang, | |
| ) | |
| # Capture a dedicated abort Event, not the worker lifecycle stop Event. | |
| # The worker is gracefully closed before the strict final translation | |
| # gate, but the same translation agent remains valid for deferred/final | |
| # units. Abortive shutdown still cancels any in-flight target call. | |
| translation_abort_event = self._translation_worker.cancel_event | |
| translation_agent.set_cancel_checker( | |
| lambda: self.cancel_checker() or translation_abort_event.is_set() | |
| ) | |
| # Backward-compatible resume path for a pre-R3.2.8.3 checkpoint that has | |
| # an outline but no Working Abstract yet. Generate it once, checkpoint it, | |
| # and reuse it thereafter; never regenerate an unchanged anchor on resume. | |
| if not working_abstract and not state.get("english_finalized"): | |
| working_front = agent.generate_working_front_matter( | |
| topic, keywords, research_logic_contract, synthesis_result, | |
| data_evidence=planning_data_md, mode=mode, | |
| ) | |
| working_abstract = str(working_front.get("abstract") or "").strip() | |
| working_keywords = list(working_front.get("keywords") or []) | |
| if not working_abstract or not working_keywords: | |
| raise RuntimeError("working_front_matter_missing") | |
| working_abstract, front_fact_rows = self._repair_validate_study_fact_contract( | |
| mode, agent, quant_engine, working_abstract, data_metadata, | |
| section_name="Working Abstract", | |
| ) | |
| if not working_abstract: | |
| raise RuntimeError("working_front_matter_empty_after_study_fact_contract") | |
| if front_fact_rows: | |
| logging.warning( | |
| "Working Abstract resume study-fact repair | mode=%s count=%s types=%s", | |
| mode, len(front_fact_rows), | |
| sorted({str(row.get("type") or "") for row in front_fact_rows}), | |
| ) | |
| quality_diagnostics.append({ | |
| "scope":"working_abstract_study_fact_integrity", "name":"Working Abstract", | |
| "status":"deterministic_repair", "count":len(front_fact_rows), | |
| "examples":front_fact_rows[:3], | |
| }) | |
| agent.global_memory["working_abstract"] = working_abstract | |
| full_draft_en = replace_abstract(full_draft_en, working_abstract) | |
| full_draft_en = self._replace_keywords(full_draft_en, working_keywords) | |
| state.update({ | |
| "working_abstract": working_abstract, | |
| "working_keywords": working_keywords, | |
| "front_matter_phase": "working", | |
| "full_draft_en": full_draft_en, | |
| "agent_state": agent.export_state(), | |
| }) | |
| self._publish_preview(full_draft_en, "working_front_matter", force=True) | |
| self._checkpoint("working_front_matter_ready", state, 15) | |
| # Queue the QA-frozen Working front matter. Translation is allowed to | |
| # overlap later English work, but only frozen English units may enter | |
| # this Executor-owned ordered worker. | |
| if working_abstract and not state.get("english_finalized"): | |
| section_order_for_front = self._unique_top_level_sections(adaptive_structure) | |
| if section_order_for_front: | |
| first_section_title = section_order_for_front[0][1] | |
| working_front_source = self._extract_front_matter(full_draft_en, first_section_title) | |
| if not working_front_source and not completed_sections: | |
| # Fresh tasks contain only Title/Abstract/Keywords here; the | |
| # first body heading does not exist yet. Preserve legacy | |
| # resume behavior when completed body sections are present. | |
| working_front_source = str(full_draft_en or '').strip() | |
| self._enqueue_translation_unit( | |
| translation_agent, | |
| unit_key='__front_matter__', unit_name='Working Front Matter', | |
| source_text=working_front_source, | |
| translation_sections=translation_sections, | |
| english_commit_text=working_front_source, | |
| ) | |
| # Resume/idempotency: completed English top-level sections whose current | |
| # source-hash translation is absent/stale are queued exactly once. | |
| for section_key, section_title in self._unique_top_level_sections(adaptive_structure): | |
| if not self._section_is_complete(adaptive_structure, completed_sections, section_key): | |
| continue | |
| frozen_section = self._extract_top_level_section(full_draft_en, section_title) | |
| self._enqueue_translation_unit( | |
| translation_agent, | |
| unit_key=section_key, | |
| unit_name=section_title, | |
| source_text=frozen_section, | |
| translation_sections=translation_sections, | |
| english_commit_text=self._english_prefix_through_top_level(full_draft_en, section_title), | |
| ) | |
| translation_sections, translation_glossary = self._collect_translation_results( | |
| translation_sections, translation_glossary, block=False | |
| ) | |
| if not citation_plan: | |
| self._progress(15, "citation_planning") | |
| citation_plan = agent.build_claim_evidence_plan( | |
| adaptive_structure, topic, synthesis_result, mode=mode, | |
| target_unique=max(1, sum(citation_top_level_reserve_targets.values())), | |
| ) | |
| top_level_counts = self._top_level_unique_tag_counts(citation_plan, citation_unit_section_map) | |
| shortfalls = [ | |
| (section_id, int(top_level_counts.get(section_id, 0)), required) | |
| for section_id, required in citation_top_level_minimums.items() | |
| if int(top_level_counts.get(section_id, 0)) < int(required) | |
| ] | |
| if shortfalls: | |
| self._record_evidence_coverage_event( | |
| state, shortfalls, phase="resume_claim_specific_plan", | |
| preview_status="limited_risk", | |
| ) | |
| state["citation_plan"] = citation_plan | |
| state["citation_minimums"] = citation_minimums | |
| state["citation_reserve_targets"] = citation_reserve_targets | |
| state["citation_top_level_minimums"] = citation_top_level_minimums | |
| state["citation_top_level_reserve_targets"] = citation_top_level_reserve_targets | |
| state["citation_unit_section_map"] = citation_unit_section_map | |
| state["citation_required_claim_counts_by_unit"] = citation_required_claim_counts_by_unit | |
| state["citation_validated_tags_by_unit"] = citation_validated_tags_by_unit | |
| quote_invocation_plan = ( | |
| dict(state.get("quote_invocation_plan") or {}) if mode == "asq" else {} | |
| ) | |
| if mode == "asq": | |
| if not quote_invocation_plan: | |
| quote_invocation_plan = self._build_qualitative_quote_invocation_plan( | |
| adaptive_structure, data_metadata | |
| ) | |
| state["quote_invocation_plan"] = quote_invocation_plan | |
| self._checkpoint("quote_plan_ready", state, 15) | |
| self._validate_qualitative_quote_invocation_plan( | |
| quote_invocation_plan, adaptive_structure, data_metadata | |
| ) | |
| total_steps = len(adaptive_structure) | |
| for sec in adaptive_structure: | |
| display_name = sec["display_name"] | |
| if display_name in completed_sections: | |
| continue | |
| if self.cancel_checker(): | |
| raise RuntimeError("task_cancel_requested") | |
| remaining = sum(1 for item in adaptive_structure if item["display_name"] not in completed_sections) | |
| adjusted_target = self._bounded_target( | |
| int(sec["target_words"]), 0.0, remaining, safe_chunk_words | |
| ) | |
| sec_title = sec["section"] | |
| subsection = sec.get("subsection") | |
| is_quarantined = any( | |
| word in sec_title.lower() | |
| for word in ["abstract", "keyword", "conclusion", "references", "appendix"] | |
| ) | |
| planned_items = list(citation_plan.get(display_name, [])) | |
| assigned_tag_list = [str(item.get("tag") or "") for item in planned_items if item.get("tag")] | |
| claim_plan_text = agent.format_claim_evidence_plan(planned_items) | |
| section_id = str(sec.get("section_id") or "") | |
| section_role = str( | |
| APP_CONFIG.get("section_reasoning_profiles", {}).get(mode, {}).get(section_id) or "" | |
| ) | |
| mode_directive = "" | |
| if section_id in {"theory", "theoretical_framing"}: | |
| mode_directive = str(APP_CONFIG.get("hypothesis_development_directive") or "") | |
| elif section_id in {"methodology", "methods"}: | |
| mode_directive = str(APP_CONFIG.get("methodological_defense_directive") or "") | |
| elif section_id == "discussion": | |
| mode_directive = str(APP_CONFIG.get("discussion_depth_directive") or "") | |
| elif section_id == "conclusion": | |
| mode_directive = str(APP_CONFIG.get("conclusion_anchor_directive") or "") | |
| if is_quarantined or section_id in {"results", "findings"}: | |
| citation_rule = ( | |
| "NO EXTERNAL LITERATURE CITATIONS ALLOWED IN THIS UNIT. " | |
| "Report only frozen empirical/qualitative evidence and the interpretation permitted by the section role." | |
| ) | |
| else: | |
| citation_rule = ( | |
| "Use the citation-first claim-evidence plan below as bounded evidence options, not as a claim quota. " | |
| "Do not create a new claim merely to consume an assigned citation. Use a source only when it directly supports a claim independently required by the frozen research logic, section role, or already established manuscript argument; omit any item that would introduce another application domain or mutate the study's facts. " | |
| "When a source is used, keep the claim no broader than the supplied evidence and attach exactly its assigned [CITE-XX] tag. " | |
| "Never stack sources, substitute tags, or use literature as evidence for this study's own facts.\n" | |
| f"<CLAIM_EVIDENCE_PLAN>\n{claim_plan_text}\n</CLAIM_EVIDENCE_PLAN>" | |
| ) | |
| if not claim_plan_text.strip(): | |
| citation_rule += ( | |
| "\nEVIDENCE-LIMITED WRITING RULE: no directly validated source is assigned to this unit. " | |
| "Continue the scholarly argument using frozen study evidence, cautious conceptual reasoning, and general scholarly knowledge. " | |
| "Do not invent citations, authors, titles, DOI values, or bibliographic metadata; do not describe general model knowledge as if it were retrieved or directly validated literature. " | |
| "Use appropriately qualified language for propositions, interpretations, and research opportunities." | |
| ) | |
| if mode == "asq" and section_id not in {"findings", "appendix"}: | |
| citation_rule += ( | |
| "\nASQ EVIDENCE-CHANNEL RULE: A sentence whose factual subject is this study's setting, participants, interviews, data sources, cases, actors, observations, events, quotations, analytic steps, or empirical findings must not carry an external [CITE-XX] tag. " | |
| "Place any prior-literature or methodological-rationale claim in a separate sentence and cite only that external claim." | |
| ) | |
| memory_context = agent._build_memory_context(sec_title, 2500) | |
| memory_block = f"\n<GLOBAL_MEMORY>\n{memory_context}\n</GLOBAL_MEMORY>\n" if memory_context else "" | |
| snippets = rag_engine.get_relevant_context(display_name) if rag_engine is not None else "" | |
| if snippets: | |
| memory_block += f"\n<TARGETED_RAG>\n{snippets}\n</TARGETED_RAG>\n" | |
| heading_directive = "Do not output the section heading; it will be inserted by the system." | |
| if subsection: | |
| heading_directive += ( | |
| f" Do not output the subsection heading '{subsection}'; it will also be inserted by the system. " | |
| "Use level-3 headings only when analytically necessary." | |
| ) | |
| elif sec_title.lower() in {"introduction", "conclusion"}: | |
| heading_directive += " Do not use additional headings." | |
| else: | |
| heading_directive += ( | |
| " Use level-2 or level-3 headings only when they clarify the argument; " | |
| "never use level-4 headings or numbered headings." | |
| ) | |
| table_rule = ( | |
| "Present only frozen statistical evidence in standard Markdown tables with plain-text captions." | |
| if sec["is_method_or_result"] and mode == "amj" else "" | |
| ) | |
| concepts = ", ".join(sec.get("concepts") or []) | |
| theme_evidence = [item for item in (sec.get("theme_evidence") or []) if isinstance(item, dict)] | |
| evidence_profiles = [] | |
| for item in theme_evidence: | |
| stage = str(item.get("stage") or "Not specified").strip() | |
| role = str(item.get("evidence_role") or "typical").strip() | |
| case_label = str(item.get("case_label") or "Not specified").strip() | |
| evidence_profiles.append(f"stage={stage}; case={case_label}; role={role}") | |
| planned_quote_count = sum( | |
| 1 for record in quote_invocation_plan.values() | |
| if str(record.get("render_task_key") or "") == self._qualitative_task_key(sec) | |
| ) | |
| qualitative_rule = ( | |
| f"Assigned qualitative concepts: {concepts}. Assigned evidence profiles: {' | '.join(evidence_profiles)}. " | |
| f"The Executor has frozen {planned_quote_count} quotation invocation(s) for this task. " | |
| "Use the assigned evidence analytically, but output narrative prose only: do not reproduce any verbatim quotation and do not output any QUOTE-XX identifier. " | |
| "The Executor inserts the frozen quotations deterministically after narrative/citation QA. Preserve stage/case/deviant meaning and use deviant evidence to qualify rather than erase the pattern." | |
| if concepts or evidence_profiles else "" | |
| ) | |
| scoped_data_md = user_data_md | |
| if mode == "asq": | |
| if section_id == "appendix": | |
| # The appendix is the only non-owning unit allowed to see | |
| # the full frozen evidence table. It is excluded from the | |
| # body duplicate-quote gate by design. | |
| scoped_data_md = user_data_md | |
| elif section_id == "findings": | |
| current_theme = str(sec.get("theme_title") or "").strip() | |
| scoped_data_md = qual_engine.build_scoped_evidence_md( | |
| data_metadata, | |
| include_themes=True, | |
| theme_titles=[current_theme] if current_theme else None, | |
| assigned_evidence=theme_evidence, | |
| include_quotes=bool(theme_evidence), | |
| include_quote_ids=False, | |
| ) | |
| elif section_id == "discussion": | |
| scoped_data_md = qual_engine.build_scoped_evidence_md( | |
| data_metadata, | |
| include_themes=True, | |
| ) | |
| else: | |
| scoped_data_md = qual_engine.build_scoped_evidence_md( | |
| data_metadata, | |
| include_themes=False, | |
| ) | |
| evidence_packet = ( | |
| f"{base_evidence_contract}\n<FROZEN_RESEARCH_EVIDENCE>\n{scoped_data_md}\n</FROZEN_RESEARCH_EVIDENCE>" | |
| if scoped_data_md else base_evidence_contract | |
| ) | |
| completed_label = ", ".join(completed_sections) | |
| anti_repeat = ( | |
| f"Already completed: {completed_label}. Do not restate their background, gap, or contribution." | |
| if completed_label else "" | |
| ) | |
| if re.search(r"\(Part\s+2/2\)", display_name, flags=re.I): | |
| anti_repeat += " Part 2/2 must extend and complement Part 1/2; do not recap its background, mechanism explanation, evidence, or contribution framing." | |
| voice = str(APP_CONFIG.get("journal_voice_profiles", {}).get(mode) or "") | |
| sys_prompt = ( | |
| APP_CONFIG.get("sys_academic_writer", "") + "\n" | |
| + evidence_packet + "\n" + voice + "\n" | |
| + APP_CONFIG.get("deep_refinement_directive", "") | |
| ) | |
| word_budget = APP_CONFIG.get("word_budget", {}) | |
| min_ratio = float(word_budget.get("section_min_ratio", 0.92)) | |
| max_ratio = float(word_budget.get("section_max_ratio", 1.08)) | |
| prompt_context_tail = context_tail[-2500:] | |
| if mode == "asq": | |
| prompt_context_tail = self._redact_qualitative_quotes(prompt_context_tail, data_metadata) | |
| user_prompt = f"""Write the assigned English manuscript unit. | |
| Section: {sec_title} | |
| Subsection: {subsection or 'None'} | |
| Section reasoning role: {section_role} | |
| Additional mode directive: {mode_directive} | |
| Planning length: approximately {int(adjusted_target * min_ratio)}-{int(adjusted_target * max_ratio)} words. This is guidance, not a content quota. Prioritize Conciseness, Accuracy, and Rigor; never add filler or remove useful reasoning solely to hit the range. | |
| {heading_directive} | |
| {citation_rule} | |
| {table_rule} | |
| {str(sec.get('required_content') or '')} | |
| {qualitative_rule} | |
| {anti_repeat} | |
| {self._canonical_concept_writer_directive(canonical_concept_ledger)} | |
| <PROVISIONAL_RESEARCH_LOGIC> | |
| {research_logic_contract} | |
| </PROVISIONAL_RESEARCH_LOGIC> | |
| <WORKING_ABSTRACT_ANCHOR> | |
| {working_abstract} | |
| </WORKING_ABSTRACT_ANCHOR> | |
| ANCHOR PRECEDENCE: Frozen research evidence and the PROVISIONAL_RESEARCH_LOGIC override the Working Abstract if any tension appears. The Working Abstract is continuity guidance only and must never force a result. | |
| {memory_block} | |
| Literature synthesis: {synthesis_result} | |
| <FROZEN_RESEARCH_EVIDENCE> | |
| {scoped_data_md} | |
| </FROZEN_RESEARCH_EVIDENCE> | |
| Previous tail: {prompt_context_tail} | |
| Output only final academic prose. Do not report the word count.""" | |
| if agent._count_tokens(sys_prompt) + agent._count_tokens(user_prompt) > 60000: | |
| user_prompt = agent._truncate_context(user_prompt, 60000 - agent._count_tokens(sys_prompt)) | |
| current_index = len(completed_sections) + 1 | |
| pct = min(88, 15 + int((current_index / max(1, total_steps)) * 72)) | |
| self._progress(pct, f"writing:{current_index}:{total_steps}") | |
| prefix = "" | |
| if sec.get("section_first"): | |
| prefix += f"# {sec_title}\n\n" | |
| if sec.get("subsection_first") and subsection: | |
| prefix += f"## {subsection}\n\n" | |
| if mode == "asq" and section_id == "appendix": | |
| # R6A-R4: Appendix is a deterministic reader-facing projection | |
| # of the Evidence Ledger. It is never regenerated by the LLM. | |
| en_raw = qual_engine.build_appendix_evidence_md(data_metadata) | |
| else: | |
| en_raw = self._stream_text( | |
| agent.stream_deepseek( | |
| sys_prompt, user_prompt, max_tokens=7000, | |
| model_role="writing", thinking=False, allow_fallback=False, | |
| ), | |
| preview_prefix=full_draft_en + prefix, | |
| preview_stage=display_name, | |
| ) | |
| en_raw = re.sub(r"<think>.*?</think>", "", en_raw, flags=re.S).strip() | |
| en_raw = re.sub(rf"(?im)^#\s*{re.escape(sec_title)}\s*\n+", "", en_raw).strip() | |
| if subsection: | |
| en_raw = re.sub(rf"(?im)^##\s*{re.escape(subsection)}\s*\n+", "", en_raw).strip() | |
| en_raw = re.sub(r"(?m)^(#{2,3})\s+[\d\.\s一二三四五六七八九十、]+\s*(.*)", r"\1 \2", en_raw) | |
| en_raw = re.sub(r"(?m)^#{4,}\s+(.*)", r"**\1**", en_raw) | |
| en_raw = re.sub(r"(?m)^#\s+(.+)$", r"## \1", en_raw) | |
| en_raw = agent.critique_and_repair_section( | |
| en_raw, section_id=section_id, section_name=display_name, mode=mode, | |
| evidence_contract=evidence_packet, | |
| ) | |
| en_raw = agent.polish_academic_section(en_raw, display_name, abstract=False, mode=mode) | |
| if mode == "asq": | |
| # Model prose has no authority over quote presence/count. | |
| en_raw = self._strip_model_controlled_qualitative_quotes(en_raw, data_metadata) | |
| allow_deterministic_citation_reinforcement = not ( | |
| mode == "asq" and section_id in {"methods", "discussion"} | |
| ) | |
| if not is_quarantined and assigned_tag_list and allow_deterministic_citation_reinforcement: | |
| en_raw = agent.reinforce_claim_specific_citation_coverage( | |
| en_raw, assigned_tag_list, section_name=display_name | |
| ) | |
| self._progress(pct, f"citation_audit:{current_index}:{total_steps}") | |
| used_before = set(agent.used_citations) | |
| en_resolved = agent.audit_and_resolve_citations( | |
| prefix + en_raw, section_name=display_name, quarantine=is_quarantined | |
| ) | |
| used_after = set(agent.used_citations) | |
| survived = used_after - used_before | |
| if assigned_tag_list and not is_quarantined: | |
| required_survival = min( | |
| len(assigned_tag_list), | |
| max(1, int(math.ceil(len(assigned_tag_list) * CITATION_SEMANTIC_SURVIVAL_RATIO))), | |
| ) | |
| if len(survived) < required_survival: | |
| missing_items = [ | |
| item for item in planned_items | |
| if str(item.get("tag") or "") not in used_after | |
| ] | |
| needed = max(0, required_survival - len(survived)) | |
| recovery_items = agent.select_contextual_citation_recovery_items( | |
| missing_items, section_name=display_name, section_context=en_resolved, | |
| needed=needed, excluded_tags=used_after, | |
| ) | |
| if len(recovery_items) < needed: | |
| recovery_items.extend( | |
| agent.fetch_contextual_jit_recovery_items( | |
| topic=topic, section_name=display_name, section_context=en_resolved, | |
| needed=needed - len(recovery_items), | |
| excluded_tags=used_after | {str(item.get("tag") or "") for item in recovery_items}, | |
| ) | |
| ) | |
| planned_tags_for_unit = {str(item.get("tag") or "") for item in planned_items} | |
| for recovery_item in recovery_items[:needed]: | |
| tag = str(recovery_item.get("tag") or "") | |
| if tag and tag not in planned_tags_for_unit: | |
| planned_items.append(recovery_item) | |
| planned_tags_for_unit.add(tag) | |
| citation_plan[display_name] = planned_items | |
| # R14-TQ2-PQ1-R1: recovery may attach evidence only to | |
| # an already-written claim. It is never authorized to create a | |
| # supplemental paragraph from a source merely to satisfy citation | |
| # survival. If no existing claim can be validated, leave prose | |
| # unchanged. Evidence coverage may later be reported as limited, | |
| # while citation correctness itself remains fail-closed. | |
| self._progress(pct, f"citation_recovery:{current_index}:{total_steps}") | |
| recovered = agent.recover_citations_on_existing_claims( | |
| en_resolved, | |
| recovery_items[:needed], | |
| section_name=display_name, | |
| needed=needed, | |
| ) | |
| if recovered != en_resolved: | |
| en_resolved = recovered | |
| used_after = set(agent.used_citations) | |
| survived = used_after - used_before | |
| if mode == "asq" and section_id == "findings": | |
| en_resolved = self._render_planned_qualitative_quotes( | |
| en_resolved, | |
| task=sec, | |
| data_metadata=data_metadata, | |
| quote_invocation_plan=quote_invocation_plan, | |
| ) | |
| # P0-R1.1-R2: every mutating deterministic operation must complete | |
| # before the Evidence Fact contract seals the English unit. | |
| en_resolved, enumeration_rows = agent.repair_and_validate_enumeration_contract( | |
| en_resolved, section_name=display_name | |
| ) | |
| if enumeration_rows: | |
| logging.warning( | |
| "Structural enumeration repaired | unit=%s count=%s details=%s", | |
| display_name, len(enumeration_rows), enumeration_rows[:3], | |
| ) | |
| quality_diagnostics.append({ | |
| "scope":"structural_enumeration", "name":display_name, | |
| "status":"deterministic_repair", "count":len(enumeration_rows), | |
| "examples":enumeration_rows[:3], | |
| }) | |
| en_resolved, duplicate_rows = self._remove_severe_cross_unit_duplicates(en_resolved, full_draft_en) | |
| if duplicate_rows: | |
| logging.warning("Severe cross-unit redundancy removed | unit=%s count=%s max_similarity=%s", display_name, len(duplicate_rows), max(row["similarity"] for row in duplicate_rows)) | |
| quality_diagnostics.append({"scope":"manuscript_redundancy","name":display_name,"status":"bounded_duplicate_removed","count":len(duplicate_rows),"examples":duplicate_rows[:3]}) | |
| en_resolved, sentence_echo_rows = agent.remove_severe_cross_unit_sentence_echoes( | |
| en_resolved, full_draft_en | |
| ) | |
| if sentence_echo_rows: | |
| logging.warning( | |
| "Severe cross-unit sentence echo removed | unit=%s count=%s max_similarity=%s", | |
| display_name, len(sentence_echo_rows), | |
| max(row["similarity"] for row in sentence_echo_rows), | |
| ) | |
| quality_diagnostics.append({ | |
| "scope":"manuscript_sentence_echo", "name":display_name, | |
| "status":"bounded_duplicate_removed", "count":len(sentence_echo_rows), | |
| "examples":sentence_echo_rows[:3], | |
| }) | |
| # Evidence Fact repair/validation is intentionally LAST. Once this | |
| # passes, no later operation may mutate this English unit. | |
| if mode == "asq": | |
| en_resolved, study_fact_rows = agent.repair_qualitative_study_fact_contract( | |
| en_resolved, data_metadata, section_name=display_name | |
| ) | |
| if study_fact_rows: | |
| logging.warning( | |
| "Qualitative Study Evidence Ledger repair | unit=%s count=%s types=%s", | |
| display_name, len(study_fact_rows), | |
| sorted({str(row.get("type") or "") for row in study_fact_rows}), | |
| ) | |
| quality_diagnostics.append({ | |
| "scope":"qualitative_study_fact_integrity", "name":display_name, | |
| "status":"deterministic_repair", "count":len(study_fact_rows), | |
| "examples":study_fact_rows[:3], | |
| }) | |
| agent.validate_qualitative_study_fact_contract( | |
| en_resolved, data_metadata, section_name=display_name | |
| ) | |
| elif mode == "amj": | |
| en_resolved, study_fact_rows = quant_engine.repair_quantitative_study_fact_contract( | |
| en_resolved, data_metadata, section_name=display_name | |
| ) | |
| if study_fact_rows: | |
| logging.warning( | |
| "Quantitative Evidence Ledger repair | unit=%s count=%s types=%s", | |
| display_name, len(study_fact_rows), | |
| sorted({str(row.get("type") or "") for row in study_fact_rows}), | |
| ) | |
| quality_diagnostics.append({ | |
| "scope":"quantitative_study_fact_integrity", "name":display_name, | |
| "status":"deterministic_repair", "count":len(study_fact_rows), | |
| "examples":study_fact_rows[:3], | |
| }) | |
| quant_engine.validate_quantitative_study_fact_contract( | |
| en_resolved, data_metadata, section_name=display_name | |
| ) | |
| # Recompute citation survival/claim demand from the exact text that | |
| # will be sealed, not from an earlier pre-cleanup intermediate. | |
| if not is_quarantined and section_id in self._citation_eligible_section_ids(mode): | |
| final_survived = set() | |
| for tag in set(survived or set()): | |
| marker = str((agent.citation_map.get(tag) or {}).get("in_text") or "").strip() | |
| if marker and marker in en_resolved: | |
| final_survived.add(tag) | |
| survived = final_survived | |
| citation_required_claim_counts_by_unit[display_name] = agent.count_external_literature_claims( | |
| en_resolved, section_name=display_name | |
| ) | |
| if assigned_tag_list: | |
| citation_validated_tags_by_unit[display_name] = sorted( | |
| survived, key=agent._citation_tag_number | |
| ) | |
| logging.info( | |
| "Top-level citation accumulation: unit=%s section=%s validated=%s external_claims=%s", | |
| display_name, citation_unit_section_map.get(display_name, section_id), | |
| len(survived), citation_required_claim_counts_by_unit[display_name], | |
| ) | |
| full_draft_en += en_resolved + "\n\n" | |
| if assigned_tag_list: | |
| logging.info( | |
| "Citation-first survival: section=%s planned=%s validated_new=%s", | |
| display_name, len(assigned_tag_list), len(survived), | |
| ) | |
| self._publish_preview(full_draft_en, display_name, force=True) | |
| # Only sealed/reconciled prose may feed subsequent manuscript memory. | |
| agent._extract_memory_from_text(en_resolved, sec_title) | |
| context_tail = " ".join(en_resolved.split()[-200:]) | |
| actual_words = self._english_count_for_target(en_resolved) | |
| if not bool(sec.get("exclude_from_word_budget")): | |
| accumulated_deviation += actual_words - adjusted_target | |
| self._upsert_quality_diagnostic( | |
| quality_diagnostics, self._section_word_diagnostic(display_name, actual_words, adjusted_target) | |
| ) | |
| completed_sections.append(display_name) | |
| section_key = str(sec.get('section_id') or sec_title) | |
| if self._section_is_complete(adaptive_structure, completed_sections, section_key): | |
| frozen_section = self._extract_top_level_section(full_draft_en, sec_title) | |
| if not frozen_section: | |
| raise RuntimeError(f"english_section_seal_source_missing:{sec_title}") | |
| section_integrity_seals[section_key] = self._build_section_integrity_seal( | |
| section_key, sec_title, frozen_section | |
| ) | |
| self._enqueue_translation_unit( | |
| translation_agent, | |
| unit_key=section_key, unit_name=sec_title, source_text=frozen_section, | |
| translation_sections=translation_sections, | |
| english_commit_text=full_draft_en, | |
| ) | |
| # Collect only already-finished target jobs. This call never waits, | |
| # so English Section N+1 can start while target Section N runs. | |
| translation_sections, translation_glossary = self._collect_translation_results( | |
| translation_sections, translation_glossary, block=False | |
| ) | |
| partial_translation = self._assemble_partial_translation(translation_sections, adaptive_structure, include_front=True) | |
| state.update( | |
| { | |
| "completed_sections": completed_sections, | |
| "full_draft_en": full_draft_en, | |
| "full_draft_translated": partial_translation, | |
| "translation_sections": translation_sections, | |
| "translation_glossary": translation_glossary, | |
| "context_tail": context_tail, | |
| "accumulated_deviation": accumulated_deviation, | |
| "quality_diagnostics": quality_diagnostics, | |
| "citation_plan": citation_plan, | |
| "citation_minimums": citation_minimums, | |
| "citation_reserve_targets": citation_reserve_targets, | |
| "citation_top_level_minimums": citation_top_level_minimums, | |
| "citation_top_level_reserve_targets": citation_top_level_reserve_targets, | |
| "citation_unit_section_map": citation_unit_section_map, | |
| "citation_required_claim_counts_by_unit": citation_required_claim_counts_by_unit, | |
| "citation_validated_tags_by_unit": citation_validated_tags_by_unit, | |
| "section_integrity_seals": section_integrity_seals, | |
| "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION, | |
| "agent_state": agent.export_state(), | |
| "data_metadata": data_metadata, | |
| } | |
| ) | |
| self._checkpoint( | |
| f"section_{len(completed_sections):03d}", state, pct, | |
| display_node=f"writing_complete:{len(completed_sections)}:{total_steps}", | |
| ) | |
| if not state.get("english_finalized"): | |
| self._progress(91, "english_validation") | |
| full_draft_en = _clean_trailing_references(full_draft_en, ["Reference", "References", "Reference List"]) | |
| section_order_for_fact = self._unique_top_level_sections(adaptive_structure) | |
| if not section_order_for_fact: | |
| raise RuntimeError("final_study_fact_section_plan_missing") | |
| # Final English validation is validation-only. Every mutable repair | |
| # was completed before the top-level section received its integrity | |
| # seal; finalization must never repair the whole manuscript again. | |
| for section_key, section_title in section_order_for_fact: | |
| sealed_section = self._extract_top_level_section(full_draft_en, section_title) | |
| if not sealed_section: | |
| raise RuntimeError(f"final_english_section_missing:{section_title}") | |
| seal = dict(section_integrity_seals.get(section_key) or {}) | |
| self._validate_section_integrity_seal(seal, sealed_section, section_title) | |
| try: | |
| if mode == "asq": | |
| agent.validate_qualitative_study_fact_contract( | |
| sealed_section, data_metadata, section_name=section_title | |
| ) | |
| elif mode == "amj": | |
| quant_engine.validate_quantitative_study_fact_contract( | |
| sealed_section, data_metadata, section_name=section_title | |
| ) | |
| except Exception as exc: | |
| prefix = ( | |
| "final_qualitative_study_fact_contract_failed" | |
| if mode == "asq" else "final_quantitative_study_fact_contract_failed" | |
| ) | |
| raise RuntimeError(f"{prefix}:{section_title}") from exc | |
| enum_checked, _enum_rows = agent.repair_and_validate_enumeration_contract( | |
| sealed_section, section_name=f"Final sealed section: {section_title}" | |
| ) | |
| if enum_checked.strip() != sealed_section.strip(): | |
| raise RuntimeError(f"final_enumeration_contract_failed:{section_title}") | |
| # Validate the completed body while the Working Abstract is still in | |
| # place. Final front matter must not be generated until all body | |
| # sections, citation cleanup, and any Discussion rebalance are frozen. | |
| agent.validate_manuscript_structure(full_draft_en, adaptive_structure, translated=False) | |
| if mode == "asq": | |
| self._validate_qualitative_quote_rendering( | |
| full_draft_en, quote_invocation_plan, data_metadata | |
| ) | |
| repeated_quotes = agent.find_repeated_qualitative_quotes(full_draft_en, data_metadata) | |
| if repeated_quotes: | |
| raise RuntimeError(f"Repeated qualitative quotations: {repeated_quotes[:5]}") | |
| # R4-R3: the body is already citation-resolved at this point. Never | |
| # rescan it for raw CITE tags; that destroys a valid citation ledger. | |
| final_citation_ledger = agent.finalize_resolved_citation_ledger(full_draft_en) | |
| references = agent.build_final_reference_lines(final_citation_ledger) | |
| if not references: | |
| self._record_evidence_coverage_event( | |
| state, [("final validated references", 0, 1)], | |
| phase="final_validated_references_empty", preview_status="limited", | |
| ) | |
| logging.warning( | |
| "Final validated reference list is empty; completing in evidence-limited mode without unrelated or invented references." | |
| ) | |
| validated_by_top_level = {} | |
| claims_by_top_level = {} | |
| for unit_name, section_id in citation_unit_section_map.items(): | |
| if section_id not in citation_top_level_minimums: | |
| continue | |
| validated_by_top_level.setdefault(section_id, set()).update( | |
| citation_validated_tags_by_unit.get(unit_name) or [] | |
| ) | |
| claims_by_top_level[section_id] = claims_by_top_level.get(section_id, 0) + int( | |
| citation_required_claim_counts_by_unit.get(unit_name, 0) or 0 | |
| ) | |
| method_section_id = 'methods' if mode == 'asq' else 'methodology' | |
| method_validated_tags = set(validated_by_top_level.get(method_section_id) or []) & set(final_citation_ledger) | |
| method_authority_tags = { | |
| tag for tag in method_validated_tags | |
| if 'method_authority' in set((agent.citation_map.get(tag) or {}).get('evidence_lanes') or []) | |
| } | |
| method_external_claims = max(0, int(claims_by_top_level.get(method_section_id, 0) or 0)) | |
| final_shortfalls = [] | |
| for section_id, planned_minimum in citation_top_level_minimums.items(): | |
| validated_tags = set(validated_by_top_level.get(section_id) or []) | |
| external_claims = max(0, int(claims_by_top_level.get(section_id, 0))) | |
| # Evidence authority is need-based, not a word-count quota. The | |
| # top-level planning minimum is therefore capped by the number of | |
| # externally-oriented claims that actually survived drafting. | |
| required = min(int(planned_minimum), external_claims) if external_claims else 0 | |
| if len(validated_tags) < required: | |
| final_shortfalls.append((section_id, len(validated_tags), required)) | |
| final_reference_count = len(references) | |
| if final_shortfalls or final_reference_count == 0: | |
| if final_shortfalls: | |
| self._record_evidence_coverage_event( | |
| state, final_shortfalls, phase="final_claim_evidence_gate", | |
| preview_status="limited", | |
| ) | |
| evidence_status = "limited" | |
| logging.warning( | |
| "Top-level claim-evidence final Gate completed with evidence warning: validated_references=%s sections=%s shortfalls=%s", | |
| final_reference_count, len(citation_top_level_minimums), final_shortfalls, | |
| ) | |
| else: | |
| evidence_status = "adequate" | |
| self._preview_evidence_status = "adequate" | |
| self._write_preview_payload(force=True) | |
| logging.info( | |
| "Top-level claim-evidence final Gate passed: validated_references=%s sections=%s", | |
| final_reference_count, len(citation_top_level_minimums), | |
| ) | |
| evidence_coverage_summary = self._evidence_coverage_summary( | |
| status=evidence_status, | |
| early_risk=bool(state.get("evidence_risk_detected")), | |
| final_reference_count=final_reference_count, | |
| minimums=citation_top_level_minimums, | |
| validated_by_top_level=validated_by_top_level, | |
| claims_by_top_level=claims_by_top_level, | |
| events=state.get("evidence_coverage_events") or self._evidence_coverage_events, | |
| method_authority_count=len(method_authority_tags), | |
| method_external_claims=method_external_claims, | |
| ) | |
| state["evidence_status"] = evidence_status | |
| state["evidence_coverage_summary"] = evidence_coverage_summary | |
| full_draft_en = full_draft_en.rstrip() + "\n\n# References\n" + "\n\n".join(references) | |
| actual_words = self._english_count_for_target(full_draft_en) | |
| self._upsert_quality_diagnostic( | |
| quality_diagnostics, self._validate_target_length(full_draft_en, target_words) | |
| ) | |
| body_only, _reference_tail = agent.validate_reference_tail( | |
| full_draft_en, minimum_references=final_reference_count | |
| ) | |
| agent.validate_manuscript_structure(body_only, adaptive_structure, translated=False) | |
| agent.enforce_resolved_claim_specific_citations(body_only) | |
| agent.validate_final_citation_bijection(body_only, _reference_tail, final_citation_ledger) | |
| # Body is now frozen. Generate both Final Abstract and Final Keywords | |
| # from the completed manuscript/evidence, then replace the provisional | |
| # front matter deterministically. No body rebalance may occur afterward. | |
| self._progress(92, "final_front_matter") | |
| final_front = agent.generate_final_front_matter( | |
| topic, keywords, body_only, data_evidence=user_data_md, mode=mode | |
| ) | |
| final_abstract = str(final_front.get("abstract") or "").strip() | |
| final_keywords = list(final_front.get("keywords") or []) | |
| if not final_abstract or not final_keywords: | |
| raise RuntimeError("final_front_matter_missing") | |
| final_abstract, final_front_fact_rows = self._repair_validate_study_fact_contract( | |
| mode, agent, quant_engine, final_abstract, data_metadata, | |
| section_name="Final Abstract", | |
| ) | |
| if not final_abstract: | |
| raise RuntimeError("final_abstract_empty_after_study_fact_contract") | |
| if final_front_fact_rows: | |
| logging.warning( | |
| "Final Abstract study-fact repair | mode=%s count=%s types=%s", | |
| mode, len(final_front_fact_rows), | |
| sorted({str(row.get("type") or "") for row in final_front_fact_rows}), | |
| ) | |
| quality_diagnostics.append({ | |
| "scope":"final_abstract_study_fact_integrity", "name":"Final Abstract", | |
| "status":"deterministic_repair", "count":len(final_front_fact_rows), | |
| "examples":final_front_fact_rows[:3], | |
| }) | |
| full_draft_en = replace_abstract(full_draft_en, final_abstract) | |
| full_draft_en = self._replace_keywords(full_draft_en, final_keywords) | |
| body_only, _reference_tail = agent.validate_reference_tail( | |
| full_draft_en, minimum_references=final_reference_count | |
| ) | |
| agent.validate_manuscript_structure(body_only, adaptive_structure, translated=False) | |
| agent.enforce_resolved_claim_specific_citations(body_only) | |
| post_front_ledger = agent.finalize_resolved_citation_ledger(body_only) | |
| if set(post_front_ledger) != set(final_citation_ledger): | |
| raise RuntimeError("final_front_matter_changed_citation_ledger") | |
| agent.validate_final_citation_bijection(body_only, _reference_tail, final_citation_ledger) | |
| self._upsert_quality_diagnostic( | |
| quality_diagnostics, self._validate_target_length(full_draft_en, target_words) | |
| ) | |
| # Final front matter has a different source hash from the Working | |
| # front matter. Retranslate only __front_matter__; unchanged body | |
| # translations are retained and reused. | |
| section_order_for_front = self._unique_top_level_sections(adaptive_structure) | |
| if section_order_for_front: | |
| final_front_source = self._extract_front_matter(body_only, section_order_for_front[0][1]) | |
| self._enqueue_translation_unit( | |
| translation_agent, | |
| unit_key='__front_matter__', unit_name='Final Front Matter', | |
| source_text=final_front_source, | |
| translation_sections=translation_sections, | |
| english_commit_text=full_draft_en, | |
| ) | |
| translation_sections, translation_glossary = self._collect_translation_results( | |
| translation_sections, translation_glossary, block=False | |
| ) | |
| partial_target = self._assemble_partial_translation( | |
| translation_sections, adaptive_structure, include_front=True | |
| ) | |
| self._publish_preview(full_draft_en, "english_complete", force=True, complete=True) | |
| state.update( | |
| { | |
| "full_draft_en": full_draft_en, | |
| "full_draft_translated": partial_target, | |
| "translation_sections": translation_sections, | |
| "translation_glossary": translation_glossary, | |
| "final_abstract": final_abstract, | |
| "final_keywords": final_keywords, | |
| "front_matter_phase": "final", | |
| "final_reference_count": final_reference_count, | |
| "final_citation_ledger": list(final_citation_ledger), | |
| "final_citation_ledger_version": FINAL_CITATION_LEDGER_VERSION, | |
| "evidence_status": str(state.get("evidence_status") or "adequate"), | |
| "evidence_risk_detected": bool(state.get("evidence_risk_detected")), | |
| "evidence_coverage_events": list(state.get("evidence_coverage_events") or []), | |
| "evidence_coverage_summary": dict(state.get("evidence_coverage_summary") or {}), | |
| "citation_plan": citation_plan, | |
| "citation_minimums": citation_minimums, | |
| "citation_reserve_targets": citation_reserve_targets, | |
| "citation_top_level_minimums": citation_top_level_minimums, | |
| "citation_top_level_reserve_targets": citation_top_level_reserve_targets, | |
| "citation_unit_section_map": citation_unit_section_map, | |
| "citation_required_claim_counts_by_unit": citation_required_claim_counts_by_unit, | |
| "citation_validated_tags_by_unit": citation_validated_tags_by_unit, | |
| "section_integrity_seals": section_integrity_seals, | |
| "english_section_seal_version": ENGLISH_SECTION_SEAL_VERSION, | |
| "quality_diagnostics": quality_diagnostics, | |
| "english_finalized": True, | |
| "translation_complete": False, | |
| "agent_state": agent.export_state(), | |
| } | |
| ) | |
| self._checkpoint("english_complete", state, 93) | |
| else: | |
| full_draft_en = str(state.get("full_draft_en") or "") | |
| full_draft_translated = str(state.get("full_draft_translated") or "") | |
| translation_sections = dict(state.get('translation_sections') or translation_sections or {}) | |
| translation_glossary = dict(state.get('translation_glossary') or translation_glossary or {}) | |
| if not state.get("translation_complete") or not full_draft_translated: | |
| self._progress(94, "translation") | |
| # English is fully frozen. Wait for all previously queued section | |
| # translations before the strict final gate; checkpoint ownership | |
| # remains on this main execution thread. | |
| translation_sections, translation_glossary = self._collect_translation_results( | |
| translation_sections, translation_glossary, block=True | |
| ) | |
| self._shutdown_translation_worker(wait=True) | |
| body_only, reference_tail = agent.validate_reference_tail( | |
| full_draft_en, minimum_references=max(0, int(state.get("final_reference_count") or 0)) | |
| ) | |
| section_order = self._unique_top_level_sections(adaptive_structure) | |
| if not section_order: | |
| raise RuntimeError('translation_section_plan_missing') | |
| front_matter = self._extract_front_matter(body_only, section_order[0][1]) | |
| final_units = [('__front_matter__', 'Front Matter', front_matter)] | |
| for key, title in section_order: | |
| final_units.append((key, title, self._extract_top_level_section(body_only, title))) | |
| for index, (unit_key, unit_name, source_text) in enumerate(final_units, start=1): | |
| translation_sections, translation_glossary = self._translate_frozen_unit( | |
| translation_agent, unit_key=unit_key, unit_name=unit_name, source_text=source_text, | |
| target_lang=target_lang, translation_sections=translation_sections, | |
| translation_glossary=translation_glossary, adaptive_structure=adaptive_structure, | |
| include_front=True, defer_failures=False, | |
| english_commit_text=full_draft_en, final_gate=True, | |
| ) | |
| partial = self._assemble_partial_translation( | |
| translation_sections, adaptive_structure, include_front=True | |
| ) | |
| state.update({ | |
| 'full_draft_en': full_draft_en, | |
| 'full_draft_translated': partial, | |
| 'translation_sections': translation_sections, | |
| 'translation_glossary': translation_glossary, | |
| 'agent_state': agent.export_state(), | |
| }) | |
| pct = min(96, 94 + int(index * 2 / max(1, len(final_units)))) | |
| current_translation = dict(translation_sections.get(unit_key) or {}) | |
| blocking = ( | |
| current_translation.get('qa_status') not in {'passed', 'passed_with_warning'} | |
| or not current_translation.get('translated_text') | |
| ) | |
| if blocking: | |
| state['translation_blocking_unit'] = str(unit_key) | |
| state['translation_blocking_unit_name'] = str(unit_name) | |
| state['translation_blocking_error_code'] = str(current_translation.get('error_code') or 'TRANSLATION_PENDING') | |
| self._checkpoint( | |
| f'translation_{unit_key}_repair_pending', state, pct, | |
| display_node=f'translation_repair_pending:{index}:{len(final_units)}', | |
| ) | |
| # Stop at the first blocking unit. The durable claim state is | |
| # now checkpointed, so resume continues exactly here instead | |
| # of spending tokens on later units or repeating prior repair. | |
| raise RuntimeError('translation_final_gate_unresolved:' + str(unit_key)) | |
| state.pop('translation_blocking_unit', None) | |
| state.pop('translation_blocking_unit_name', None) | |
| state.pop('translation_blocking_error_code', None) | |
| self._checkpoint( | |
| f'translation_{unit_key}', state, pct, | |
| display_node=f'translation_final_gate:{index}:{len(final_units)}', | |
| ) | |
| unresolved_units = [ | |
| key for key, _name, _source in final_units | |
| if (translation_sections.get(key) or {}).get('qa_status') not in {'passed', 'passed_with_warning'} | |
| or not (translation_sections.get(key) or {}).get('translated_text') | |
| ] | |
| if unresolved_units: | |
| raise RuntimeError('translation_final_gate_unresolved:' + ','.join(unresolved_units[:8])) | |
| full_draft_translated = self._assemble_partial_translation( | |
| translation_sections, adaptive_structure, include_front=True | |
| ).rstrip() | |
| full_draft_translated += ( | |
| f"\n\n# {agent._localized_reference_heading(target_lang)}\n" + reference_tail.strip() | |
| ) | |
| full_draft_translated = _normalize_target_manuscript_typography( | |
| full_draft_translated, target_lang | |
| ) | |
| agent.validate_translation_alignment(full_draft_en, full_draft_translated) | |
| agent.validate_reference_preservation(full_draft_en, full_draft_translated, target_lang) | |
| self._publish_target_preview( | |
| full_draft_translated, 'translation_complete', force=True, complete=True | |
| ) | |
| state.update( | |
| { | |
| "full_draft_translated": full_draft_translated, | |
| "translation_sections": translation_sections, | |
| "translation_glossary": translation_glossary, | |
| "translation_complete": True, | |
| "agent_state": agent.export_state(), | |
| } | |
| ) | |
| self._checkpoint("translation_complete", state, 96) | |
| normalized_target = _normalize_target_manuscript_typography( | |
| full_draft_translated, target_lang | |
| ) | |
| if normalized_target != full_draft_translated: | |
| full_draft_translated = normalized_target | |
| state["full_draft_translated"] = full_draft_translated | |
| existing_output = str(state.get("output_filename") or "").strip() | |
| if state.get("package_validated") and existing_output: | |
| existing_path = self.output_dir / os.path.basename(existing_output) | |
| if existing_path.is_file() and existing_path.stat().st_size > 0: | |
| self._progress(99, "package_validated") | |
| result = {"output_filename": existing_path.name, "output_path": str(existing_path)} | |
| if str(state.get("evidence_status") or "") == "limited": | |
| result.update({ | |
| "quality_status": "evidence_limited", | |
| "safe_error_code": "W_EVIDENCE_COVERAGE_LIMITED", | |
| "safe_error_detail": self._evidence_warning_detail(), | |
| }) | |
| return result | |
| self._progress(97, "packaging") | |
| bj_time = datetime.now(timezone(timedelta(hours=8))).strftime("%Y%m%d_%H%M") | |
| package = build_validated_package( | |
| english_markdown=full_draft_en, | |
| translated_markdown=full_draft_translated, | |
| english_filename=f"Research_EN_{bj_time}.docx", | |
| translated_filename=f"Research_{output_suffix}_{bj_time}.docx", | |
| data_metadata=data_metadata, | |
| mode=mode, | |
| citation_audit=agent.citation_audit_log, | |
| literature_metrics=agent.get_source_metrics(), | |
| quality_diagnostics=quality_diagnostics, | |
| evidence_coverage=state.get("evidence_coverage_summary") or { | |
| "status": str(state.get("evidence_status") or "adequate"), | |
| "early_risk_detected": bool(state.get("evidence_risk_detected")), | |
| "final_reference_count": int(state.get("final_reference_count") or 0), | |
| "sections": [], | |
| "events": list(state.get("evidence_coverage_events") or []), | |
| }, | |
| ) | |
| self._progress(98, "package_validation") | |
| output_filename = f"SmartPaper_{bj_time}.zip" | |
| pending = self.output_dir / (".pending_" + output_filename) | |
| pending.write_bytes(package.zip_bytes) | |
| final = self.output_dir / output_filename | |
| os.replace(pending, final) | |
| state.update( | |
| { | |
| "output_filename": output_filename, | |
| "package_validated": True, | |
| "evidence_status": str(state.get("evidence_status") or "adequate"), | |
| "evidence_coverage_summary": dict(state.get("evidence_coverage_summary") or {}), | |
| "agent_state": agent.export_state(), | |
| } | |
| ) | |
| self._checkpoint("package_validated", state, 99, display_node="package_validated") | |
| result = {"output_filename": output_filename, "output_path": str(final)} | |
| if str(state.get("evidence_status") or "") == "limited": | |
| result.update({ | |
| "quality_status": "evidence_limited", | |
| "safe_error_code": "W_EVIDENCE_COVERAGE_LIMITED", | |
| "safe_error_detail": self._evidence_warning_detail(), | |
| }) | |
| return result | |
| def run(self) -> dict[str, Any]: | |
| try: | |
| return self._run_impl() | |
| finally: | |
| self._shutdown_translation_worker(wait=False) | |
| def _progress_activity(self) -> None: | |
| # Activity remains Executor-owned. We also snapshot actual request | |
| # counters so the next Executor heartbeat can persist live UI metrics. | |
| ds_calls, s2_calls, oa_calls = self._compute_request_counts() | |
| try: | |
| self.progress_callback( | |
| activity_only=True, | |
| ds_calls=ds_calls, | |
| s2_calls=s2_calls, | |
| oa_calls=oa_calls, | |
| ) | |
| except TypeError: | |
| pass | |