# -*- coding: utf-8 -*- """SmartPaper R3-Free UI: Streamlit is an observer, never the paper executor.""" import hashlib import html import json import logging import os import re import secrets import shutil import subprocess import sys import time import textwrap from datetime import datetime, timezone from pathlib import Path import streamlit as st import streamlit.components.v1 as components from config import APP_CONFIG from managed_auth import ( cancel_generation_task, generation_task_status, get_product_ui_projection, recover_generation_task, require_alipay_login, render_logged_in_account, render_payment_purchase, start_generation_task, ) from i18n import ( LANGUAGE_OPTIONS, get_amj_csv_template, get_asq_txt_template, get_output_suffix, get_target_flag, get_target_language, tr, ) from smartpaper_restore_bridge import edge_bridge st.set_page_config( page_title="Article Agent", page_icon="📊", layout="wide", initial_sidebar_state="collapsed", ) # A3-R5D: default-off, benchmark-only startup probe; never blocks the UI. try: from core.a3_glm_live_probe import run_startup_probe_if_enabled run_startup_probe_if_enabled() except Exception as _a3_probe_exc: logging.error("[A3_R5D_GLM_PROBE] bootstrap_failed error_class=%s", _a3_probe_exc.__class__.__name__) # A3-R5H: default-off Gemini Free-Tier startup probe; never blocks the UI. try: from core.a3_gemini_live_probe import run_startup_probe_if_enabled as run_gemini_startup_probe_if_enabled run_gemini_startup_probe_if_enabled() except Exception as _a3_gemini_probe_exc: logging.error("[A3_R5H_GEMINI_PROBE] bootstrap_failed error_class=%s", _a3_gemini_probe_exc.__class__.__name__) WORKSPACE_DIR = APP_CONFIG.get("workspace_dir", ".workspace") os.makedirs(WORKSPACE_DIR, exist_ok=True) # The display-only preview cache must exist from the first Streamlit script run. # Runtime preview files remain ephemeral and carry no task/execution authority. (Path(__file__).resolve().parent / "static" / "previews").mkdir(parents=True, exist_ok=True) os.environ["STREAMLIT_SERVER_ENABLE_CORS"] = "false" os.environ["STREAMLIT_SERVER_ENABLE_XSRF_PROTECTION"] = "false" WECHAT_BLOCKER_JS = """""" components.html(WECHAT_BLOCKER_JS, height=0, width=0) def inject_file_uploader_i18n(language: str) -> None: """Localize uploaders by semantic label, never by visual/tab position.""" payload = { "browse": tr(language, "browse_files"), "uploaders": { tr(language, "upload_reference"): { "title": tr(language, "drop_docs"), "limit": tr(language, "limit_docs"), }, tr(language, "upload_interviews"): { "title": tr(language, "drop_interviews"), "limit": tr(language, "limit_interviews"), }, tr(language, "upload_stats"): { "title": tr(language, "drop_data"), "limit": tr(language, "limit_data"), }, }, } script = f""" """ components.html(script, height=0, width=0) def inject_wechat_browser_notice() -> None: """Warn immediately when SmartPaper is opened inside WeChat or WeCom.""" script = """ """ components.html(script, height=0, width=0) for key, default in { "is_running": False, "zip_buffer": None, "zip_filename": "", "run_mode": "asq", "sp_generation_task": None, "sp_pending_launch": None, "sp_pending_resume_task_id": "", "sp_pending_cancel_task_id": "", "sp_payment_required": False, "sp_terminal_notice": None, "sp_task_recovery_initialized": False, "sp_last_ui_task_status": "", "sp_scroll_to_generation_action": False, "sp_completed_download_scroll_task_id": "", }.items(): if key not in st.session_state: st.session_state[key] = default def _safe_key(value: str) -> str: cleaned = re.sub(r"[^A-Za-z0-9_.-]", "_", str(value or "").strip()) return cleaned[:100] or "anonymous" def _new_generation_idempotency_key() -> str: return f"gen_{int(time.time() * 1000)}_{secrets.token_urlsafe(18)}" def _atomic_json(path: Path, value: dict) -> None: path.parent.mkdir(parents=True, exist_ok=True) temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_text(json.dumps(value, ensure_ascii=False), encoding="utf-8") os.replace(temporary, path) def _canonical_request_hash(value: dict) -> str: material = dict(value 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 _task_dir(task_id: str) -> Path: return Path(_user_workspace) / "tasks" / _safe_key(task_id) def _local_resume_capability(task: dict) -> tuple[bool, str]: """Return whether this HF runtime has the local material required to resume. Worker/D1 remains authoritative for the remote task state. This function answers only the local capability question and never changes task/quota state. """ task = dict(task or {}) task_id = str(task.get("task_id") or "").strip() if not task_id: return False, "task_id_missing" task_dir = _task_dir(task_id) request_path = task_dir / "request.json" if not request_path.is_file(): return False, "request_missing" try: request_payload = json.loads(request_path.read_text(encoding="utf-8")) except Exception: return False, "request_unreadable" if not isinstance(request_payload, dict): return False, "request_invalid" if str(request_payload.get("task_id") or "").strip() != task_id: return False, "request_task_mismatch" request_hash = str(request_payload.get("request_hash") or "").strip().lower() if not re.fullmatch(r"[a-f0-9]{64}", request_hash): return False, "request_hash_invalid" if request_hash != _canonical_request_hash(request_payload): return False, "request_hash_mismatch" task_root = task_dir.resolve() for field in ("reference_files", "data_files", "interview_files"): items = request_payload.get(field) or [] if not isinstance(items, list): return False, f"{field}_invalid" for item in items: local_path = (task_dir / str(item)).resolve() if task_root not in local_path.parents or not local_path.is_file(): return False, "input_missing" try: checkpoint_seq = int(task.get("checkpoint_seq") or 0) except (TypeError, ValueError): return False, "checkpoint_seq_invalid" if checkpoint_seq <= 0: return True, "ready" expected_stage = str(task.get("checkpoint_stage") or "").strip() try: candidates = sorted((task_dir / "checkpoints").glob(f"checkpoint_{checkpoint_seq:05d}_*.json")) except Exception: candidates = [] for candidate in candidates: try: payload = json.loads(candidate.read_text(encoding="utf-8")) if not isinstance(payload, dict): continue if str(payload.get("request_hash") or "").strip().lower() != request_hash: continue if int(payload.get("checkpoint_seq") or 0) != checkpoint_seq: continue if expected_stage and str(payload.get("checkpoint_stage") or "").strip() != expected_stage: continue return True, "ready" except Exception: continue return False, "checkpoint_missing" def _task_preview_path(task_id: str) -> str: """Return the unguessable local static preview URL for this task, if any.""" request_path = _task_dir(task_id) / "request.json" if not request_path.is_file(): return "" try: payload = json.loads(request_path.read_text(encoding="utf-8")) token = str(payload.get("preview_token") or "").strip() if not re.fullmatch(r"[A-Za-z0-9_-]{32,128}", token): return "" return f"/app/static/previews/{token}.json" except Exception: return "" def _cleanup_task_artifacts(task_id: str) -> None: """Delete local stage artifacts only after authoritative cancellation.""" task_dir = _task_dir(task_id) token = "" request_path = task_dir / "request.json" if request_path.is_file(): try: payload = json.loads(request_path.read_text(encoding="utf-8")) token = str(payload.get("preview_token") or "").strip() except Exception: token = "" if re.fullmatch(r"[A-Za-z0-9_-]{32,128}", token): try: (Path(__file__).resolve().parent / "static" / "previews" / f"{token}.json").unlink(missing_ok=True) except Exception as exc: logging.warning("[SP_UI] cancel_preview_cleanup_error task_id=%s error=%s", task_id, type(exc).__name__) shutil.rmtree(task_dir, ignore_errors=True) def _task_mode(task: dict) -> str: mode = str((task or {}).get("mode") or st.session_state.get("run_mode") or "asq").strip().lower() return "amj" if mode == "amj" else "asq" def _task_output_path(task: dict) -> Path | None: task_id = str(task.get("task_id") or "").strip() filename = os.path.basename(str(task.get("output_filename") or "").strip()) if not task_id or not filename: return None return _task_dir(task_id) / "outputs" / filename def _save_uploaded(items, inputs_dir: Path, prefix: str) -> list[str]: saved = [] for index, item in enumerate(items or []): if item is None: continue original = os.path.basename(str(getattr(item, "name", f"input_{index}"))) safe_name = re.sub(r"[^A-Za-z0-9_.-]", "_", original)[:140] or f"input_{index}" filename = f"{prefix}_{index:02d}_{safe_name}" path = inputs_dir / filename data = item.getvalue() if hasattr(item, "getvalue") else item.read() path.write_bytes(bytes(data or b"")) saved.append(str(Path("inputs") / filename)) try: item.seek(0) except Exception: pass return saved def _persist_task_request(task: dict, mode: str, lang_code: str) -> dict: task_id = str(task.get("task_id") or "").strip() numeric_id = (st.session_state.get("sp_user") or {}).get("numeric_id") if not task_id or not isinstance(numeric_id, int) or numeric_id <= 0: raise RuntimeError("task_identity_missing") task_dir = _task_dir(task_id) inputs_dir = task_dir / "inputs" inputs_dir.mkdir(parents=True, exist_ok=True) if mode == "amj": refs = _save_uploaded(st.session_state.get("up_amj_ref") or [], inputs_dir, "ref") data = _save_uploaded([st.session_state.get("up_amj_data")], inputs_dir, "data") interviews = [] else: refs = _save_uploaded(st.session_state.get("up_asq_ref") or [], inputs_dir, "ref") data = [] interviews = _save_uploaded([st.session_state.get("up_asq_data")], inputs_dir, "interview") request_data = { "task_id": task_id, "user_id": numeric_id, "mode": mode, "interface_language": lang_code, "target_lang": get_target_language(lang_code), "output_suffix": get_output_suffix(lang_code), "topic": str(st.session_state.get("sp_topic") or DEFAULT_TOPIC), "keywords": str(st.session_state.get("sp_keywords") or DEFAULT_KEYWORDS), "target_words": int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]), # High-entropy, display-only token. It is part of the immutable task # request/hash, but it grants no Worker/D1 authority. "preview_token": secrets.token_urlsafe(32), "reference_files": refs, "data_files": data, "interview_files": interviews, "created_at": datetime.now(timezone.utc).isoformat(), } request_data["request_hash"] = _canonical_request_hash(request_data) _atomic_json(task_dir / "request.json", request_data) return request_data def _pid_alive(pid: int) -> bool: if not isinstance(pid, int) or pid <= 0: return False try: os.kill(pid, 0) return True except OSError: return False def _pid_is_task_executor(pid: int, task_id: str, task_dir: Path) -> bool: """Best-effort Linux identity check before trusting/killing a launch PID. Hugging Face Spaces run on Linux. A stale PID can be reused, so a bare ``os.kill(pid, 0)`` is never enough to identify our executor. """ if not _pid_alive(pid): return False try: raw = (Path("/proc") / str(pid) / "cmdline").read_bytes() command = raw.replace(b"\x00", b" ").decode("utf-8", errors="ignore") except Exception: return False return ( "task_executor.py" in command and f"--task-id {task_id}" in command and str(task_dir.resolve()) in command ) def _launch_executor(task_id: str, *, force: bool = False) -> int: """Start a detached executor without making Streamlit its lifecycle owner. ``subprocess.Popen`` only proves that the OS accepted the child process. R3.2 therefore records the PID and returns immediately. The executor owns claim/heartbeat/checkpoint/finalization; Worker/D1 owns task authority and startup-lease expiry. Streamlit never waits for claim and never terminates an executor because a UI-side timer expired. """ task_dir = _task_dir(task_id) request_path = task_dir / "request.json" if not request_path.is_file(): raise RuntimeError("task_request_manifest_missing") request_data = json.loads(request_path.read_text(encoding="utf-8")) numeric_id = int(request_data.get("user_id") or 0) request_hash = str(request_data.get("request_hash") or "").strip() if numeric_id <= 0 or not request_hash: raise RuntimeError("task_request_identity_invalid") launch_marker_path = task_dir / "executor_launch.json" claim_marker_path = task_dir / "executor_claim.json" if launch_marker_path.is_file(): try: marker = json.loads(launch_marker_path.read_text(encoding="utf-8")) pid = int(marker.get("pid") or 0) if _pid_is_task_executor(pid, task_id, task_dir) and not force: logging.info("[SP_UI] executor_already_spawned task_id=%s pid=%s", task_id, pid) return pid except Exception: pass # A stale local claim marker can never establish ownership for a new # launch/resume. Epoch fencing in Worker/D1, not process termination here, # prevents an older executor from overwriting a newer epoch. try: claim_marker_path.unlink(missing_ok=True) except TypeError: if claim_marker_path.exists(): claim_marker_path.unlink() log_path = task_dir / "executor.log" task_dir.mkdir(parents=True, exist_ok=True) command = [ sys.executable, str(Path(__file__).resolve().parent / "task_executor.py"), "--task-dir", str(task_dir.resolve()), "--task-id", str(task_id), "--user-id", str(numeric_id), ] executor_env = os.environ.copy() executor_env["SMARTPAPER_EXECUTOR_LOG_PATH"] = str(log_path.resolve()) process = subprocess.Popen( command, cwd=str(Path(__file__).resolve().parent), stdin=subprocess.DEVNULL, # Inherit stdout/stderr so executor bootstrap/errors are visible in HF # Container Logs; task_executor also writes the same log to executor.log. close_fds=True, start_new_session=True, env=executor_env, ) _atomic_json( launch_marker_path, { "pid": process.pid, "task_id": str(task_id), "request_hash": request_hash, "launched_at": datetime.now(timezone.utc).isoformat(), }, ) logging.info("[SP_UI] executor_spawned task_id=%s pid=%s", task_id, process.pid) return int(process.pid) def _process_pending_cancel() -> None: task_id = str(st.session_state.get("sp_pending_cancel_task_id") or "").strip() if not task_id: return result = cancel_generation_task(task_id) if result.get("cancelled"): _cleanup_task_artifacts(task_id) st.session_state["sp_pending_cancel_task_id"] = "" st.session_state["sp_generation_task"] = dict(result.get("task") or {}) or None st.rerun() def _process_pending_resume() -> None: task_id = str(st.session_state.get("sp_pending_resume_task_id") or "").strip() if not task_id: return task = dict(st.session_state.get("sp_generation_task") or {}) capable, reason = _local_resume_capability(task) if str(task.get("task_id") or "").strip() != task_id or not capable: logging.warning("[SP_UI] resume_unavailable task_id=%s reason=%s", task_id, reason) st.session_state["sp_pending_resume_task_id"] = "" lang_code = str(st.session_state.get("sp_lang_code") or "zh") st.session_state["sp_terminal_notice"] = { "kind": "warning", "message": tr(lang_code, "resume_unavailable_runtime_replaced"), } st.rerun() _launch_executor(task_id, force=True) st.session_state["sp_pending_resume_task_id"] = "" st.session_state["sp_scroll_to_generation_action"] = True st.rerun() def _process_pending_launch(lang_code: str) -> None: pending = st.session_state.get("sp_pending_launch") if not isinstance(pending, dict): return mode = str(pending.get("mode") or "").strip().lower() idempotency_key = str(pending.get("idempotency_key") or "").strip() projection = get_product_ui_projection() selected_product_tier = ( str(st.session_state.get("sp_product_tier") or "").strip().upper() if projection.get("mode") == "a5_product" else None ) try: response = start_generation_task( idempotency_key=idempotency_key, mode=mode, interface_language=lang_code, target_words=int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]), product_tier=selected_product_tier, ) except Exception as exc: st.session_state["sp_pending_launch"] = None raw = str(exc or "") if "payment_required" in raw: st.session_state["sp_payment_required"] = True st.error(tr(lang_code, "quota_exhausted")) render_payment_purchase(lang_code) elif "a5_product_task_path_not_authorized" in raw: st.error(tr(lang_code, "product_generation_not_active")) else: st.error(raw) return task = dict(response.get("task") or {}) task_id = str(task.get("task_id") or "").strip() if not task_id: st.session_state["sp_pending_launch"] = None st.error("task_start_missing_task_id") return st.session_state["sp_generation_task"] = task st.session_state["sp_last_ui_task_status"] = str(task.get("status") or "") try: request_path = _task_dir(task_id) / "request.json" if not request_path.is_file(): _persist_task_request(task, mode, lang_code) _launch_executor(task_id) except Exception as exc: st.session_state["sp_pending_launch"] = None st.session_state["sp_pending_cancel_task_id"] = task_id st.session_state["sp_terminal_notice"] = {"kind": "error", "message": str(exc)[:800]} st.rerun() st.session_state["sp_pending_launch"] = None st.session_state["sp_payment_required"] = False st.session_state["sp_scroll_to_generation_action"] = True st.rerun() inject_wechat_browser_notice() DEFAULT_TOPIC = "Generative Artificial Intelligence in Ubiquitous Learning: Evaluating a Chatbot-based Recommendation Engine for Personalized and Context-aware Education" DEFAULT_KEYWORDS = "Ubiquitous computing; Context awareness; Deep learning; Personalized educational; Educational technologies; HCI; Recommendation system; LLMs; Generative AI; ChatGPT; Prompting engineering; Few-shot prompting" def _current_language_code() -> str: choice = st.session_state.get("interface_language_choice") return LANGUAGE_OPTIONS.get(choice, "zh") authenticated_user = require_alipay_login(_current_language_code()) if not bool(st.session_state.get("sp_task_recovery_initialized")): recover_generation_task() st.session_state["sp_task_recovery_initialized"] = True _recovered_task = st.session_state.get("sp_generation_task") or {} _recovered_mode = str((_recovered_task or {}).get("mode") or "").strip().lower() if _recovered_mode in {"amj", "asq"}: # The task record, not the first visual tab, owns the research mode after refresh. st.session_state["run_mode"] = _recovered_mode _user_workspace = os.path.join(WORKSPACE_DIR, "users", _safe_key(authenticated_user.get("id"))) os.makedirs(_user_workspace, exist_ok=True) ACTIVE_TASK_STATUSES = {"created", "running", "resumable", "cancelling"} LIVE_TASK_STATUSES = {"created", "running", "cancelling"} def _localized_task_status(lang_code: str, status: str) -> str: key = f"task_status_{str(status or '').strip().lower()}" value = tr(lang_code, key) return value if value != key else tr(lang_code, "draft_stage_processing") def _localized_monitor_node(lang_code: str, node: str) -> str: """Translate stable executor progress codes without exposing engineering identifiers.""" value = str(node or "").strip() exact = { "inputs_and_evidence": "draft_stage_inputs", "S2 + OpenAlex": "draft_stage_search", "literature_synthesis": "draft_stage_synthesis", "research_logic": "draft_stage_research_logic", "working_front_matter": "draft_stage_working_front", "outline_planning": "draft_stage_outline", "citation_planning": "draft_stage_citation", "english_validation": "draft_stage_english_validation", "final_front_matter": "draft_stage_final_front", "translation": "draft_stage_translation", "translation_complete": "translation_complete_label", "packaging": "draft_stage_packaging", "package_validation": "draft_stage_package_validation", "package_validated": "draft_stage_package_validated", } if value in exact: return tr(lang_code, exact[value]) for prefix, key in ( ("writing:", "draft_stage_writing_unit"), ("citation_audit:", "draft_stage_citation_audit_unit"), ("citation_recovery:", "draft_stage_citation_recovery_unit"), ("writing_complete:", "draft_stage_writing_complete_unit"), ("translation_final_gate:", "draft_stage_translation"), ): if value.startswith(prefix): parts = value[len(prefix):].split(":") if key == "draft_stage_translation": return tr(lang_code, key) if len(parts) >= 2: return tr(lang_code, key, current=parts[0], total=parts[1]) return tr(lang_code, "draft_stage_processing") if value.startswith("translation_"): return tr(lang_code, "draft_stage_translation") if value.startswith("section_"): return tr(lang_code, "draft_stage_writing") return tr(lang_code, "draft_stage_processing") def _markdown_safe_html(markup: str) -> str: """Prevent Streamlit Markdown from reclassifying indented HTML as code. Only indentation before actual HTML tags is removed. Text rendered inside preview containers keeps its own leading whitespace. """ lines = [] for line in textwrap.dedent(str(markup or "")).strip().splitlines(): stripped = line.lstrip() lines.append(stripped if stripped.startswith("<") else line) return "\n".join(lines) def _render_monitor_box(task: dict, lang_code: str) -> None: """Render the approved monitor location with DOM targets for live updates.""" status = str(task.get("status") or "") node = str(task.get("current_node") or status or "") status_label = _localized_task_status(lang_code, status) node_label = _localized_monitor_node(lang_code, node) progress = max(0, min(100, int(task.get("progress_pct") or 0))) checkpoint_seq = int(task.get("checkpoint_seq") or 0) checkpoint_stage = str(task.get("checkpoint_stage") or "") ds_calls = int(task.get("ds_calls") or 0) s2_calls = int(task.get("s2_calls") or 0) oa_calls = int(task.get("oa_calls") or 0) progress_markup = "" if status in ACTIVE_TASK_STATUSES: progress_markup = textwrap.dedent(f"""
{progress}% — {html.escape(node_label)}
""").strip() st.markdown(_markdown_safe_html(f"""
{progress_markup}
{tr(lang_code, 'monitor_pipeline')}: {html.escape(status_label)}
{tr(lang_code, 'monitor_node')}: {html.escape(node_label)}
{tr(lang_code, 'monitor_log')}: {html.escape(tr(lang_code, 'monitor_checkpoint_saved', seq=checkpoint_seq))}
• {html.escape(node_label)}

{tr(lang_code, 'monitor_compute')}:
🧠 DeepSeek: {ds_calls} 📚 S2: {s2_calls} 🌐 OpenAlex: {oa_calls}
"""), unsafe_allow_html=True) def _watch_task_status_live(task: dict) -> dict | None: """Start browser-only read polling for live UI state. DOM updates remain read-only and never participate in task liveness. During execution, progress and manuscript preview updates stay entirely in the browser; Streamlit is notified only once at terminal/resumable state. This prevents checkpoint-driven full-page reruns and visible flicker. """ task_id = str(task.get("task_id") or "").strip() status = str(task.get("status") or "") if not task_id or status not in LIVE_TASK_STATUSES: return None api_origin = str(st.session_state.get("sp_auth_api_origin") or "").strip() if not api_origin: logging.warning("[SP_UI] live_watch_missing_api_origin task_id=%s", task_id) return None expected_request_hash = str(task.get("request_hash") or "").strip().lower() if not expected_request_hash: try: request_path = _task_dir(task_id) / "request.json" if request_path.is_file(): expected_request_hash = str(json.loads(request_path.read_text(encoding="utf-8")).get("request_hash") or "").strip().lower() except Exception: expected_request_hash = "" try: watch_result = edge_bridge( action="task_watch", command_id=f"watch:{task_id}", api_origin=api_origin, payload={ "task_id": task_id, "poll_interval_ms": 2000, "preview_interval_ms": 700, "preview_path": _task_preview_path(task_id), "expected_execution_epoch": int(task.get("execution_epoch") or 0), "expected_request_hash": expected_request_hash, "initial_checkpoint_seq": int(task.get("checkpoint_seq") or 0), "best_effort_page_exit_cancel": False, "draft_status_labels": { "default": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_preparing"), "inputs_and_evidence": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_inputs"), "S2 + OpenAlex": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_search"), "literature_synthesis": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_synthesis"), "research_logic": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_research_logic"), "outline_planning": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_outline"), "citation_planning": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_citation"), "writing": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_writing"), "working_front_matter": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_working_front"), "writing_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_writing_unit"), "citation_audit_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_citation_audit_unit"), "citation_recovery_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_citation_recovery_unit"), "writing_complete_unit": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_writing_complete_unit"), "english_validation": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_english_validation"), "final_front_matter": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_final_front"), "translation": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_translation"), "packaging": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_packaging"), "package_validation": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_package_validation"), "package_validated": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_package_validated"), "processing": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_stage_processing"), "checkpoint_saved": tr(str(st.session_state.get("sp_lang_code") or "zh"), "monitor_checkpoint_saved"), "status_created": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_created"), "status_running": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_running"), "status_resumable": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_resumable"), "status_cancelling": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_cancelling"), "status_completed": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_completed"), "status_failed": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_failed"), "status_cancelled": tr(str(st.session_state.get("sp_lang_code") or "zh"), "task_status_cancelled"), "evidence_limited": tr(str(st.session_state.get("sp_lang_code") or "zh"), "evidence_limited_running"), "progress": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_progress"), "s2": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_s2"), "oa": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_oa"), "current": tr(str(st.session_state.get("sp_lang_code") or "zh"), "draft_detail_current"), }, "target_status_labels": { "waiting": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_waiting"), "translating": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_running"), "qa": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_qa"), "final_gate": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_final_gate"), "repairing": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_repairing"), "repair_pending": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_repair_pending"), "adjudicating": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_adjudicating"), "counts": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_progress_counts"), "ready": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_section_ready"), "complete": tr(str(st.session_state.get("sp_lang_code") or "zh"), "translation_complete_label"), }, }, key=f"sp_task_watch_{_safe_key(task_id)}", ) boundary_task = watch_result.get("task") if isinstance(watch_result, dict) and watch_result.get("boundary") else None if isinstance(watch_result, dict) and watch_result.get("boundary"): navdiag = watch_result.get("ui_navdiag") if isinstance(navdiag, dict): try: safe_diag = json.dumps(navdiag, ensure_ascii=False, separators=(",", ":"))[:4000] logging.info("[SP_UI_NAVDIAG] task_id=%s snapshot=%s", task_id, safe_diag) except Exception: logging.info("[SP_UI_NAVDIAG] task_id=%s snapshot_unserializable", task_id) if isinstance(boundary_task, dict) and str(boundary_task.get("task_id") or "") == task_id: st.session_state["sp_generation_task"] = dict(boundary_task) st.session_state["sp_last_ui_task_status"] = str(boundary_task.get("status") or "") return dict(boundary_task) except Exception as exc: logging.warning("[SP_UI] live_watch_error task_id=%s error=%s", task_id, type(exc).__name__) return None def _load_confirmed_checkpoint(task: dict) -> dict | None: """Load only the checkpoint sequence already acknowledged by Worker/D1.""" task_id = str(task.get("task_id") or "").strip() checkpoint_seq = int(task.get("checkpoint_seq") or 0) if not task_id or checkpoint_seq <= 0: return None task_dir = _task_dir(task_id) request_path = task_dir / "request.json" if not request_path.is_file(): return None try: request_payload = json.loads(request_path.read_text(encoding="utf-8")) expected_hash = str(request_payload.get("request_hash") or "") candidates = sorted((task_dir / "checkpoints").glob(f"checkpoint_{checkpoint_seq:05d}_*.json")) if not candidates: return None payload = json.loads(candidates[-1].read_text(encoding="utf-8")) if not isinstance(payload, dict): return None if expected_hash and str(payload.get("request_hash") or "") != expected_hash: return None if int(payload.get("checkpoint_seq") or 0) != checkpoint_seq: return None return payload except Exception as exc: logging.warning("[SP_UI] checkpoint_preview_error task_id=%s error=%s", task_id, type(exc).__name__) return None _PREVIEW_UNNUMBERED_HEADINGS = { "abstract", "keyword", "keywords", "references", "reference list", "摘要", "关键词", "关键字", "参考文献", "аннотация", "ключевое слово", "ключевые слова", "литература", "список литературы", "аңдатпа", "түйіндеме", "түйін сөздер", "түйінді сөздер", "кілт сөздер", "әдебиеттер", "пайдаланылған әдебиеттер", } _PREVIEW_UNNUMBERED_PREFIXES = ("appendix", "附录", "приложение", "қосымша") def _normalize_preview_display_text(value: str) -> str: """Mirror the browser heading presentation for initial/terminal rerenders. Live updates are normalized by the browser bridge. A terminal Streamlit rerender used to fall back to raw Markdown, so completed pages could show ``#`` markers or lose the Arabic numbering users saw during generation. Keep one deterministic server-side equivalent for the initial/terminal DOM. """ lines = str(value or "").replace("\r", "").split("\n") heading_rows = [] for index, line in enumerate(lines): match = re.match(r"^(#{1,3})\s+(.+?)\s*$", line) if not match: continue title = re.sub(r"^\d+(?:\.\d+){0,2}\s+", "", match.group(2).strip()) heading_rows.append((index, len(match.group(1)), title)) if not heading_rows: return "\n".join(lines) def unnumbered(title: str) -> bool: key = str(title or "").strip().casefold() return key in _PREVIEW_UNNUMBERED_HEADINGS or any(key.startswith(prefix) for prefix in _PREVIEW_UNNUMBERED_PREFIXES) has_front_matter = any(unnumbered(title) for _index, _level, title in heading_rows) first_h1 = next((row for row in heading_rows if row[1] == 1), None) counters = [0, 0, 0] for index, level, title in heading_rows: is_known_unnumbered = unnumbered(title) is_title = bool(has_front_matter and first_h1 and index == first_h1[0] and not is_known_unnumbered) if is_known_unnumbered or is_title: lines[index] = title continue if level == 1: counters[0] += 1 counters[1] = counters[2] = 0 lines[index] = f"{counters[0]} {title}" elif level == 2: if counters[0] == 0: counters[0] = 1 counters[1] += 1 counters[2] = 0 lines[index] = f"{counters[0]}.{counters[1]} {title}" else: if counters[0] == 0: counters[0] = 1 if counters[1] == 0: counters[1] = 1 counters[2] += 1 lines[index] = f"{counters[0]}.{counters[1]}.{counters[2]} {title}" return "\n".join(lines) def _render_checkpoint_preview(task: dict, lang_code: str) -> None: """Render stable bilingual DOM targets; browser mutates them in place.""" checkpoint = _load_confirmed_checkpoint(task) english = str((checkpoint or {}).get("full_draft_en") or "").strip() target = str((checkpoint or {}).get("full_draft_translated") or "").strip() english_initial = _normalize_preview_display_text(english) english_status_initial = "" if english else tr(lang_code, "draft_preparing") english_status_display = "none" if english else "block" target_initial = _normalize_preview_display_text(target) if target else tr(lang_code, "translation_waiting") english_placeholder = "0" if english else "1" target_placeholder = "0" if target else "1" left, right = st.columns([1, 1], gap="medium") with left: st.markdown(tr(lang_code, "english_draft")) st.markdown( _markdown_safe_html(f"""
{html.escape(english_status_initial)}
{html.escape(english_initial)}
"""), unsafe_allow_html=True, ) with right: st.markdown(tr(lang_code, "target_draft", flag=get_target_flag(lang_code))) st.markdown( _markdown_safe_html(f"""
{html.escape(target_initial)}
"""), unsafe_allow_html=True, ) st.markdown("", unsafe_allow_html=True) def _scroll_to_completed_download_once(task_id: str) -> None: """Scroll only after the task-scoped download anchor actually exists. Success identity is browser-session scoped and is written only after scrollIntoView executes. A missed early attempt therefore remains retryable on later Streamlit reruns. """ task_id = str(task_id or "").strip() if not task_id: return components.html( f""" """, height=0, width=0, ) def _render_task_status_once(lang_code: str) -> None: task = st.session_state.get("sp_generation_task") or {} task_id = str(task.get("task_id") or "").strip() if not task_id: return status = str(task.get("status") or "") # First-paint invariant: render useful, localized progress and manuscript # placeholders before any browser bridge call that may yield once via # st.stop(). The browser watcher mutates these stable DOM targets in place # after the bridge handshake, so no duplicate monitor/preview is needed. st.session_state["sp_last_ui_task_status"] = status _render_monitor_box(task, lang_code) _render_checkpoint_preview(task, lang_code) if status in ACTIVE_TASK_STATUSES: try: result = generation_task_status(task_id) task = dict(result.get("task") or task) st.session_state["sp_generation_task"] = task status = str(task.get("status") or "") except Exception as exc: logging.warning("[SP_UI] status_poll_error task_id=%s error=%s", task_id, type(exc).__name__) if status in LIVE_TASK_STATUSES: boundary_task = _watch_task_status_live(task) if isinstance(boundary_task, dict): task = boundary_task status = str(task.get("status") or "") st.session_state["sp_last_ui_task_status"] = status if status == "resumable": resume_capable, _ = _local_resume_capability(task) st.warning(tr(lang_code, "checkpoint_warning" if resume_capable else "resume_unavailable_runtime_replaced")) elif status == "completed": output_path = _task_output_path(task) if output_path and output_path.is_file(): st.markdown( f"
", unsafe_allow_html=True, ) if str(task.get("safe_error_code") or "") == "W_EVIDENCE_COVERAGE_LIMITED": st.warning(tr(lang_code, "evidence_limited_completed")) st.success(tr(lang_code, "success")) st.download_button(label=tr(lang_code, "download"), data=output_path.read_bytes(), file_name=output_path.name, mime="application/zip") _scroll_to_completed_download_once(task_id) else: st.warning("The task completed, but the local output file is unavailable after runtime replacement.") elif status == "failed": error_message, translation_note = _localized_terminal_task_error(task, lang_code) st.error(error_message) if translation_note: st.warning(translation_note) elif status == "cancelled": detail = str(task.get("safe_error_detail") or "") if detail: st.warning(detail) def _localized_terminal_task_error(task: dict, lang_code: str) -> tuple[str, str]: """Map stable business error codes to UI-language messages. Engineering details remain in task/log state; the user-facing terminal view does not expose a long English runtime exception in non-English interfaces. """ code = str(task.get("safe_error_code") or "").strip() detail = str(task.get("safe_error_detail") or "").strip() raw = f"{code} {detail}" if "E_LITERATURE_INSUFFICIENT" in raw: return tr(lang_code, "literature_insufficient"), tr(lang_code, "upstream_translation_stopped") return detail or code or tr(lang_code, "node_error"), "" def _inject_generation_action_scroll(mode: str) -> None: """Scroll immediately to the active generation action without a server wait.""" target_id = "sp-generation-action-asq" if str(mode or "").lower() == "asq" else "sp-generation-action-amj" components.html( f""" """, height=0, width=0, ) def _scroll_to_generation_action_once() -> None: """One-time post-launch/resume scroll to the monitor/live manuscript.""" if not bool(st.session_state.get("sp_scroll_to_generation_action")): return st.session_state["sp_scroll_to_generation_action"] = False mode = str(st.session_state.get("run_mode") or "asq") fallback_id = "sp-generation-action-asq" if mode.lower() == "asq" else "sp-generation-action-amj" components.html( f""" """, height=0, width=0, ) def _select_active_task_tab(task: dict, lang_code: str) -> None: """Keep the existing visual tab aligned with the immutable task mode after rerun/refresh.""" status = str((task or {}).get("status") or "") mode = str((task or {}).get("mode") or "").strip().lower() if mode not in {"amj", "asq"}: return if status not in (ACTIVE_TASK_STATUSES | {"resumable", "completed", "failed", "cancelled"}): return label = tr(lang_code, "tab_asq" if mode == "asq" else "tab_amj") components.html( f""" """, height=0, width=0, ) def render_workspace_controls(user): top_toolbar = st.container() with top_toolbar: language_col, account_col = st.columns([5.4, 4.6]) with language_col: lang_choice = st.radio( "Language", list(LANGUAGE_OPTIONS.keys()), index=0, horizontal=True, label_visibility="collapsed", key="interface_language_choice", ) lang_code = LANGUAGE_OPTIONS[lang_choice] st.session_state["sp_lang_code"] = lang_code st.session_state["sp_target_lang"] = get_target_language(lang_code) st.session_state["sp_output_suffix"] = get_output_suffix(lang_code) st.session_state["sp_target_flag"] = get_target_flag(lang_code) with account_col: render_logged_in_account(lang_code, user) task = st.session_state.get("sp_generation_task") or {} task_status = str(task.get("status") or "") task_mode = str(task.get("mode") or "").strip().lower() if task_mode in {"amj", "asq"}: st.session_state["run_mode"] = task_mode active = task_status in ACTIVE_TASK_STATUSES st.session_state["is_running"] = active st.markdown(f"

{tr(lang_code, 'title')}

", unsafe_allow_html=True) if task_status == "resumable": resume_capable, resume_reason = _local_resume_capability(task) if resume_capable: st.warning(tr(lang_code, "checkpoint_warning")) else: logging.warning( "[SP_UI] local_resume_capability_missing task_id=%s reason=%s", str(task.get("task_id") or ""), resume_reason, ) st.warning(tr(lang_code, "resume_unavailable_runtime_replaced")) colA, colB = st.columns([1, 1]) if colA.button(tr(lang_code, "resume"), key="resume_task", disabled=not resume_capable): st.session_state["sp_pending_resume_task_id"] = str(task.get("task_id") or "") st.rerun() if colB.button(tr(lang_code, "clear"), key="clear_task"): st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "") st.rerun() st.markdown("---") st.text_input(tr(lang_code, "topic_label"), value=DEFAULT_TOPIC, disabled=active, key="sp_topic") st.text_input(tr(lang_code, "keywords_label"), value=DEFAULT_KEYWORDS, disabled=active, key="sp_keywords") product_ui = get_product_ui_projection() if product_ui.get("mode") == "a5_product": products = product_ui.get("products") or {} tier_options = [tier for tier in ("STANDARD", "FLAGSHIP") if tier in products] if len(tier_options) != 2: st.error(tr(lang_code, "product_generation_not_active")) st.stop() selected_tier = str(st.session_state.get("sp_product_tier") or "STANDARD").strip().upper() if selected_tier not in tier_options: selected_tier = "STANDARD" st.session_state["sp_product_tier"] = selected_tier selected_tier = st.radio( tr(lang_code, "product_tier_label"), tier_options, index=tier_options.index(selected_tier), format_func=lambda tier: tr(lang_code, f"product_tier_{tier.lower()}"), horizontal=True, disabled=active, key="sp_product_tier", ) selected_product = products[selected_tier] word_min = int(selected_product["min_body_words"]) word_max = int(selected_product["max_body_words"]) word_label = tr( lang_code, "words_label_product", min_words=word_min, max_words=word_max, ) else: st.session_state.pop("sp_product_tier", None) word_min = int(APP_CONFIG["target_words_min"]) word_max = int(APP_CONFIG["target_words_max"]) word_label = tr( lang_code, "words_label", min_words=word_min, max_words=word_max, ) current_words = int(st.session_state.get("sp_words") or APP_CONFIG["target_words_default"]) bounded_words = min(word_max, max(word_min, current_words)) if st.session_state.get("sp_words") != bounded_words: st.session_state["sp_words"] = bounded_words st.number_input( word_label, min_value=word_min, max_value=word_max, value=bounded_words, step=1000, disabled=active, key="sp_words", ) tab_asq, tab_amj = st.tabs([tr(lang_code, "tab_asq"), tr(lang_code, "tab_amj")]) with tab_asq: st.markdown(tr(lang_code, "asq_config")) st.info(tr(lang_code, "asq_info")) st.file_uploader(tr(lang_code, "upload_reference"), type=["pdf", "doc", "docx"], accept_multiple_files=True, key="up_asq_ref", disabled=active) st.file_uploader(tr(lang_code, "upload_interviews"), type=["txt", "pdf", "doc", "docx"], key="up_asq_data", disabled=active) with st.expander(tr(lang_code, "asq_expander")): st.markdown(tr(lang_code, "asq_hint")) template = get_asq_txt_template(lang_code) st.code(template, language="text") st.download_button(label=tr(lang_code, "download_asq_template"), data=template, file_name="ASQ_Interview_Template.txt", mime="text/plain", use_container_width=True) st.markdown("
", unsafe_allow_html=True) if active and _task_mode(task) == "asq": if st.button(tr(lang_code, "cancel_generation"), key="cancel_asq"): st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "") st.rerun() elif st.button(tr(lang_code, "start_asq"), disabled=active, key="btn_asq"): st.session_state["run_mode"] = "asq" st.session_state["sp_pending_launch"] = {"mode": "asq", "idempotency_key": _new_generation_idempotency_key()} st.session_state["sp_scroll_to_generation_action"] = True st.info(tr(lang_code, "pipeline_start")) _inject_generation_action_scroll("asq") with tab_amj: st.markdown(tr(lang_code, "amj_config")) st.info(tr(lang_code, "amj_info")) st.file_uploader(tr(lang_code, "upload_reference"), type=["pdf", "doc", "docx"], accept_multiple_files=True, key="up_amj_ref", disabled=active) st.file_uploader(tr(lang_code, "upload_stats"), type=["csv", "xlsx", "xls"], key="up_amj_data", disabled=active) with st.expander(tr(lang_code, "amj_expander")): st.markdown(tr(lang_code, "amj_hint")) template = get_amj_csv_template(lang_code) st.code(template, language="csv") st.download_button(label=tr(lang_code, "download_amj_template"), data=template, file_name="AMJ_Data_Template.csv", mime="text/csv", use_container_width=True) st.markdown("
", unsafe_allow_html=True) if active and _task_mode(task) == "amj": if st.button(tr(lang_code, "cancel_generation"), key="cancel_amj"): st.session_state["sp_pending_cancel_task_id"] = str(task.get("task_id") or "") st.rerun() elif st.button(tr(lang_code, "start_amj"), disabled=active, key="btn_amj"): st.session_state["run_mode"] = "amj" st.session_state["sp_pending_launch"] = {"mode": "amj", "idempotency_key": _new_generation_idempotency_key()} st.session_state["sp_scroll_to_generation_action"] = True st.info(tr(lang_code, "pipeline_start")) _inject_generation_action_scroll("amj") _select_active_task_tab(task, lang_code) inject_file_uploader_i18n(lang_code) render_workspace_controls(authenticated_user) lang_code = str(st.session_state.get("sp_lang_code") or _current_language_code()) _notice = st.session_state.pop("sp_terminal_notice", None) if isinstance(_notice, dict) and _notice.get("message"): (st.error if _notice.get("kind") == "error" else st.warning)(str(_notice.get("message"))) _process_pending_cancel() _process_pending_resume() _process_pending_launch(lang_code) if st.session_state.get("sp_payment_required"): render_payment_purchase(lang_code) # R3.2.4 keeps the approved task-monitor location. Active progress is updated # inside the browser by read-only edge polling; Streamlit reruns only on a # user action or a terminal/resumable boundary, so there is no periodic flicker. _task_for_monitor = st.session_state.get("sp_generation_task") or {} if str(_task_for_monitor.get("task_id") or "").strip(): st.markdown("
", unsafe_allow_html=True) # Install the one-time navigation script before task-status bridge calls. # If the first bridge handshake yields via st.stop(), the browser still # receives the anchor and can position the user at visible task progress. _scroll_to_generation_action_once() _render_task_status_once(lang_code)