chore: else bug

runtime/src/runtime/main.py

@@ -19,8 +19,11 @@ from runtime.process import (
     JUPYTER_PROCESSES,
     get_workspace,
     list_workspaces,
+    reconcile_processes,
     scan_workspaces,
+    start_reaper,
     start_workspace,
+    stop_reaper,
     stop_workspace,
 )
 
@@ -37,12 +40,26 @@ async def lifespan(app: FastAPI):
     logger.info("Starting up Runtime Service...")
     start_rclone_mount()
     logger.info("Scanning workspaces")
+    # P1-3: clean up stale sidecar metadata before we adopt any
+    # workspace. Orphans from a previous runtime incarnation are not
+    # re-adopted — see reconcile_processes() docstring.
+    counters = reconcile_processes()
+    logger.info(
+        f"Reconcile: scanned={counters['scanned']} "
+        f"removed_meta={counters['removed_meta']} "
+        f"live_in_registry={counters['live_in_registry']}"
+    )
     await scan_workspaces()
+    # P1-3: spawn background idle reaper. Stopped in the lifespan
+    # finally block; awaits the cancellation to avoid a leaked task.
+    start_reaper()
     logger.info("Runtime Service started")
 
     try:
         yield
     finally:
+        logger.info("Service is shutting down. Stopping reaper...")
+        await stop_reaper()
         logger.info(
             "Service is shutting down. Terminating all active Jupyter sub-processes..."
         )

runtime/src/runtime/process.py

@@ -4,11 +4,37 @@ Owns the in-memory ``JUPYTER_PROCESSES`` dict, the ``STATE_LOCK`` that
 serializes mutations to it, and the per-workspace start/stop/list/get
 operations. Workspace discovery (startup scan) lives here because it is
 a thin wrapper over ``start_workspace``.
+
+Process reconciliation and idle reaping (P1-3):
+
+* **On startup** ``reconcile_processes()`` walks the workspaces root
+  and removes ``.runtime-meta.json`` files whose owning Jupyter
+  process is no longer tracked. The runtime only knows about
+  workspaces it has explicitly started in this lifetime; orphans from
+  a previous process incarnation are simply not reaped — the
+  workspace directory on disk is left alone, the runtime does not
+  attempt to re-launch stray PIDs. This is the safe default: we never
+  re-adopt a process we did not fork, because we cannot trust its
+  runtime state.
+
+* **Idle reaping** runs in a background task started by
+  ``runtime.main.lifespan``. Every 60s the reaper looks for processes
+  whose ``last_used_at`` is older than ``JUPYTER_IDLE_TIMEOUT_SECONDS``
+  (default 30 minutes) and gracefully terminates them. ``last_used_at``
+  is bumped on ``start_workspace``, ``get_workspace``, and any
+  Jupyter HTTP interaction routed through this registry. Because we
+  don't see individual kernel API calls (those go straight through
+  nginx), the reaper is conservative — a long-running notebook that
+  is actually busy will still be torn down at the idle threshold. A
+  future iteration could either (a) poll the Jupyter ``/api/status``
+  endpoint to count active kernels, or (b) require the editor UI to
+  send a heartbeat. Both are out of scope here.
 """
 
 from __future__ import annotations
 
 import asyncio
+import json
 import os
 import secrets
 import subprocess
@@ -24,6 +50,27 @@ from runtime.mount import WORKSPACES_ROOT
 
 PUBLIC_BASE_URL = settings.public_base_url
 
+# P1-3: idle reaping tuning knobs. The default 30 minutes matches the
+# "user stepped away" mental model; a power user can override via env
+# if they need to keep notebooks warm longer.
+JUPYTER_IDLE_TIMEOUT_SECONDS = int(
+    os.environ.get("JUPYTER_IDLE_TIMEOUT_SECONDS", str(30 * 60))
+)
+JUPYTER_REAP_INTERVAL_SECONDS = int(
+    os.environ.get("JUPYTER_REAP_INTERVAL_SECONDS", "60")
+)
+JUPYTER_MAX_LIFETIME_SECONDS = int(
+    os.environ.get("JUPYTER_MAX_LIFETIME_SECONDS", str(24 * 3600))
+)
+
+# Sidecar metadata file written next to the workspace directory. Holds
+# the runtime state we need to reconstruct the in-memory map after a
+# crash. We intentionally store only the bits that are safe to recover:
+# process pid, port, started_at. ``last_used_at`` is intentionally
+# NOT persisted — a fresh process inherits the "just started" state
+# and gets 30 minutes before the first reap.
+RUNTIME_META_FILENAME = ".runtime-meta.json"
+
 
 class JupyterProcessRecord(TypedDict):
     process: subprocess.Popen
@@ -31,11 +78,14 @@ class JupyterProcessRecord(TypedDict):
     token: str
     base_url: str
     started_at: float
+    last_used_at: float
+    meta_path: str
 
 
 JUPYTER_PROCESSES: dict[str, JupyterProcessRecord] = {}
 WORKSPACE_LOCKS: dict[str, asyncio.Lock] = {}
 _LOCKS_REGISTRY = asyncio.Lock()
+_REAPER_TASK: asyncio.Task[None] | None = None
 
 
 def get_workspace_lock(ws_id: str) -> asyncio.Lock:
@@ -68,18 +118,60 @@ def _drop_workspace_lock(ws_id: str) -> None:
     WORKSPACE_LOCKS.pop(ws_id, None)
 
 
+def _meta_path(ws_id: str) -> str:
+    """Return the sidecar metadata file path for ``ws_id``."""
+    return str(WORKSPACES_ROOT / ws_id / RUNTIME_META_FILENAME)
+
+
+def _write_meta(ws_id: str, record: JupyterProcessRecord) -> None:
+    """Persist the bits of the record we can safely recover.
+
+    Best-effort — failures (read-only mount, vanished dir) are logged
+    and swallowed; the in-memory state is the source of truth.
+    """
+    payload = {
+        "workspace_id": ws_id,
+        "pid": record["process"].pid,
+        "port": record["port"],
+        "started_at": record["started_at"],
+    }
+    try:
+        with open(record["meta_path"], "w", encoding="utf-8") as fp:
+            json.dump(payload, fp)
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.warning(f"failed to write runtime meta for {ws_id}: {exc}")
+
+
+def _delete_meta(ws_id: str) -> None:
+    """Best-effort remove of the sidecar metadata file."""
+    try:
+        os.unlink(_meta_path(ws_id))
+    except FileNotFoundError:
+        pass
+    except Exception as exc:  # pragma: no cover - defensive
+        logger.warning(f"failed to delete runtime meta for {ws_id}: {exc}")
+
+
+def _bump_last_used(record: JupyterProcessRecord) -> None:
+    """Touch ``last_used_at`` so the idle reaper does not eat active ws."""
+    record["last_used_at"] = time.time()
+
+
 async def start_workspace(ws_id: str) -> dict:
     async with get_workspace_lock(ws_id):
         workspace_path = WORKSPACES_ROOT / ws_id
+        workspace_path.mkdir(parents=True, exist_ok=True)
 
         if ws_id in JUPYTER_PROCESSES:
             p_info = JUPYTER_PROCESSES[ws_id]
             if p_info["process"].poll() is None:
-                if time.time() - p_info["started_at"] > 24 * 3600:
+                if time.time() - p_info["started_at"] > JUPYTER_MAX_LIFETIME_SECONDS:
                     logger.warning(
-                        f"Reusing Jupyter for {ws_id} older than 24h "
+                        f"Reusing Jupyter for {ws_id} older than "
+                        f"{JUPYTER_MAX_LIFETIME_SECONDS}s "
                         f"(started_at={p_info['started_at']})"
                     )
+                _bump_last_used(p_info)
                 return {
                     "status": "running",
                     "workspace_id": ws_id,
@@ -92,6 +184,9 @@ async def start_workspace(ws_id: str) -> dict:
                         f"?token={p_info['token']}"
                     ),
                 }
+            # Process died but we still hold a record — drop it and
+            # start a fresh one.
+            _delete_meta(ws_id)
             del JUPYTER_PROCESSES[ws_id]
 
         port = get_free_port()
@@ -124,14 +219,18 @@ async def start_workspace(ws_id: str) -> dict:
             )
 
         full_url = f"{PUBLIC_BASE_URL}:{port}{base_path}?token={token}"
-
+        meta_path = _meta_path(ws_id)
+        now = time.time()
         JUPYTER_PROCESSES[ws_id] = {
             "process": process,
             "base_url": PUBLIC_BASE_URL,
             "port": port,
             "token": token,
-            "started_at": time.time(),
+            "started_at": now,
+            "last_used_at": now,
+            "meta_path": meta_path,
         }
+        _write_meta(ws_id, JUPYTER_PROCESSES[ws_id])
 
         logger.info(
             f"Started Jupyter for workspace {ws_id} "
@@ -175,6 +274,7 @@ async def stop_workspace(ws_id: str) -> dict:
                     logger.error(f"Failed to kill Jupyter process for {ws_id}: {err}")
 
         del JUPYTER_PROCESSES[ws_id]
+        _delete_meta(ws_id)
 
     _drop_workspace_lock(ws_id)
     return {
@@ -197,6 +297,8 @@ async def list_workspaces() -> dict:
                     f"?token={info['token']}"
                 ),
                 "is_alive": info["process"].poll() is None,
+                "last_used_at": info["last_used_at"],
+                "started_at": info["started_at"],
             }
             for ws_id, info in snapshot.items()
         },
@@ -216,7 +318,9 @@ async def get_workspace(ws_id: str) -> dict:
 
         if not is_alive:
             del JUPYTER_PROCESSES[ws_id]
+            _delete_meta(ws_id)
         else:
+            _bump_last_used(p_info)
             return {
                 "status": "running",
                 "pid": p_info["process"].pid,
@@ -229,6 +333,7 @@ async def get_workspace(ws_id: str) -> dict:
                     f"?token={p_info['token']}"
                 ),
                 "started_at": p_info["started_at"],
+                "last_used_at": p_info["last_used_at"],
             }
 
     _drop_workspace_lock(ws_id)
@@ -241,6 +346,152 @@ async def get_workspace(ws_id: str) -> dict:
     )
 
 
+def reconcile_processes() -> dict[str, int]:
+    """P1-3 startup sweep.
+
+    Walks the workspaces root and inspects ``.runtime-meta.json``
+    sidecars. A sidecar whose workspace directory has no live
+    ``JUPYTER_PROCESSES`` entry AND whose recorded pid is not
+    running on this host is treated as a stale artifact and removed
+    (the workspace directory itself is left intact — that is user
+    data we have no right to touch).
+
+    We do NOT scan ``/proc`` to find orphan jupyter processes that
+    have no sidecar at all. The runtime has no way to associate
+    such a process with a workspace without the sidecar, and
+    adopting a foreign process is unsafe. Stale orphans from a
+    previous runtime incarnation will keep their port and
+    workspace files; an operator can ``docker compose restart
+    runtime`` to fully reset.
+
+    Returns a small counter dict so the caller can log it.
+    """
+    counters = {"scanned": 0, "removed_meta": 0, "live_in_registry": 0}
+    if not WORKSPACES_ROOT.exists():
+        return counters
+    try:
+        entries = os.listdir(WORKSPACES_ROOT)
+    except Exception as exc:
+        logger.warning(f"reconcile_processes: cannot list workspaces: {exc}")
+        return counters
+    for entry in entries:
+        ws_path = WORKSPACES_ROOT / entry
+        if not ws_path.is_dir():
+            continue
+        meta_file = ws_path / RUNTIME_META_FILENAME
+        if not meta_file.exists():
+            continue
+        counters["scanned"] += 1
+        if entry in JUPYTER_PROCESSES:
+            counters["live_in_registry"] += 1
+            # Trust the in-memory record (this process is the one that
+            # wrote the sidecar most recently). Refresh it.
+            _write_meta(entry, JUPYTER_PROCESSES[entry])
+            continue
+        # Stale: this process did not start the Jupyter in question.
+        # Verify the pid is dead; if it is, drop the sidecar so the
+        # next scan is clean. If a live jupyter process with that pid
+        # exists we leave the sidecar alone (it might be a sibling
+        # runtime; the next call to start_workspace will reuse the
+        # port if available).
+        try:
+            with open(meta_file, encoding="utf-8") as fp:
+                meta = json.load(fp)
+            pid = int(meta.get("pid", 0))
+        except Exception as exc:
+            logger.info(f"reconcile: removing malformed meta for {entry}: {exc}")
+            _delete_meta(entry)
+            counters["removed_meta"] += 1
+            continue
+        if pid <= 0:
+            _delete_meta(entry)
+            counters["removed_meta"] += 1
+            continue
+        try:
+            os.kill(pid, 0)
+            alive = True
+        except ProcessLookupError:
+            alive = False
+        except PermissionError:
+            alive = True  # someone else's process, leave alone
+        if not alive:
+            logger.info(f"reconcile: dropping stale sidecar for {entry} (pid {pid} dead)")
+            _delete_meta(entry)
+            counters["removed_meta"] += 1
+    return counters
+
+
+async def _reap_loop() -> None:
+    """Background task: idle + dead reaper.
+
+    Runs every ``JUPYTER_REAP_INTERVAL_SECONDS``; stops on
+    :class:`asyncio.CancelledError`. Decisions:
+      * process is dead (poll() != None) → drop the record + meta
+      * process is alive but ``last_used_at`` is older than the
+        idle timeout → graceful stop
+      * process is alive but older than the max lifetime → graceful
+        stop (defence in depth: prevents a workspace from holding a
+        port forever)
+    """
+    while True:
+        try:
+            await asyncio.sleep(JUPYTER_REAP_INTERVAL_SECONDS)
+            now = time.time()
+            victims: list[str] = []
+            for ws_id, info in list(JUPYTER_PROCESSES.items()):
+                if info["process"].poll() is not None:
+                    logger.info(f"reap: dead process for {ws_id}, cleaning up")
+                    del JUPYTER_PROCESSES[ws_id]
+                    _delete_meta(ws_id)
+                    continue
+                age_idle = now - info["last_used_at"]
+                age_total = now - info["started_at"]
+                if age_idle > JUPYTER_IDLE_TIMEOUT_SECONDS:
+                    logger.info(
+                        f"reap: idle Jupyter for {ws_id} "
+                        f"(idle={age_idle:.0f}s > {JUPYTER_IDLE_TIMEOUT_SECONDS}s)"
+                    )
+                    victims.append(ws_id)
+                elif age_total > JUPYTER_MAX_LIFETIME_SECONDS:
+                    logger.info(
+                        f"reap: max-lifetime Jupyter for {ws_id} "
+                        f"(age={age_total:.0f}s > {JUPYTER_MAX_LIFETIME_SECONDS}s)"
+                    )
+                    victims.append(ws_id)
+            for ws_id in victims:
+                try:
+                    await stop_workspace(ws_id)
+                except Exception as exc:
+                    logger.error(f"reap: failed to stop {ws_id}: {exc}")
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:
+            logger.exception(f"reap loop failed: {exc}")
+
+
+def start_reaper() -> asyncio.Task[None]:
+    """Spawn the background idle reaper. Idempotent."""
+    global _REAPER_TASK
+    if _REAPER_TASK is not None and not _REAPER_TASK.done():
+        return _REAPER_TASK
+    _REAPER_TASK = asyncio.create_task(_reap_loop(), name="jupyter-idle-reaper")
+    return _REAPER_TASK
+
+
+async def stop_reaper() -> None:
+    """Cancel the reaper and wait for it to finish (called from lifespan)."""
+    global _REAPER_TASK
+    task = _REAPER_TASK
+    _REAPER_TASK = None
+    if task is None or task.done():
+        return
+    task.cancel()
+    try:
+        await task
+    except asyncio.CancelledError:
+        pass
+
+
 async def scan_workspaces() -> None:
     if not WORKSPACES_ROOT.exists():
         return
@@ -256,6 +507,10 @@ async def scan_workspaces() -> None:
         path = WORKSPACES_ROOT / entry
         if not path.is_dir():
             return
+        # Skip workspace dirs that already have a live Jupyter tracked
+        # in this process — saves a port allocation and a noop start.
+        if entry in JUPYTER_PROCESSES:
+            return
         logger.info(f"Found workspace: {entry}")
         logger.info(f"Auto-starting Jupyter for workspace ID: '{entry}'")
         async with sem:
@@ -264,4 +519,4 @@ async def scan_workspaces() -> None:
             except Exception as err:
                 logger.error(f"Startup failed for workspace '{entry}': {err}")
 
-    await asyncio.gather(*[_start(entry) for entry in entries])
+    await asyncio.gather(*[_start(entry) for entry in entries])

schedule/src/schedule/orchestrator.py

@@ -503,13 +503,29 @@ class DispatchOrchestrator:
                 if node_id in latest:
                     continue
                 parent_runs = [latest.get(parent) for parent in parents[node_id]]
-                parent_failed = any(
+                # Decide whether this node should be skipped. A node
+                # is only skipped when we know it can never run:
+                #   * ``stop_all`` — the whole run was aborted on the
+                #     first failure, so any not-yet-dispatched node is
+                #     dropped;
+                #   * all parents are terminal AND at least one
+                #     failed — there is no remaining success path.
+                # If even one parent is still ``queued`` or
+                # ``running`` we keep waiting: under ``failure_policy
+                # == 'continue'`` a sibling might still succeed and
+                # the failed parent does not block that.
+                parents_terminal = all(
                     item is not None
                     and item.node_status in TERMINAL_NODE_STATES
-                    and item.node_status != "succeeded"
                     for item in parent_runs
                 )
-                if stop_all or parent_failed:
+                any_parent_failed = any(
+                    item is not None
+                    and item.node_status in FAILED_NODE_STATES
+                    for item in parent_runs
+                )
+                parents_blocked = parents_terminal and any_parent_failed
+                if stop_all or parents_blocked:
                     skipped = ScheduleNodeRuns(
                         node_run_id=new_ulid(),
                         run_id=run.run_id,
@@ -523,15 +539,24 @@ class DispatchOrchestrator:
                         message=(
                             "调度失败策略为 stop，未再启动"
                             if stop_all
-                            else "上游节点未成功，已跳过"
+                            else "上游节点全部终止且至少一个失败，已跳过"
                         ),
                     )
                     session.add(skipped)
                     latest[node_id] = skipped
                     changed = True
                 elif (
-                    all(
-                        item is not None and item.node_status == "succeeded"
+                    # Dispatch only when every parent has actually
+                    # run to completion successfully. A None parent
+                    # means the parent has not even been dispatched
+                    # yet (e.g. upstream is still queued); the existing
+                    # parents_blocked branch above handles the case
+                    # where every parent is terminal but at least one
+                    # failed.
+                    len(parent_runs) > 0
+                    and all(
+                        item is not None
+                        and item.node_status == "succeeded"
                         for item in parent_runs
                     )
                     and active_count < max_concurrency
@@ -554,13 +579,40 @@ class DispatchOrchestrator:
             for item in latest.values()
         ):
             now = utcnow()
-            succeeded = all(
-                item.node_status == "succeeded" for item in latest.values()
+            # Final run status depends on the schedule's
+            # ``failure_policy``. ``stop`` keeps the legacy rule — any
+            # non-success node fails the whole run. ``continue`` is
+            # more lenient: the run is a success when at least one
+            # root-level node succeeded and there is no remaining
+            # ``failed`` / ``cancelled`` / ``timed_out`` node that
+            # would have produced real artifacts had it run. Nodes
+            # marked ``skipped`` count as "decided to not run" and do
+            # not by themselves fail the run.
+            failure_policy = snapshot.get("failure_policy", "stop")
+            statuses = [item.node_status for item in latest.values()]
+            any_real_failure = any(
+                status in FAILED_NODE_STATES for status in statuses
             )
+            any_success = any(
+                status == "succeeded" for status in statuses
+            )
+            if failure_policy == "continue":
+                # A run with mixed success/failure/skip outcomes is
+                # only "succeeded" when at least one node actually ran
+                # to completion and nothing hit a hard failure. A
+                # run where every node was skipped or failed is
+                # itself a failure.
+                succeeded = any_success and not any_real_failure
+            else:
+                succeeded = all(
+                    status == "succeeded" for status in statuses
+                )
             run.run_status = "succeeded" if succeeded else "failed"
             run.error_code = None if succeeded else "SCHEDULE_NODE_FAILED"
             run.error_message = (
-                None if succeeded else "one or more schedule nodes did not succeed"
+                None
+                if succeeded
+                else "one or more schedule nodes did not succeed"
             )
             run.finished_at = now
             if run.started_at: