Step 4A-6-C-R3.2.4 Live Progress UX & Terminal Notification Fix

.gitattributes

@@ -33,4 +33,3 @@ saved_model/**/* filter=lfs diff=lfs merge=lfs -text
 *.zip filter=lfs diff=lfs merge=lfs -text
 *.zst filter=lfs diff=lfs merge=lfs -text
 *tfevents* filter=lfs diff=lfs merge=lfs -text
-core/__pycache__/pipeline_agent.cpython-313.pyc filter=lfs diff=lfs merge=lfs -text

app.py

@@ -30,6 +30,7 @@ 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",
@@ -267,6 +268,7 @@ for key, default in {
     "sp_terminal_notice": None,
     "sp_task_recovery_initialized": False,
     "sp_last_ui_task_status": "",
+    "sp_scroll_to_generation_action": False,
 }.items():
     if key not in st.session_state:
         st.session_state[key] = default
@@ -494,7 +496,7 @@ def _process_pending_resume() -> None:
         st.rerun()
     _launch_executor(task_id, force=True)
     st.session_state["sp_pending_resume_task_id"] = ""
-    st.session_state["sp_scroll_to_task_monitor"] = True
+    st.session_state["sp_scroll_to_generation_action"] = True
     st.rerun()
 
 
@@ -541,7 +543,7 @@ def _process_pending_launch(lang_code: str) -> None:
         st.rerun()
     st.session_state["sp_pending_launch"] = None
     st.session_state["sp_payment_required"] = False
-    st.session_state["sp_scroll_to_task_monitor"] = True
+    st.session_state["sp_scroll_to_generation_action"] = True
     st.rerun()
 
 
@@ -564,26 +566,80 @@ os.makedirs(_user_workspace, exist_ok=True)
 
 
 ACTIVE_TASK_STATUSES = {"created", "running", "resumable", "cancelling"}
+LIVE_TASK_STATUSES = {"created", "running", "cancelling"}
 
 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 "")
     progress = max(0, min(100, int(task.get("progress_pct") or 0)))
-    if status in {"created", "running", "resumable", "cancelling"}:
-        st.progress(progress, text=f"{progress}% — {node}")
+    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 = f"""
+        <div id='sp-live-progress-wrap' style='margin:0 0 12px 0;'>
+          <div id='sp-live-progress-text' style='margin-bottom:7px;'>{progress}% — {node}</div>
+          <div style='width:100%;height:10px;background:rgba(128,128,128,.24);border-radius:999px;overflow:hidden;'>
+            <div id='sp-live-progress-fill' style='width:{progress}%;height:100%;background:currentColor;border-radius:999px;transition:width .35s ease;'></div>
+          </div>
+        </div>
+        """
+
     st.markdown(f"""
-    <div class='log-box'>
-        <div><b>{tr(lang_code, 'monitor_pipeline')}:</b> {status}</div><div><b>{tr(lang_code, 'monitor_node')}:</b> {node}</div><div><b>{tr(lang_code, 'monitor_log')}:</b> checkpoint {int(task.get('checkpoint_seq') or 0)} · {str(task.get('checkpoint_stage') or '')}</div><hr>
+    {progress_markup}
+    <div class='log-box' id='sp-live-monitor'>
+        <div><b>{tr(lang_code, 'monitor_pipeline')}:</b> <span id='sp-live-status'>{status}</span></div>
+        <div><b>{tr(lang_code, 'monitor_node')}:</b> <span id='sp-live-node'>{node}</span></div>
+        <div><b>{tr(lang_code, 'monitor_log')}:</b> checkpoint <span id='sp-live-checkpoint-seq'>{checkpoint_seq}</span> · <span id='sp-live-checkpoint-stage'>{checkpoint_stage}</span></div><hr>
         <div><b>{tr(lang_code, 'monitor_compute')}:</b></div>
-        <div style='display: flex; gap: 8px;'>
-            <span style='background: #fef3c7; padding: 4px 8px; border-radius: 4px; color: black;'>🧠 DeepSeek: <b>{int(task.get('ds_calls') or 0)}</b></span>
-            <span style='background: #dbeafe; padding: 4px 8px; border-radius: 4px; color: black;'>📚 S2: <b>{int(task.get('s2_calls') or 0)}</b></span>
-            <span style='background: #d1fae5; padding: 4px 8px; border-radius: 4px; color: black;'>🌐 OpenAlex: <b>{int(task.get('oa_calls') or 0)}</b></span>
+        <div style='display:flex;gap:8px;'>
+            <span style='background: #fef3c7;padding:4px 8px;border-radius:4px;color:black;'>🧠 DeepSeek: <b id='sp-live-ds'>{ds_calls}</b></span>
+            <span style='background: #dbeafe;padding:4px 8px;border-radius:4px;color:black;'>📚 S2: <b id='sp-live-s2'>{s2_calls}</b></span>
+            <span style='background: #d1fae5;padding:4px 8px;border-radius:4px;color:black;'>🌐 OpenAlex: <b id='sp-live-oa'>{oa_calls}</b></span>
         </div>
     </div>
     """, unsafe_allow_html=True)
 
 
+def _watch_task_status_live(task: dict) -> None:
+    """Start browser-only read polling for active task UI state.
+
+    The component updates monitor DOM locally. It sends a Streamlit value only
+    when a terminal task state is observed, causing exactly one boundary rerun.
+    It never sends heartbeat, claim, finish, or cancellation commands.
+    """
+    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
+    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
+    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": 2500},
+            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(boundary_task, dict) and str(boundary_task.get("task_id") or "") == task_id:
+            # Consume the watcher's authoritative boundary snapshot directly so
+            # the ordinary 5-second task-status cache cannot delay terminal UI.
+            st.session_state["sp_generation_task"] = dict(boundary_task)
+            st.session_state["sp_last_ui_task_status"] = str(boundary_task.get("status") or "")
+            st.rerun()
+    except Exception as exc:
+        logging.warning("[SP_UI] live_watch_error task_id=%s error=%s", task_id, type(exc).__name__)
+
+
 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()
@@ -591,7 +647,7 @@ def _render_task_status_once(lang_code: str) -> None:
         return
     previous_status = str(st.session_state.get("sp_last_ui_task_status") or str(task.get("status") or ""))
     status = str(task.get("status") or "")
-    if status in {"created", "running", "resumable", "cancelling"}:
+    if status in ACTIVE_TASK_STATUSES:
         try:
             result = generation_task_status(task_id)
             task = dict(result.get("task") or task)
@@ -601,7 +657,9 @@ def _render_task_status_once(lang_code: str) -> None:
             logging.warning("[SP_UI] status_poll_error task_id=%s error=%s", task_id, type(exc).__name__)
     st.session_state["sp_last_ui_task_status"] = status
     _render_monitor_box(task, lang_code)
-    if status == "completed":
+    if status in LIVE_TASK_STATUSES:
+        _watch_task_status_live(task)
+    elif status == "completed":
         output_path = _task_output_path(task)
         if output_path and output_path.is_file():
             st.success(tr(lang_code, "success"))
@@ -615,27 +673,35 @@ def _render_task_status_once(lang_code: str) -> None:
         if detail:
             st.warning(detail)
 
-    # Only boundary transitions need one full app rerun so static controls pick
-    # up resume/terminal state. Running progress never reruns the whole page.
+    # Only terminal/resume boundaries may rerun the whole Streamlit app. Live
+    # progress is DOM-only and therefore cannot reintroduce periodic flicker.
     if previous_status and status != previous_status and status in {"resumable", "completed", "failed", "cancelled"}:
         st.rerun()
 
 
-def _scroll_to_task_monitor_once() -> None:
-    """One-time UX correction after launch/resume; never polls or reruns."""
-    if not bool(st.session_state.get("sp_scroll_to_task_monitor")):
+def _scroll_to_generation_action_once() -> None:
+    """One-time post-launch/resume scroll to the paper-generation action area."""
+    if not bool(st.session_state.get("sp_scroll_to_generation_action")):
         return
