Spaces:
Running
Running
R3.2.8.6-R1.3C-KK1-R6A-R3 Qualitative Appendix Citation Relevance and Quote Output Integrity Final Fix
9c60e2c verified | # -*- 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] | |
| MIN_FINAL_REFERENCES = 50 | |
| LITERATURE_EVIDENCE_TARGET = 75 | |
| CITATION_PLAN_TARGET = 70 | |
| TRANSLATION_CONTRACT_VERSION = 'hierarchical-multilingual-v3' | |
| 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() | |
| 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, | |
| 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.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) | |
| 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 = {} | |
| self._qualitative_quote_sources = {} | |
| 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) -> None: | |
| seq, normalized_stage = self.checkpoints.save(stage, state) | |
| self._progress(pct, 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 _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 _prepare_qualitative_translation_source(cls, source, quote_sources, quote_ledger): | |
| """Use one task-level target rendering for every frozen qualitative quote. | |
| The first translated occurrence keeps the English quote in context and | |
| appends its protected QUOTE tag so the target rendering can be captured. | |
| Later occurrences (notably the Appendix) send only the protected tag and | |
| are restored from the already QA-passed task-level quote ledger. | |
| """ | |
| 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: | |
| if tag in (quote_ledger or {}): | |
| value = value.replace(tagged, f"[{tag}]") | |
| reused.append(tag) | |
| else: | |
| pending.append(tag) | |
| continue | |
| if wrapped not in value: | |
| continue | |
| if tag in (quote_ledger or {}): | |
| value = value.replace(wrapped, f"[{tag}]") | |
| reused.append(tag) | |
| else: | |
| value = value.replace(wrapped, tagged) | |
| pending.append(tag) | |
| return value, pending, reused | |
| def _finalize_qualitative_translation_quotes(cls, translated, pending, reused, quote_ledger): | |
| value = str(translated or "") | |
| ledger = dict(quote_ledger or {}) | |
| quote_pattern = r'(?P<quote>"[^"\n]{4,2500}"|“[^”\n]{4,2500}”|«[^»\n]{4,2500}»|„[^“\n]{4,2500}“)' | |
| for tag in pending or []: | |
| pattern = re.compile( | |
| quote_pattern + rf"\s*(?<!\w)\[?{re.escape(tag)}\]?(?!\w)", | |
| flags=re.UNICODE, | |
| ) | |
| matches = list(pattern.finditer(value)) | |
| if len(matches) != 1: | |
| raise RuntimeError( | |
| f"Qualitative quote translation ledger capture failed: {tag} matches={len(matches)}" | |
| ) | |
| rendered_quote = matches[0].group("quote") | |
| ledger[tag] = rendered_quote | |
| value = value[:matches[0].start()] + rendered_quote + value[matches[0].end():] | |
| for tag in reused or []: | |
| 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 translation ledger reuse 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 _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 | |
| ), | |
| "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 _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 _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 _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 ''), | |
| } | |
| self._translation_worker = _OrderedTranslationWorker(self._run_translation_job) | |
| 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 | |
| source_hash = self._text_hash(source) | |
| fact_ledger_hash = self._translation_ledger_hash(agent, source) | |
| existing = dict((translation_sections or {}).get(unit_key) or {}) | |
| if ( | |
| existing.get('qa_status') in {'passed', 'passed_with_warning'} | |
| and existing.get('source_hash') == 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('translated_text') | |
| ): | |
| return False | |
| pending_key = (str(unit_key), source_hash, fact_ledger_hash) | |
| if pending_key in self._translation_worker_pending: | |
| return False | |
| future = self._translation_worker.submit({ | |
| 'unit_key': str(unit_key), | |
| 'unit_name': str(unit_name), | |
| 'source_text': source, | |
| 'source_hash': 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, | |
| }) | |
| 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_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, | |
| ): | |
| source = str(source_text or '').strip() | |
| if not source: | |
| return translation_sections, translation_glossary | |
| source_hash = self._text_hash(source) | |
| contract_version = TRANSLATION_CONTRACT_VERSION | |
| fact_ledger_hash = self._translation_ledger_hash(agent, source) | |
| existing = dict((translation_sections or {}).get(unit_key) or {}) | |
| if ( | |
| existing.get('qa_status') in {'passed', 'passed_with_warning'} | |
| and existing.get('source_hash') == 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('translated_text') | |
| ): | |
| return translation_sections, translation_glossary | |
| partial = self._assemble_partial_translation(translation_sections, adaptive_structure, include_front=include_front) | |
| self._publish_target_preview( | |
| partial, f'translating:{unit_name}', force=True, complete=False, | |
| english_commit_text=english_commit_text, | |
| ) | |
| updated_sections = dict(translation_sections or {}) | |
| quote_ledger_record = dict(updated_sections.get("__quote_ledger__") or {}) | |
| quote_ledger = ( | |
| dict(quote_ledger_record.get("translations") or {}) | |
| if quote_ledger_record.get("target_lang") in {None, "", target_lang} | |
| else {} | |
| ) | |
| translation_source, pending_quote_tags, reused_quote_tags = self._prepare_qualitative_translation_source( | |
| source, getattr(self, "_qualitative_quote_sources", {}), quote_ledger | |
| ) | |
| def publish_translation_stage(stage): | |
| value = str(stage or '').strip().casefold() | |
| if value in {'qa', 'repairing'}: | |
| preview_stage = f'translation_qa:{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: | |
| translate_method = agent.translate_frozen_section | |
| call_kwargs = { | |
| 'glossary': translation_glossary, 'section_name': unit_name, 'max_attempts': 2, | |
| } | |
| try: | |
| parameters = 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 | |
| ) | |
| except (TypeError, ValueError): | |
| supports_stage_callback = False | |
| if supports_stage_callback: | |
| call_kwargs['stage_callback'] = publish_translation_stage | |
| 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__'] = { | |
| 'target_lang': target_lang, | |
| 'translations': dict(quote_ledger), | |
| 'updated_at': datetime.now(timezone.utc).isoformat(), | |
| } | |
| except PipelineCancelledError: | |
| raise | |
| except Exception as error: | |
| if not defer_failures: | |
| raise | |
| 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, | |
| 'target_lang': target_lang, | |
| 'glossary_version': self._glossary_version(translation_glossary), | |
| 'translation_contract_version': contract_version, | |
| 'fact_ledger_hash': fact_ledger_hash, | |
| 'translated_text': '', | |
| 'qa_status': 'repair_pending', | |
| 'error_code': error_code, | |
| 'attempts': int(existing.get('attempts') or 0) + 1, | |
| 'deferred_at': datetime.now(timezone.utc).isoformat(), | |
| } | |
| logging.warning( | |
| 'Translation deferred until final gate | unit=%s code=%s error=%s', | |
| unit_name, error_code, type(error).__name__, | |
| ) | |
| 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 {}) | |
| updated_sections[unit_key] = { | |
| 'source_hash': 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': str(result.get('fact_ledger_hash') or fact_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:], | |
| '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._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 _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) | |
| state = self.checkpoints.load_latest() or {} | |
| quality_diagnostics = list(state.get("quality_diagnostics") or []) | |
| 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"]) | |
| 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 "" | |
| user_data_md, _, data_metadata = qual_engine.resolve_qualitative_data(topic, keywords, raw_interviews) | |
| 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) | |
| 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") | |
| refs_text = agent.fetch_references_parallel( | |
| topic, | |
| keywords, | |
| min_required=LITERATURE_EVIDENCE_TARGET, | |
| progress_callback=lambda: self._progress(9, "S2 + OpenAlex"), | |
| ) | |
| if len(agent.citation_map) < MIN_FINAL_REFERENCES: | |
| raise RuntimeError( | |
| f"Insufficient eligible references: {len(agent.citation_map)}/{MIN_FINAL_REFERENCES}" | |
| ) | |
| 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") | |
| synthesis_result = self._stream_text( | |
| agent.stream_synthesize_literature(topic, refs_text, user_core_literature=user_lit_context) | |
| ) | |
| 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") | |
| research_logic_response = self._stream_text( | |
| agent.stream_generate_research_logic( | |
| topic, keywords, synthesis_result, refs_text, | |
| data_evidence=planning_data_md, mode=mode, | |
| ) | |
| ) | |
| 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") | |
| 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")) | |
| 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_plan = agent.build_claim_evidence_plan( | |
| adaptive_structure, topic, synthesis_result, mode=mode, target_unique=CITATION_PLAN_TARGET | |
| ) | |
| planned_unique = { | |
| item.get("tag") | |
| for items in citation_plan.values() | |
| for item in items | |
| if item.get("tag") | |
| } | |
| if len(planned_unique) < MIN_FINAL_REFERENCES: | |
| raise RuntimeError( | |
| "Insufficient claim-evidence plan before drafting: " | |
| f"{len(planned_unique)}/{MIN_FINAL_REFERENCES}." | |
| ) | |
| state.update( | |
| { | |
| "adaptive_structure": adaptive_structure, | |
| "citation_plan": citation_plan, | |
| "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": {}, | |
| "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 {} | |
| translation_sections = dict(state.get('translation_sections') or {}) | |
| translation_glossary = dict(state.get('translation_glossary') or {}) | |
| working_abstract = str(state.get("working_abstract") or "").strip() | |
| working_keywords = list(state.get("working_keywords") or []) | |
| 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 '') | |
| # 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()) | |
| 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") | |
| 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=CITATION_PLAN_TARGET | |
| ) | |
| state["citation_plan"] = citation_plan | |
| quote_ownership = ( | |
| self._build_qualitative_quote_ownership(adaptive_structure, data_metadata) | |
| if mode == "asq" | |
| else {} | |
| ) | |
| 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. Every listed item has already been assigned to this unit on evidence relevance: use each assigned item exactly once unless it would contradict the frozen research evidence. " | |
| "For each item, write one distinct 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 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_labels = [] | |
| for item in theme_evidence: | |
| tag = str(item.get("tag") or "").strip() | |
| 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() | |
| if tag: | |
| evidence_labels.append(f"{tag} [stage={stage}; case={case_label}; role={role}]") | |
| qualitative_rule = ( | |
| f"Assigned qualitative concepts: {concepts}. Assigned quotation tags: {', '.join(evidence_labels)}. " | |
| "Use only these assigned quotation tags in this Findings unit and do not repeat a tag used elsewhere. " | |
| "When invoking a quotation, output only its [QUOTE-XX] tag at the exact position; do not reproduce or paraphrase the quotation because the system inserts the frozen verbatim text. " | |
| "Preserve any frozen stage/case/deviant role and use deviant evidence to qualify rather than erase the pattern." | |
| if concepts or evidence_labels 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), | |
| ) | |
| 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 "" | |
| ) | |
| 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} | |
| <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, display_name) | |
| prefix = "" | |
| if sec.get("section_first"): | |
| prefix += f"# {sec_title}\n\n" | |
| if sec.get("subsection_first") and subsection: | |
| prefix += f"## {subsection}\n\n" | |
| 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) | |
| # Bind only the current Findings unit's frozen quotation ownership. | |
| # Other ASQ units cannot see or reproduce verbatim quote text. | |
| if mode == "asq" and "appendix" not in sec_title.lower(): | |
| en_raw = self._bind_owned_qualitative_quotes( | |
| en_raw, | |
| display_name=display_name, | |
| data_metadata=data_metadata, | |
| quote_ownership=quote_ownership, | |
| prior_manuscript=full_draft_en, | |
| ) | |
| 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 | |
| ) | |
| 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) * 0.78))), | |
| ) | |
| 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}, | |
| ) | |
| ) | |
| supplemental_raw = agent.build_citation_recovery_paragraph( | |
| recovery_items[:needed], section_name=display_name, section_context=en_resolved | |
| ) | |
| if supplemental_raw: | |
| supplemental_resolved = agent.audit_and_resolve_citations( | |
| supplemental_raw, | |
| section_name=f"{display_name} — citation recovery", | |
| quarantine=False, | |
| ) | |
| if supplemental_resolved.strip(): | |
| en_resolved = en_resolved.rstrip() + "\n\n" + supplemental_resolved.strip() | |
| used_after = set(agent.used_citations) | |
| survived = used_after - used_before | |
| 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) | |
| agent._extract_memory_from_text(en_raw, sec_title) | |
| context_tail = " ".join(en_raw.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) | |
| 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, | |
| "agent_state": agent.export_state(), | |
| "data_metadata": data_metadata, | |
| } | |
| ) | |
| self._checkpoint(f"section_{len(completed_sections):03d}", state, pct) | |
| if not state.get("english_finalized"): | |
| self._progress(91, "english_validation") | |
| full_draft_en = _clean_trailing_references(full_draft_en, ["Reference", "References", "Reference List"]) | |
| # 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": | |
| repeated_quotes = agent.find_repeated_qualitative_quotes(full_draft_en, data_metadata) | |
| if repeated_quotes: | |
| raise RuntimeError(f"Repeated qualitative quotations: {repeated_quotes[:5]}") | |
| references = [] | |
| seen_refs = set() | |
| for tag in sorted(agent.used_citations, key=agent._citation_tag_number): | |
| ref = str((agent.citation_map.get(tag) or {}).get("ref") or "").strip() | |
| if ref and ref not in seen_refs: | |
| references.append(f"- {ref}") | |
| seen_refs.add(ref) | |
| if len(references) < MIN_FINAL_REFERENCES: | |
| raise RuntimeError( | |
| "Insufficient final cited references after semantic validation: " | |
| f"{len(references)}/{MIN_FINAL_REFERENCES}." | |
| ) | |
| 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=MIN_FINAL_REFERENCES | |
| ) | |
| agent.validate_manuscript_structure(body_only, adaptive_structure, translated=False) | |
| agent.enforce_resolved_claim_specific_citations(body_only) | |
| # 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") | |
| 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=MIN_FINAL_REFERENCES | |
| ) | |
| agent.validate_manuscript_structure(body_only, adaptive_structure, translated=False) | |
| agent.enforce_resolved_claim_specific_citations(body_only) | |
| 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", | |
| "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=MIN_FINAL_REFERENCES | |
| ) | |
| 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, | |
| ) | |
| 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)))) | |
| self._checkpoint(f'translation_{unit_key}', state, pct) | |
| 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() | |
| ) | |
| agent.validate_translation_alignment(full_draft_en, full_draft_translated) | |
| 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) | |
| 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") | |
| return {"output_filename": existing_path.name, "output_path": str(existing_path)} | |
| 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, | |
| ) | |
| 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, | |
| "agent_state": agent.export_state(), | |
| } | |
| ) | |
| self._checkpoint("package_validated", state, 99) | |
| return {"output_filename": output_filename, "output_path": str(final)} | |
| 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 | |