Step 4A-6-C-R3.2 Executor Ownership Correction and UI Rollback (#10)

- Step 4A-6-C-R3.2 Executor Ownership Correction and UI Rollback (3fd0f77d70e84a8faaf668c4f5fb0e9e9915cb59)

app.py

@@ -7,7 +7,6 @@ import os
 import re
 import secrets
 import shutil
-import signal
 import subprocess
 import sys
 import time
@@ -396,12 +395,13 @@ def _pid_is_task_executor(pid: int, task_id: str, task_dir: Path) -> bool:
 
 
 def _launch_executor(task_id: str, *, force: bool = False) -> int:
-    """Launch only after a PID-bound claim handshake is observed.
+    """Start a detached executor without making Streamlit its lifecycle owner.
 
-    ``subprocess.Popen`` proves only that the OS accepted a child process. R3.1
-    waits until that exact child has successfully claimed the D1 task and
-    written ``executor_claim.json``. A child that exits/import-fails before
-    claim is therefore never reported as a running executor.
+    ``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"
@@ -419,25 +419,15 @@ def _launch_executor(task_id: str, *, force: bool = False) -> int:
         try:
             marker = json.loads(launch_marker_path.read_text(encoding="utf-8"))
             pid = int(marker.get("pid") or 0)
-            verified_executor = _pid_is_task_executor(pid, task_id, task_dir)
-            if verified_executor and not force:
-                claim_marker = {}
-                if claim_marker_path.is_file():
-                    try:
-                        claim_marker = json.loads(claim_marker_path.read_text(encoding="utf-8"))
-                    except Exception:
-                        claim_marker = {}
-                if int(claim_marker.get("pid") or 0) == pid and int(claim_marker.get("execution_epoch") or 0) > 0:
-                    return pid
-            if verified_executor and force:
-                try:
-                    os.kill(pid, signal.SIGTERM)
-                except OSError:
-                    pass
+            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 claim file must never satisfy a new launch/resume handshake.
+    # 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:
@@ -465,42 +455,17 @@ def _launch_executor(task_id: str, *, force: bool = False) -> int:
         start_new_session=True,
         env=executor_env,
     )
-    _atomic_json(launch_marker_path, {"pid": process.pid, "launched_at": datetime.now(timezone.utc).isoformat()})
+    _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)
-
-    timeout_seconds = float(APP_CONFIG.get("executor_bootstrap_timeout_seconds", 20))
-    poll_seconds = max(0.05, float(APP_CONFIG.get("executor_bootstrap_poll_ms", 100)) / 1000.0)
-    deadline = time.monotonic() + timeout_seconds
-    while time.monotonic() < deadline:
-        if claim_marker_path.is_file():
-            try:
-                claim_marker = json.loads(claim_marker_path.read_text(encoding="utf-8"))
-            except Exception:
-                claim_marker = {}
-            if (
-                int(claim_marker.get("pid") or 0) == int(process.pid)
-                and str(claim_marker.get("task_id") or "") == str(task_id)
-                and str(claim_marker.get("request_hash") or "") == request_hash
-                and int(claim_marker.get("execution_epoch") or 0) > 0
-                and str(claim_marker.get("executor_instance_id") or "").strip()
-            ):
-                logging.info(
-                    "[SP_UI] executor_claim_confirmed task_id=%s pid=%s epoch=%s",
-                    task_id, process.pid, claim_marker.get("execution_epoch"),
-                )
-                return int(process.pid)
-        return_code = process.poll()
-        if return_code is not None:
-            logging.error("[SP_UI] executor_bootstrap_exit task_id=%s pid=%s rc=%s", task_id, process.pid, return_code)
-            raise RuntimeError(f"executor_bootstrap_failed_exit_{return_code}")
-        time.sleep(poll_seconds)
-
-    try:
-        process.terminate()
-    except Exception:
-        pass
-    logging.error("[SP_UI] executor_claim_timeout task_id=%s pid=%s", task_id, process.pid)
-    raise RuntimeError("executor_claim_timeout")
+    return int(process.pid)
 
 
 def _process_pending_cancel() -> None:
@@ -529,6 +494,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.rerun()
 
 
@@ -575,6 +541,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.rerun()
 
 
@@ -597,7 +564,6 @@ 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 "")
@@ -655,21 +621,27 @@ def _render_task_status_once(lang_code: str) -> None:
         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:
+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")):
         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)
+    st.session_state["sp_scroll_to_task_monitor"] = False
+    components.html(
+        """
+        <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);
+        })();
+        </script>
+        """,
+        height=0,
+        width=0,
+    )
+
 
 def render_workspace_controls(user):
     top_toolbar = st.container()
@@ -695,10 +667,6 @@ def render_workspace_controls(user):
 
     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"))
@@ -771,4 +739,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.
+_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()
+
 

task_executor.py

@@ -43,9 +43,31 @@ logging.basicConfig(
 
 
 CLAIM_HANDSHAKE_FILENAME = "executor_claim.json"
-EXECUTOR_CLAIM_ATTEMPTS = 3
-EXECUTOR_CLAIM_TIMEOUT_SECONDS = 4
-EXECUTOR_CLAIM_BACKOFF_SECONDS = 0.6
+
+
+def _bounded_env_float(name: str, default: float, minimum: float, maximum: float) -> float:
+    try:
+        value = float(os.environ.get(name, str(default)))
+    except (TypeError, ValueError):
+        value = float(default)
+    return max(float(minimum), min(float(maximum), value))
+
+
+# The current Worker startup lease is 90 seconds. R3.2 deliberately keeps the
+# default executor claim window below that lease while allowing materially
+# slower HF->Cloudflare TLS handshakes than R3.1.1's 4-second timeout.
+EXECUTOR_CLAIM_WINDOW_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_WINDOW_SECONDS", 70.0, 20.0, 80.0
+)
+EXECUTOR_CLAIM_REQUEST_TIMEOUT_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_REQUEST_TIMEOUT_SECONDS", 18.0, 5.0, 30.0
+)
+EXECUTOR_CLAIM_BACKOFF_INITIAL_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_BACKOFF_INITIAL_SECONDS", 1.0, 0.2, 5.0
+)
+EXECUTOR_CLAIM_BACKOFF_MAX_SECONDS = _bounded_env_float(
+    "SMARTPAPER_EXECUTOR_CLAIM_BACKOFF_MAX_SECONDS", 4.0, 1.0, 10.0
+)
 
 
 def _new_executor_instance_id() -> str:
@@ -151,27 +173,40 @@ class ExecutorApi:
             "request_hash": request_hash,
             "executor_instance_id": executor_instance_id,
         }
-        for attempt in range(1, EXECUTOR_CLAIM_ATTEMPTS + 1):
+        deadline = time.monotonic() + EXECUTOR_CLAIM_WINDOW_SECONDS
+        attempt = 0
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                raise RuntimeError("executor_claim_deadline_exhausted")
+            attempt += 1
+            request_timeout = max(1.0, min(EXECUTOR_CLAIM_REQUEST_TIMEOUT_SECONDS, remaining))
             try:
                 return self.post(
                     "/executor/task/claim",
                     payload,
-                    timeout=EXECUTOR_CLAIM_TIMEOUT_SECONDS,
+                    timeout=request_timeout,
                 )
             except (
                 requests.exceptions.SSLError,
                 requests.exceptions.Timeout,
                 requests.exceptions.ConnectionError,
             ) as exc:
-                if attempt >= EXECUTOR_CLAIM_ATTEMPTS:
+                remaining = deadline - time.monotonic()
+                if remaining <= 0:
                     raise
-                delay = EXECUTOR_CLAIM_BACKOFF_SECONDS * (2 ** (attempt - 1))
+                delay = min(
+                    EXECUTOR_CLAIM_BACKOFF_MAX_SECONDS,
+                    EXECUTOR_CLAIM_BACKOFF_INITIAL_SECONDS * (2 ** max(0, attempt - 1)),
+                    max(0.0, remaining - 0.25),
+                )
                 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,
+                    "[SP_EXECUTOR] claim_transport_retry task_id=%s attempt=%s error=%s delay=%.1fs remaining=%.1fs",
+                    task_id, attempt, type(exc).__name__, delay, remaining,
                 )
+                if delay <= 0:
+                    raise
                 time.sleep(delay)
-        raise RuntimeError("executor_claim_retry_exhausted")
 
     def heartbeat(self, payload: dict) -> dict:
         return self.post("/executor/task/heartbeat", payload)