-    st.session_state["sp_scroll_to_task_monitor"] = False
+    st.session_state["sp_scroll_to_generation_action"] = False
+    mode = str(st.session_state.get("run_mode") or "amj").lower()
+    target_id = "sp-generation-action-asq" if mode == "asq" else "sp-generation-action-amj"
     components.html(
-        """
+        f"""
         <script>
-        (() => {
+        (() => {{
           const parentDocument = window.parent.document;
-          window.parent.setTimeout(() => {
-            const target = parentDocument.getElementById('sp-task-monitor-anchor');
-            if (target) target.scrollIntoView({behavior: 'auto', block: 'start'});
-          }, 120);
-        })();
+          const targetId = {json.dumps(target_id)};
+          let done = false;
+          const locate = () => {{
+            if (done) return;
+            const target = parentDocument.getElementById(targetId);
+            if (!target) return;
+            done = true;
+            target.scrollIntoView({{behavior:'auto', block:'center'}});
+          }};
+          [60, 160, 320, 650, 1100].forEach((delay) => window.parent.setTimeout(locate, delay));
+        }})();
         </script>
         """,
         height=0,
@@ -702,6 +768,7 @@ def render_workspace_controls(user):
             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("<div id='sp-generation-action-amj'></div>", unsafe_allow_html=True)
         if 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()}
@@ -717,6 +784,7 @@ def render_workspace_controls(user):
             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("<div id='sp-generation-action-asq'></div>", unsafe_allow_html=True)
         if 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()}
@@ -739,13 +807,13 @@ _process_pending_launch(lang_code)
 if st.session_state.get("sp_payment_required"):
     render_payment_purchase(lang_code)
 
-# R3.2 UI rollback: keep the task monitor at its pre-R3.1.1 location.
-# It refreshes only on a normal Streamlit rerun (user action/manual page refresh);
-# there is no periodic fragment and therefore no 5-second page flicker.
+# 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("<div id='sp-task-monitor-anchor'></div>", unsafe_allow_html=True)
     _render_task_status_once(lang_code)
-    _scroll_to_task_monitor_once()
+    _scroll_to_generation_action_once()
 
 

core/generation_runner.py

@@ -74,16 +74,30 @@ class GenerationRunner:
         self.output_dir.mkdir(parents=True, exist_ok=True)
         self.agent = None
 
-    def _progress(self, pct: int, node: str, *, checkpoint_seq=None, checkpoint_stage=None) -> None:
+    def _compute_request_counts(self) -> tuple[int, int, int]:
+        """Return one consistent UI metric: actual outbound request attempts."""
         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 {}
+        return (
+            int(getattr(agent, "ds_calls", 0) or 0),
+            int(s2_metric.get("request_count", 0) or 0),
+            int(oa_metric.get("request_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=int(getattr(agent, "ds_calls", 0) or 0),
-            s2_calls=int(getattr(agent, "s2_calls", 0) or 0),
-            oa_calls=int(getattr(agent, "oa_calls", 0) or 0),
+            ds_calls=ds_calls,
+            s2_calls=s2_calls,
+            oa_calls=oa_calls,
         )
 
     def _checkpoint(self, stage: str, state: dict[str, Any], pct: int) -> None:
@@ -527,10 +541,15 @@ Output only final academic prose. Do not report the word count."""
         return {"output_filename": output_filename, "output_path": str(final)}
 
     def _progress_activity(self) -> None:
-        # This call intentionally does not update D1 directly. The executor's
-        # heartbeat thread uses it only to prove that real backend work is still
-        # occurring, which avoids browser-driven or timer-only zombie liveness.
+        # 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)
+            self.progress_callback(
+                activity_only=True,
+                ds_calls=ds_calls,
+                s2_calls=s2_calls,
+                oa_calls=oa_calls,
+            )
         except TypeError:
             pass

smartpaper_restore_bridge/frontend/index.html

@@ -19,7 +19,9 @@
   const MAX_TOKEN_LENGTH = 4096;
   const MAX_RESULT_AGE_MS = 10 * 60 * 1000;
   const REQUEST_TIMEOUT_MS = 20000;
+  const TERMINAL_TASK_STATUSES = new Set(["completed", "failed", "cancelled"]);
   let lastResponseSignature = "";
+  let activeTaskWatch = null;
 
   const ACTION_PATHS = Object.freeze({
     task_start: "/edge/task/start",
@@ -172,6 +174,90 @@
     writeJson(RESULT_PREFIX + commandId, { saved_at:Date.now(), value });
   }
 
+  function asCount(value) {
+    const number = Number(value || 0);
+    return Number.isFinite(number) ? Math.max(0, Math.trunc(number)) : 0;
+  }
+
+  function updateText(parentDocument, id, value) {
+    const element = parentDocument.getElementById(id);
+    if (element) element.textContent = String(value ?? "");
+  }
+
+  function updateTaskMonitor(task) {
+    let parentDocument;
+    try { parentDocument = window.parent.document; }
+    catch (_) { return; }
+    const status = String(task && task.status || "");
+    const node = String(task && task.current_node || status || "");
+    const progress = Math.max(0, Math.min(100, asCount(task && task.progress_pct)));
+    updateText(parentDocument, "sp-live-status", status);
+    updateText(parentDocument, "sp-live-node", node);
+    updateText(parentDocument, "sp-live-checkpoint-seq", asCount(task && task.checkpoint_seq));
+    updateText(parentDocument, "sp-live-checkpoint-stage", String(task && task.checkpoint_stage || ""));
+    updateText(parentDocument, "sp-live-ds", asCount(task && task.ds_calls));
+    updateText(parentDocument, "sp-live-s2", asCount(task && task.s2_calls));
+    updateText(parentDocument, "sp-live-oa", asCount(task && task.oa_calls));
+    updateText(parentDocument, "sp-live-progress-text", `${progress}% — ${node}`);
+    const fill = parentDocument.getElementById("sp-live-progress-fill");
+    if (fill) fill.style.width = `${progress}%`;
+  }
+
+  function stopTaskWatch() {
+    const watch = activeTaskWatch;
+    activeTaskWatch = null;
+    if (watch && watch.timer) window.clearTimeout(watch.timer);
+    if (watch && watch.controller) {
+      try { watch.controller.abort(); } catch (_) {}
+    }
+  }
+
+  function startTaskWatch(commandId, args) {
+    stopTaskWatch();
+    const apiOrigin = normalizeOrigin(args.api_origin || "");
+    const payload = args.payload && typeof args.payload === "object" ? args.payload : {};
+    const taskId = typeof payload.task_id === "string" ? payload.task_id.trim() : "";
+    const requestedInterval = Number(payload.poll_interval_ms || 2500);
+    const intervalMs = Math.max(1500, Math.min(10000, Number.isFinite(requestedInterval) ? requestedInterval : 2500));
+    if (!apiOrigin || !taskId) {
+      sendValue({ ok:false, action:"task_watch", command_id:commandId, error:"invalid_task_watch" });
+      return;
+    }
+
+    const watch = { commandId, taskId, apiOrigin, intervalMs, timer:null, controller:null, stopped:false };
+    activeTaskWatch = watch;
+
+    const schedule = () => {
+      if (activeTaskWatch !== watch || watch.stopped) return;
+      watch.timer = window.setTimeout(poll, watch.intervalMs);
+    };
+
+    const poll = async () => {
+      if (activeTaskWatch !== watch || watch.stopped) return;
+      const result = await postJson(watch.apiOrigin, ACTION_PATHS.task_status, { task_id:watch.taskId });
+      if (activeTaskWatch !== watch || watch.stopped) return;
+      const task = result && result.task && typeof result.task === "object" ? result.task : null;
+      if (result.ok && task) {
+        updateTaskMonitor(task);
+        const status = String(task.status || "");
+        if (TERMINAL_TASK_STATUSES.has(status) || status === "resumable") {
+          watch.stopped = true;
+          activeTaskWatch = null;
+          sendValue(Object.assign({}, result, {
+            action:"task_watch",
+            command_id:commandId,
+            terminal:TERMINAL_TASK_STATUSES.has(status),
+            boundary:true
+          }));
+          return;
+        }
+      }
+      schedule();
+    };
+
+    poll();
+  }
+
   async function handleEdgeAction(action, commandId, args) {
     const path = ACTION_PATHS[action];
     if (!path) return { ok:false, error:"unsupported_edge_action", action, command_id:commandId };
@@ -196,6 +282,11 @@
     const args = message.args || {};
     const action = typeof args.action === "string" ? args.action : "read";
     const commandId = typeof args.command_id === "string" ? args.command_id : "";
+    if (action === "task_watch") {
+      setFrameHeight();
+      startTaskWatch(commandId, args);
+      return;
+    }
     let response;
     try {
       response = ["read","write","delete"].includes(action)

task_executor.py

@@ -276,6 +276,15 @@ class ExecutionState:
     def touch(self, *, progress_pct=None, current_node=None, checkpoint_seq=None, checkpoint_stage=None, ds_calls=None, s2_calls=None, oa_calls=None, activity_only=False):
         with self._lock:
             self._last_activity = time.monotonic()
+            # Request counters are observational metrics, not task-liveness
+            # authority. They may advance on activity-only callbacks while
+            # progress/node/checkpoint ownership remains unchanged.
+            if ds_calls is not None:
+                self._ds_calls = max(self._ds_calls, int(ds_calls))
+            if s2_calls is not None:
+                self._s2_calls = max(self._s2_calls, int(s2_calls))
+            if oa_calls is not None:
+                self._oa_calls = max(self._oa_calls, int(oa_calls))
             if not activity_only:
                 if progress_pct is not None:
                     self._progress_pct = max(self._progress_pct, max(0, min(100, int(progress_pct))))
@@ -284,12 +293,6 @@ class ExecutionState:
                 if checkpoint_seq is not None and int(checkpoint_seq) >= self._checkpoint_seq:
                     self._checkpoint_seq = int(checkpoint_seq)
                     self._checkpoint_stage = str(checkpoint_stage or self._checkpoint_stage)[:120]
-                if ds_calls is not None:
-                    self._ds_calls = max(self._ds_calls, int(ds_calls))
-                if s2_calls is not None:
-                    self._s2_calls = max(self._s2_calls, int(s2_calls))
-                if oa_calls is not None:
-                    self._oa_calls = max(self._oa_calls, int(oa_calls))
 
     def cancelled(self) -> bool:
         with self._lock: