feat: robust support for remote job re-attachment

docs/remote_job_monitoring.md

@@ -101,6 +101,40 @@ to the agent even when a monitor probe updates the job while that command runs.
 Strict revision checks remain in place for lifecycle transitions and provider
 reconciliation, where accepting stale state would be unsafe.
 
+## Executor Timeout and Remote-Job Handoff
+
+A step executor is a bounded LLM session; a remote job is durable. The two have
+independent lifetimes, so a step executor is never kept alive merely to babysit
+a running job.
+
+When `SUB_STEP_TIMEOUT` (default 3600s) elapses, the runner checks the durable
+job store for a job still owned by that node:
+
+- **No active job** — the step times out as before and returns
+  `needs_replanning`.
+- **An active job** — the executor is granted a single bounded grace window
+  (`STEP_REMOTE_JOB_GRACE_TIMEOUT`, default 300s) to let a nearly finished step
+  complete. If it is still unfinished afterwards, the executor is released and
+  the step returns `waiting` rather than `needs_replanning`. This is a handoff,
+  not a failure: dependents are not blocked, and the job keeps running with no
+  executor attached.
+
+The runner writes `status: waiting` and the job identity onto the execution
+graph node itself, so the handoff does not depend on the orchestrator LLM
+calling `set_node_status`. `reconcile_recovery_state` then keeps the node
+`waiting` while the job is still in progress and moves it back to `pending`
+once the job settles, so a fresh executor can collect its results. The identical
+path also covers a crashed or restarted executor, so there is one recovery
+mechanism rather than two.
+
+Re-attachment is explicit rather than accidental. When a node that already owns
+a job runs again, the runner injects the job's identity into the executor's
+`prior_context` with instructions to call `get_e2b_job_status` and never call
+`submit_e2b_sandbox` for that step. In Flash mode, which has no execution graph,
+a step's node ID is derived from its label or a hash of its action, so a repeated
+step keeps the same submission idempotency key and re-attaches instead of
+creating a duplicate sandbox.
+
 ## Controls and Ownership
 
 The middleware exposes owner-scoped controls:

src/matcreator/agents/execution_agent/agent.py

@@ -59,15 +59,22 @@
        a. Call `set_node_status(node_id="...", status="failed", result=<replan_reason>)`.
        b. Call `mark_dependents_blocked(failed_node_id="...")`.
        c. Call `to_planner(reason=<replan_reason>)`. STOP — do not run any further nodes.
+   - `status == "waiting"`: the executor timed out but its remote job is STILL RUNNING.
+       This is NOT a failure. The node status and its `remote_job` are already recorded.
+       a. Do NOT call `set_node_status`, `mark_dependents_blocked`, or re-run the node.
+       b. Do NOT start a replacement job for it.
+       c. Continue processing the other results in this batch, then follow step 6.
    - `status == "cancelled"`:
        a. Call `to_planner("execution cancelled by user")`. STOP.
 5. After handling all results from a batch, call `get_ready_nodes()` again.
-6. If it returns count == 0 and any graph node has status `waiting`, call
+6. If it returns count == 0 and `waiting_nodes` is non-empty, call
     `to_planner(reason="waiting for remote job completion")`. STOP.
 7. Execution is complete only when every graph node has status `success`.
 
 ## Rules
 - NEVER execute code directly — all work goes through `run_node_executor`.
+- A ready node that carries a `remote_job` was resumed from a remote-job handoff. Run it
+  normally: its executor re-attaches to that job automatically and must never resubmit it.
 - When a batch has both successes and one failure, process the success nodes first
   (validate_summarize + set_node_status), then handle the failure last (set_node_status,
   mark_dependents_blocked, to_planner).

src/matcreator/agents/execution_agent/recovery.py

@@ -14,6 +14,23 @@
 _RECOVERY_DIR = "recovery"
 _STALE_AFTER_SECONDS = int(os.environ.get("STEP_RECOVERY_STALE_AFTER", "60"))
 
+# A job in one of these states is still doing work on the provider, so no
+# executor can make progress on it yet.  ``succeeded``/``collecting`` are
+# deliberately excluded: those mean results are ready to be collected, which is
+# exactly the point at which a node should run again.
+_REMOTE_JOB_IN_PROGRESS_STATUSES = frozenset(
+    {
+        "created",
+        "submitting",
+        "queued",
+        "running",
+        "pause_requested",
+        "paused",
+        "resume_requested",
+        "resuming",
+    }
+)
+
 
 def _now() -> str:
     return datetime.now(timezone.utc).isoformat()
@@ -231,7 +248,25 @@ def _mark_attempt_stale(latest_path: Path, attempt: dict[str, Any]) -> None:
     _write_attempt(attempt)
 
 
-def _active_remote_job(attempt: dict[str, Any]) -> dict[str, Any] | None:
+def _refresh_remote_job_reference(attempt: dict[str, Any]) -> None:
+    """Reload the persisted remote-job identity into an in-memory attempt.
+
+    ``record_remote_job_reference`` writes through the durable record, so a live
+    attempt dict only learns about a submitted job on its next heartbeat.
+    """
+    latest_path = attempt.get("_latest_path")
+    if not latest_path:
+        return
+    persisted = _read_json(Path(latest_path))
+    if isinstance(persisted.get("remote_job"), dict):
+        attempt["remote_job"] = persisted["remote_job"]
+
+
+def _active_remote_job(
+    attempt: dict[str, Any],
+    *,
+    statuses: frozenset[str] = ACTIVE_REMOTE_JOB_STATUSES,
+) -> dict[str, Any] | None:
     reference = attempt.get("remote_job")
     if not isinstance(reference, dict) or not reference.get("job_id"):
         return None
@@ -243,12 +278,34 @@ def _active_remote_job(attempt: dict[str, Any]) -> dict[str, Any] | None:
         job
         and job.get("session_id") == attempt.get("session_id")
         and job.get("node_id") == attempt.get("node_id")
-        and job.get("status") in ACTIVE_REMOTE_JOB_STATUSES
+        and job.get("status") in statuses
     ):
         return job
     return None
 
 
+def remote_job_reference(job: dict[str, Any]) -> dict[str, Any]:
+    """Return the identity subset of a remote job stored on a graph node."""
+    return {
+        "job_id": job["job_id"],
+        "provider": job["provider"],
+        "external_id": job["external_id"],
+        "status": job["status"],
+    }
+
+
+def active_remote_job_for_attempt(attempt: dict[str, Any]) -> dict[str, Any] | None:
+    """Return the still-active remote job owned by a live attempt, if any.
+
+    Used by the step executor when its wall-clock timeout fires: a timeout that
+    leaves a tracked remote job running is a handoff, not a failure.
+    """
+    if not attempt:
+        return None
+    _refresh_remote_job_reference(attempt)
+    return _active_remote_job(attempt)
+
+
 def reconcile_recovery_state(
     state: Any,
     workspace_dir: str | Path,
@@ -299,12 +356,7 @@ def reconcile_recovery_state(
             if remote_job is not None:
                 node["status"] = "waiting"
                 node["result"] = "Recovered active remote job; waiting for provider completion."
-                node["remote_job"] = {
-                    "job_id": remote_job["job_id"],
-                    "provider": remote_job["provider"],
-                    "external_id": remote_job["external_id"],
-                    "status": remote_job["status"],
-                }
+                node["remote_job"] = remote_job_reference(remote_job)
                 node["recovery"] = {
                     "attempt": attempt.get("attempt"),
                     "status": "waiting_remote",
@@ -323,6 +375,45 @@ def reconcile_recovery_state(
             actions.append({"node_id": node_id, "action": "reset_stale_running", "status": "pending"})
             continue
 
+        # A step executor that timed out while its remote job was still running
+        # hands the node off as ``waiting`` rather than failing it.  Keep it
+        # waiting until the provider settles, then let the node run again so a
+        # fresh executor can re-attach and collect the results.  Skip nodes the
+        # planner has already resumed, otherwise the retry would be undone here.
+        resumed_by_planner = (
+            node_status == "pending"
+            and isinstance(node.get("recovery"), dict)
+            and node["recovery"].get("status") == "waiting_remote"
+        )
+        if (
+            attempt_status == "waiting"
+            and node_status in ("pending", "running", "waiting")
+            and not resumed_by_planner
+        ):
+            remote_job = _active_remote_job(
+                attempt, statuses=_REMOTE_JOB_IN_PROGRESS_STATUSES
+            )
+            if remote_job is not None:
+                node["status"] = "waiting"
+                node["result"] = _attempt_summary(attempt)
+                node["remote_job"] = remote_job_reference(remote_job)
+                node["recovery"] = {
+                    "attempt": attempt.get("attempt"),
+                    "status": "waiting_remote",
+                    "recovered_at": _now(),
+                }
+                actions.append({"node_id": node_id, "action": "wait_for_remote_job", "status": "waiting"})
+            else:
+                node["status"] = "pending"
+                node["result"] = "Remote job is no longer in progress; re-running node to collect its results."
+                node["recovery"] = {
+                    "attempt": attempt.get("attempt"),
+                    "status": "remote_job_settled",
+                    "recovered_at": _now(),
+                }
+                actions.append({"node_id": node_id, "action": "resume_after_remote_job", "status": "pending"})
+            continue
+
         if node_status in ("pending", "running") and attempt_status in {
             "success",
             "failed",

src/matcreator/agents/execution_agent/step_executor.py

@@ -135,6 +135,15 @@ def _fill_missing_fields(self) -> "StepExecutorResult":
 3. **If submission.json exists but outputs are missing**, reuse the same submission file (dpdispatcher is idempotent — it skips completed tasks). Do NOT regenerate submission.json.
 4. **Never resubmit a job that already completed** — this wastes GPU time and creates duplicate training runs.
 
+## Re-attaching to an existing E2B job (CRITICAL)
+If your `prior_context` contains "REMOTE JOB ALREADY SUBMITTED", a tracked sandbox job
+for this exact step is already running:
+1. Call `get_e2b_job_status` with the given `job_id` FIRST.
+2. NEVER call `submit_e2b_sandbox` for that step — it would duplicate a running job.
+3. If the job finished, collect its outputs with `download_e2b_output` and report success.
+4. If it is still running, call `submit_step_result(status="needs_replanning", ...)` stating
+   that the job has not finished yet and quoting its job_id and status.
+
 ## User controls for E2B sandboxes
 `get_e2b_job_status` may return `user_control` when the user paused or terminated
 the sandbox from the UI. This does not cancel your executor. Treat it as the