Step 4A-6-C-R3.1.1 Executor Claim Transport Idempotency and UI Monitor Hotfix

app.py

@@ -596,6 +596,81 @@ _user_workspace = os.path.join(WORKSPACE_DIR, "users", _safe_key(authenticated_u
 os.makedirs(_user_workspace, exist_ok=True)
 
 
+ACTIVE_TASK_STATUSES = {"created", "running", "resumable", "cancelling"}
+AUTO_MONITOR_TASK_STATUSES = {"created", "running", "cancelling"}
+
+def _render_monitor_box(task: dict, lang_code: str) -> None:
+    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}")
+    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>
+        <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>
+    </div>
+    """, unsafe_allow_html=True)
+
+
+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
+    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"}:
+        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__)
+    st.session_state["sp_last_ui_task_status"] = status
+    _render_monitor_box(task, lang_code)
+    if status == "completed":
+        output_path = _task_output_path(task)
+        if output_path and output_path.is_file():
+            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")
+        else:
+            st.warning("The task completed, but the local output file is unavailable after runtime replacement.")
+    elif status == "failed":
+        st.error(str(task.get("safe_error_detail") or task.get("safe_error_code") or tr(lang_code, "node_error")))
+    elif status == "cancelled":
+        detail = str(task.get("safe_error_detail") or "")
+        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.
+    if previous_status and status != previous_status and status in {"resumable", "completed", "failed", "cancelled"}:
+        st.rerun()
+
+
+def _render_task_monitor(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
+    task_status = str(task.get("status") or "")
+    monitor_run_every = 5 if task_status in AUTO_MONITOR_TASK_STATUSES else None
+    fragment = getattr(st, "fragment", None)
+    if fragment is not None:
+        @fragment(run_every=monitor_run_every)
+        def _task_monitor_fragment():
+            _render_task_status_once(lang_code)
+        _task_monitor_fragment()
+    else:
+        _render_task_status_once(lang_code)
+
 def render_workspace_controls(user):
     top_toolbar = st.container()
     with top_toolbar:
@@ -615,11 +690,16 @@ def render_workspace_controls(user):
 
     task = st.session_state.get("sp_generation_task") or {}
     task_status = str(task.get("status") or "")
-    active = task_status in {"created", "running", "resumable", "cancelling"}
+    active = task_status in ACTIVE_TASK_STATUSES
     st.session_state["is_running"] = active
 
     st.markdown(f"<div class='main-title'><h2>{tr(lang_code, 'title')}</h2></div>", unsafe_allow_html=True)
 
+    # R3.1.1: the live task monitor sits directly below the title so a full
+    # rerun after pressing Generate lands on the progress area, not above a
+    # long static form. Only this small monitor auto-refreshes while running.
+    _render_task_monitor(lang_code)
+
     if task_status == "resumable":
         st.warning(tr(lang_code, "checkpoint_warning"))
         colA, colB = st.columns([1, 1])
@@ -692,66 +772,3 @@ if st.session_state.get("sp_payment_required"):
     render_payment_purchase(lang_code)
 
 
-def _render_monitor_box(task: dict) -> None:
-    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}")
-    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>
-        <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>
-    </div>
-    """, unsafe_allow_html=True)
-
-
-def _render_task_status_once() -> None:
-    task = st.session_state.get("sp_generation_task") or {}
-    task_id = str(task.get("task_id") or "").strip()
-    if not task_id:
-        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"}:
-        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__)
-    st.session_state["sp_last_ui_task_status"] = status
-    _render_monitor_box(task)
-    if status == "completed":
-        output_path = _task_output_path(task)
-        if output_path and output_path.is_file():
-            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")
-        else:
-            st.warning("The task completed, but the local output file is unavailable after runtime replacement.")
-    elif status == "failed":
-        st.error(str(task.get("safe_error_detail") or task.get("safe_error_code") or tr(lang_code, "node_error")))
-    elif status == "cancelled":
-        detail = str(task.get("safe_error_detail") or "")
-        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.
-    if previous_status and status != previous_status and status in {"resumable", "completed", "failed", "cancelled"}:
-        st.rerun()
-
-
-if getattr(st, "fragment", None) is not None:
-    @st.fragment(run_every=5)
-    def _task_monitor_fragment():
-        _render_task_status_once()
-    _task_monitor_fragment()
-else:
-    _render_task_status_once()

task_executor.py

@@ -15,6 +15,7 @@ import json
 import logging
 import os
 import re
+import secrets
 import sys
 import threading
 import time
@@ -42,6 +43,14 @@ logging.basicConfig(
 
 
 CLAIM_HANDSHAKE_FILENAME = "executor_claim.json"
+EXECUTOR_CLAIM_ATTEMPTS = 3
+EXECUTOR_CLAIM_TIMEOUT_SECONDS = 4
+EXECUTOR_CLAIM_BACKOFF_SECONDS = 0.6
+
+
+def _new_executor_instance_id() -> str:
+    """Create one stable identity for this executor claim attempt sequence."""
+    return f"exec_{secrets.token_urlsafe(18)}"
 
 
 def _safe(value, limit=500):
@@ -135,11 +144,34 @@ class ExecutorApi:
             raise ExecutorApiError(error, response.status_code, data)
         return data
 
-    def claim(self, *, task_id: str, user_id: int, request_hash: str) -> dict:
-        return self.post(
-            "/executor/task/claim",
-            {"task_id": task_id, "user_id": user_id, "request_hash": request_hash},
-        )
+    def claim(self, *, task_id: str, user_id: int, request_hash: str, executor_instance_id: str) -> dict:
+        payload = {
+            "task_id": task_id,
+            "user_id": user_id,
+            "request_hash": request_hash,
+            "executor_instance_id": executor_instance_id,
+        }
+        for attempt in range(1, EXECUTOR_CLAIM_ATTEMPTS + 1):
+            try:
+                return self.post(
+                    "/executor/task/claim",
+                    payload,
+                    timeout=EXECUTOR_CLAIM_TIMEOUT_SECONDS,
+                )
+            except (
+                requests.exceptions.SSLError,
+                requests.exceptions.Timeout,
+                requests.exceptions.ConnectionError,
+            ) as exc:
+                if attempt >= EXECUTOR_CLAIM_ATTEMPTS:
+                    raise
+                delay = EXECUTOR_CLAIM_BACKOFF_SECONDS * (2 ** (attempt - 1))
+                logging.warning(
+                    "[SP_EXECUTOR] claim_transport_retry task_id=%s attempt=%s/%s error=%s delay=%.1fs",
+                    task_id, attempt, EXECUTOR_CLAIM_ATTEMPTS, type(exc).__name__, delay,
+                )
+                time.sleep(delay)
+        raise RuntimeError("executor_claim_retry_exhausted")
 
     def heartbeat(self, payload: dict) -> dict:
         return self.post("/executor/task/heartbeat", payload)
@@ -272,13 +304,21 @@ def _bootstrap_claim(*, task_dir: Path, task_id: str, user_id: int, request_data
     request_hash = str(request_data.get("request_hash") or "").strip()
     if not request_hash:
         raise RuntimeError("request_hash_missing")
-    claim = api.claim(task_id=task_id, user_id=user_id, request_hash=request_hash)
+    instance_id = _new_executor_instance_id()
+    claim = api.claim(
+        task_id=task_id,
+        user_id=user_id,
+        request_hash=request_hash,
+        executor_instance_id=instance_id,
+    )
     task = dict(claim.get("task") or {})
     execution = dict(claim.get("execution") or {})
     epoch = int(execution.get("execution_epoch") or task.get("execution_epoch") or 0)
-    instance_id = str(execution.get("executor_instance_id") or "")
-    if epoch <= 0 or not instance_id:
+    claimed_instance_id = str(execution.get("executor_instance_id") or "")
+    if epoch <= 0 or not claimed_instance_id:
         raise RuntimeError("executor_claim_identity_missing")
+    if claimed_instance_id != instance_id:
+        raise RuntimeError("executor_claim_identity_mismatch")
     if int(task.get("target_words") or 0) != int(request_data.get("target_words") or 0):
         raise RuntimeError("target_words_claim_mismatch")
     try: