diff --git a/docs/en/configuration/experiments.md b/docs/en/configuration/experiments.md
index ccce37de..fe927239 100644
--- a/docs/en/configuration/experiments.md
+++ b/docs/en/configuration/experiments.md
@@ -38,6 +38,51 @@ tasks:
   graph_node_classification: {}
 ```
 
+## Explicit continuation to a finite timeout
+
+The default auto-resume policy stops after at most 100 resumes. A long run
+whose Agent repeatedly exits successfully can reach that ceiling before its
+time budget. To use the timeout as the ceiling for successful exits, opt in:
+
+```bash
+sforge run --task <task-id> --agent codex --timeout 64800 --resume-until-timeout
+```
+
+The Python `run_agent` API accepts `resume_until_timeout=True`. The mode
+requires an explicit positive integer `timeout` argument or `agent_timeout`
+configuration and an Agent resume command. It conflicts with
+`--disable-auto-resume`. The same container, registration and Agent resume
+command are retained; submission limits, cooldown and scoring are unchanged.
+
+One monotonic deadline starts after initial command preparation at the first
+solver dispatch. Initial preparation and final collection remain separate
+costs. Later command preparation and time between segments consume the same
+deadline; resumes never renew it. Stop and remaining time are checked again
+after preparation, immediately before dispatch. Cancellation, any nonzero process exit, an Agent-owned
+terminal blocker, a timed-out segment, or a segment shorter than one second
+can still end work early. This mode does **not** guarantee a minimum duration.
+Keep any time deficit and failure evidence instead of treating an early exit
+as a completed budget. Backend timeout drain can add collection cost after
+solver work stops.
+
+Stock Agents inherit no additional classifier for successful process output.
+A quota message with exit code zero requires an explicit Agent
+`resume_block_reason` implementation to block continuation; this flag does not
+infer that condition from arbitrary text.
+
+The opt-in run writes `resume-policy.json`; the CLI's `final_result.json` and
+Python result include the same `resume_policy`. It records the original
+deadline, actual resumes, process time, solver-phase elapsed time, full
+invocation elapsed time, stop reason and unused seconds. No policy field or
+file is added to default runs. To disable the mode, omit the flag or pass
+`False` in Python.
+
+This is in-process continuation, not checkpoint restoration. An existing
+`started_at` or policy receipt rejects a new opted-in invocation in that run
+directory. Preserve the old attempt; a cold restart needs a new run identity
+and remains a separate experiment. Existing frozen runs are not upgraded or
+relabelled by enabling this flag on a later run.
+
 ## Fields
 
 ### `env`
diff --git a/docs/zh/configuration/experiments.md b/docs/zh/configuration/experiments.md
index 1f918ba7..154419fc 100644
--- a/docs/zh/configuration/experiments.md
+++ b/docs/zh/configuration/experiments.md
@@ -38,6 +38,38 @@ tasks:
   graph_node_classification: {}
 ```
 
+## 显式按有限超时继续原会话
+
+默认自动续跑最多100次。长任务若反复正常退出，可能在时间预算尚未耗尽时触及上限。
+需要以有限超时约束正常退出后的续跑时，显式启用：
+
+```bash
+sforge run --task <task-id> --agent codex --timeout 64800 --resume-until-timeout
+```
+
+Python `run_agent` 接受 `resume_until_timeout=True`。该模式要求显式传入正整数
+`timeout` 或配置 `agent_timeout`，并且Agent有原生续跑命令；与
+`--disable-auto-resume` 冲突。容器、registration和Agent续跑命令保持同一条链，
+提交次数限制、冷却与评分不变。
+
+第一次命令准备完成后，在solver dispatch边界确定唯一的monotonic截止时间。
+首次准备与最后收集分别计成本，后续命令准备和段间开销消耗同一截止时间；续跑不会
+重新计时。准备完成、实际dispatch前再次检查Stop和剩余时间。Stop、非零退出、Agent明确的终态阻塞、
+单段超时及不足一秒的短段仍会提前结束，因此这**不是最低时长保证**。提前结束时须
+保留欠时和失败证据。后端的超时收尾可能在solver停止后继续消耗收集时间。
+
+默认Agent未新增对成功进程输出的quota解析；退出码为零的quota信息如需阻止续跑，
+须由Agent的 `resume_block_reason` 明确提供，此flag不会从任意文本推断该条件。
+
+显式模式写入 `resume-policy.json`；CLI的 `final_result.json` 和Python结果包含
+相同的 `resume_policy`，记录原截止时间、实际续跑数、进程时长、solver阶段时长、
+本次完整调用时长、停止原因和未用秒数。默认运行不增加该字段或文件。省略flag或
+在Python传入 `False` 即关闭。
+
+该模式只在同一次调用内续跑，不负责checkpoint恢复。已有 `started_at` 或policy
+receipt的目录拒绝再次以该模式启动。保留原attempt；冷启动需要新的run身份，独立
+计为另一实验。后继启用新flag不会升级或重标已有冻结运行。
+
 ## 字段说明
 
 ### `env`
diff --git a/sforge/cli.py b/sforge/cli.py
index 171b40dc..1b6d35db 100644
--- a/sforge/cli.py
+++ b/sforge/cli.py
@@ -246,6 +246,7 @@ def _run_single_task(
         disable_stop_hook=disable_stop_hook,
         disable_auto_eval=disable_auto_eval,
         disable_auto_resume=disable_auto_resume,
+        resume_until_timeout=getattr(args, "resume_until_timeout", False),
         internet=internet,
         verbose=verbose,
         shutdown_event=shutdown_event,
@@ -886,8 +887,12 @@ def main():
                        dest="disable_auto_eval", help="Disable background auto-evaluation daemon")
     p_run.add_argument("--disable-stop-hook", action="store_true", default=False,
                        dest="disable_stop_hook", help="Disable the agent stop hook (agent can exit normally)")
-    p_run.add_argument("--disable-auto-resume", action="store_true", default=False,
+    resume_group = p_run.add_mutually_exclusive_group()
+    resume_group.add_argument("--disable-auto-resume", action="store_true", default=False,
                        dest="disable_auto_resume", help="Disable auto-resume on abnormal agent exit")
+    resume_group.add_argument("--resume-until-timeout", action="store_true", default=False,
+                       help="Resume successful exits in the same session until the explicit timeout; "
+                            "failure, stop and short-segment guards still apply (default: max 100 resumes)")
     p_run.add_argument("--max-submissions", type=int, default=None, dest="max_submissions",
                        help="Maximum number of agent submissions per run (default: unlimited)")
     p_run.add_argument("--submission-cooldown", type=int, default=None, dest="submission_cooldown",
diff --git a/sforge/harness/agent/base.py b/sforge/harness/agent/base.py
index e28c004e..e14c7e03 100644
--- a/sforge/harness/agent/base.py
+++ b/sforge/harness/agent/base.py
@@ -22,6 +22,7 @@
 
 from sforge.harness.agent.bundle import ensure_bundle, inject_bundle
 from sforge.harness.backend import ContainerBackend, ContainerHandle
+from sforge.harness.backend.base import StreamingExecResult
 from sforge.harness.config import SForgeConfig
 
 
@@ -91,6 +92,15 @@ def augment_env(self, env: dict[str, str], model: str | None) -> None:
         mirrors, extra_env) have already been applied.
         """
 
+    def resume_block_reason(self, result: StreamingExecResult) -> str | None:
+        """An Agent-owned terminal blocker for explicit deadline continuation.
+
+        Inspect only the current invocation. The runner also stops on any
+        nonzero exit, cancellation, timeout or short segment in that mode.
+        Returning None does not authorize a new session or a renewed budget.
+        """
+        return None
+
     def format_run_cmd(
         self,
         prompt_path: str,
diff --git a/sforge/harness/run_agent.py b/sforge/harness/run_agent.py
index f703b35f..e5708775 100644
--- a/sforge/harness/run_agent.py
+++ b/sforge/harness/run_agent.py
@@ -83,6 +83,7 @@ class RunResult:
     runtime_seconds: float = 0.0
     resume_count: int = 0
     usage: dict | None = None  # LLM gateway totals (tokens, requests, by model)
+    resume_policy: dict | None = None  # Present only for the explicit timeout policy.
 
     def to_dict(self) -> dict:
         d = {
@@ -100,6 +101,8 @@ def to_dict(self) -> dict:
             d["best_score"] = self.best_score
         if self.usage is not None:
             d["usage"] = self.usage
+        if self.resume_policy is not None:
+            d["resume_policy"] = self.resume_policy
         return d
 
 
@@ -408,6 +411,7 @@ def run_agent(
     shutdown_event: threading.Event | None = None,
     max_submissions: int | None = None,
     submission_cooldown: int | None = None,
+    resume_until_timeout: bool = False,
 ) -> RunResult:
     """Run an agent on a task with iterative evaluation.
 
@@ -419,11 +423,40 @@ def run_agent(
     3. Install agent + tools (submit script, stop hook, auto-eval daemon)
     4. Run agent (high max-turns for iterative work)
     5. Read final state, extract archive
+
+    ``resume_until_timeout`` explicitly replaces the 100-resume ceiling for
+    successful exits with one finite solver-phase deadline. It requires an
+    explicit timeout (argument or config), the same Agent's resume command,
+    and a fresh run directory. Setup and collection costs remain separate;
+    time between segments consumes this deadline. Failure, cancellation and
+    short-segment guards can still end the run early. This is not checkpoint
+    or cross-process session restoration.
     """
+    run_started = time.monotonic()
+    if type(resume_until_timeout) is not bool:
+        raise ValueError("resume_until_timeout must be a boolean")
     effective_timeout = timeout or config.agent_timeout or agent.timeout
+    if resume_until_timeout:
+        declared_timeout = timeout if timeout is not None else config.agent_timeout
+        if type(declared_timeout) is not int or declared_timeout < 1:
+            raise ValueError("resume_until_timeout requires an explicit positive integer timeout")
+        if disable_auto_resume or not isinstance(agent.resume_cmd, str) or not agent.resume_cmd.strip():
+            raise ValueError("resume_until_timeout requires enabled auto-resume and an Agent resume command")
+        effective_timeout = declared_timeout
 
     log_dir = config.log_dir / "runs" / run_id / task_spec.task_id
     log_dir.mkdir(parents=True, exist_ok=True)
+    resume_policy = None
+    policy_path = log_dir / "resume-policy.json"
+    if resume_until_timeout:
+        if (log_dir / "started_at").exists():
+            raise ValueError("resume_until_timeout requires a fresh run; retain the original attempt")
+        resume_policy = {"mode": "until_timeout", "timeout_seconds": effective_timeout,
+                         "status": "starting", "stop_reason": "not_started"}
+        # An uncertain prior invocation must not restart its deadline or claim
+        # a cold session is an in-process continuation.
+        with policy_path.open("x") as stream:
+            json.dump(resume_policy, stream)
     logger = setup_logger(
         f"agent.{task_spec.task_id}.{run_id}",
         log_dir / "run_agent.log",
@@ -447,8 +480,14 @@ def run_agent(
     auto_eval_stop = None
     session_stop = threading.Event()
     admin_secret = ""
+    phase_started = phase_finished = None
+    solver_dispatch_active = False
+    stop_reason = "runner_error"
 
     try:
+        if resume_until_timeout and shutdown_event is not None and shutdown_event.is_set():
+            stop_reason = "shutdown_requested"
+            return RunResult(resume_policy=resume_policy)
         admin_secret = get_admin_secret(config.log_dir)
         # 0. Clean up stale iptables chains from previous runs that were killed
         if backend.backend_name == "docker":
@@ -714,24 +753,67 @@ def run_agent(
         started_at = time.time()
         from datetime import datetime
         started_at_iso = datetime.fromtimestamp(started_at).strftime("%Y-%m-%dT%H:%M:%S")
-        (log_dir / "started_at").write_text(f"{started_at_iso}\n{started_at}\n")
+        if not resume_until_timeout:
+            (log_dir / "started_at").write_text(f"{started_at_iso}\n{started_at}\n")
 
         on_chunk_cb = None
 
         while remaining_timeout > 0:
-            is_resume = resume_count > 0
+            if resume_until_timeout:
+                if shutdown_event is not None and shutdown_event.is_set():
+                    stop_reason = "shutdown_requested"
+                    break
+                if phase_started is not None:
+                    remaining_timeout = phase_deadline - time.monotonic()
+                    if remaining_timeout < 1:
+                        agent_timed_out = True
+                        stop_reason = "budget_exhausted"
+                        break
+            is_resume = bool(all_output_parts) if resume_until_timeout else resume_count > 0
             run_cmd = agent.format_run_cmd(
                 prompt_path, model=model,
                 internet=internet, resume=is_resume,
             )
             if is_resume:
+                limit = "deadline" if resume_until_timeout else MAX_RESUMES
+                attempt = resume_count + 1 if resume_until_timeout else resume_count
                 logger.info(
-                    f"Resuming agent (attempt {resume_count}/{MAX_RESUMES}, "
+                    f"Resuming agent (attempt {attempt}/{limit}, "
                     f"{remaining_timeout:.0f}s left): {run_cmd[:200]}..."
                 )
             else:
                 logger.info(f"Running agent: {run_cmd[:200]}... (timeout={remaining_timeout:.0f}s)")
 
+            if resume_until_timeout:
+                # Formatting can prepare a session or context and take real
+                # time. Initial formatting is setup; later formatting consumes
+                # the one deadline. Never dispatch with its earlier remainder.
+                if shutdown_event is not None and shutdown_event.is_set():
+                    stop_reason = "shutdown_requested"
+                    break
+                if phase_started is None:
+                    phase_started = time.monotonic()
+                    started_at = time.time()
+                    phase_deadline = phase_started + effective_timeout
+                    started_at_iso = datetime.fromtimestamp(started_at).strftime("%Y-%m-%dT%H:%M:%S")
+                    (log_dir / "started_at").write_text(f"{started_at_iso}\n{started_at}\n")
+                    resume_policy.update(status="running", started_at_unix=started_at,
+                                         deadline_unix=started_at + effective_timeout)
+                    policy_path.write_text(json.dumps(resume_policy, indent=2) + "\n")
+                if shutdown_event is not None and shutdown_event.is_set():
+                    stop_reason = "shutdown_requested"
+                    break
+                remaining_timeout = phase_deadline - time.monotonic()
+                # Backends take whole seconds; a fractional remainder cannot
+                # launch another process, even after slow preparation or IO.
+                if remaining_timeout < 1:
+                    agent_timed_out = True
+                    stop_reason = "budget_exhausted"
+                    break
+                if is_resume:
+                    resume_count += 1
+            if resume_until_timeout:
+                solver_dispatch_active = True
             seg_result = backend.exec_run_with_timeout(
                 handle,
                 ["/bin/bash", "-c", run_cmd],
@@ -744,31 +826,53 @@ def run_agent(
                 log_append=is_resume,
                 on_chunk=on_chunk_cb,
             )
+            if resume_until_timeout:
+                solver_dispatch_active = False
             all_output_parts.append(seg_result.output)
             total_runtime += seg_result.elapsed_seconds
 
+            if resume_until_timeout and shutdown_event is not None and shutdown_event.is_set():
+                stop_reason = "shutdown_requested"
+                break
             if seg_result.timed_out:
                 agent_timed_out = True
+                stop_reason = "segment_timeout"
                 break
 
             if not can_resume:
+                stop_reason = "auto_resume_disabled"
                 break
+            if resume_until_timeout:
+                # A provider may already expose a typed terminal blocker. Do
+                # not infer one from arbitrary prose or retry failed processes
+                # without the legacy attempt ceiling.
+                blocked = agent.resume_block_reason(seg_result)
+                if blocked is not None:
+                    stop_reason = "provider_blocked"
+                    logger.warning("Agent resume blocked by provider: %s", blocked)
+                    break
+                if seg_result.exit_code != 0:
+                    stop_reason = "process_failed"
+                    break
             if seg_result.elapsed_seconds < MIN_RUNTIME_FOR_RESUME:
+                stop_reason = "short_segment"
                 logger.warning(
                     f"Agent exited after only {seg_result.elapsed_seconds:.1f}s "
                     f"(< {MIN_RUNTIME_FOR_RESUME}s), not resuming (likely systematic failure)"
                 )
                 break
-            if resume_count >= MAX_RESUMES:
+            if not resume_until_timeout and resume_count >= MAX_RESUMES:
                 logger.warning(f"Max resume attempts ({MAX_RESUMES}) reached")
                 break
 
-            remaining_timeout -= seg_result.elapsed_seconds
-            resume_count += 1
+            if not resume_until_timeout:
+                remaining_timeout -= seg_result.elapsed_seconds
+                resume_count += 1
             logger.info(f"Agent exited after {seg_result.elapsed_seconds:.1f}s, will resume")
 
         agent_output = "\n".join(all_output_parts)
         runtime = total_runtime
+        phase_finished = time.monotonic()
         logger.info(
             f"Agent finished: runtime={runtime:.1f}s, timed_out={agent_timed_out}, "
             f"resumes={resume_count}"
@@ -843,6 +947,7 @@ def run_agent(
                 timed_out=agent_timed_out,
                 runtime_seconds=runtime,
                 resume_count=resume_count,
+                resume_policy=resume_policy,
             )
             logger.info(
                 f"Done (game): best_score={best_score} ({best_round}), "
@@ -919,6 +1024,7 @@ def run_agent(
                 timed_out=agent_timed_out,
                 runtime_seconds=runtime,
                 resume_count=resume_count,
+                resume_policy=resume_policy,
             )
 
             logger.info(
@@ -941,6 +1047,9 @@ def run_agent(
         return result
 
     except KeyboardInterrupt:
+        if phase_started is not None and phase_finished is None and not solver_dispatch_active:
+            phase_finished = time.monotonic()
+        stop_reason = "interrupted"
         logger.info("Run interrupted by user (Ctrl+C)")
 
         # Stop auto-eval thread
@@ -1010,8 +1119,12 @@ def run_agent(
             agent_output="Stopped by user (Ctrl+C)",
             timed_out=False,
             resume_count=resume_count,
+            runtime_seconds=locals().get("total_runtime", 0.0) if resume_until_timeout else 0.0,
+            resume_policy=resume_policy,
         )
     except Exception as e:
+        if phase_started is not None and phase_finished is None and not solver_dispatch_active:
+            phase_finished = time.monotonic()
         logger.error(f"Error: {e}\n{traceback.format_exc()}")
         if backend.backend_name == "e2b":
             # E2B transport or lifecycle failures must remain visible to the
@@ -1049,8 +1162,12 @@ def run_agent(
             agent_output=str(e),
             timed_out=False,
             resume_count=locals().get("resume_count", 0),
+            runtime_seconds=locals().get("total_runtime", 0.0) if resume_until_timeout else 0.0,
+            resume_policy=resume_policy,
         )
     finally:
+        if phase_started is not None and phase_finished is None:
+            phase_finished = time.monotonic()
         session_stop.set()
         if auto_eval_stop is not None:
             auto_eval_stop.set()
@@ -1067,4 +1184,16 @@ def run_agent(
             signal.signal(signal.SIGINT, prev_handler)
         else:
             backend.cleanup_container(handle, logger)
+        if resume_policy is not None:
+            elapsed = 0.0 if phase_started is None else phase_finished - phase_started
+            resume_policy.update(status="terminal", stop_reason=stop_reason,
+                phase_elapsed_seconds=elapsed,
+                remaining_seconds=max(0.0, effective_timeout - elapsed),
+                process_runtime_seconds=locals().get("total_runtime", 0.0),
+                total_elapsed_seconds=time.monotonic() - run_started,
+                resume_count=locals().get("resume_count", 0))
+            try:
+                policy_path.write_text(json.dumps(resume_policy, indent=2) + "\n")
+            except OSError as exc:
+                logger.warning("Could not persist resume policy receipt: %s", type(exc).__name__)
         close_logger(logger)
diff --git a/tests/test_resume_until_timeout.py b/tests/test_resume_until_timeout.py
new file mode 100644
index 00000000..0c4fca2b
--- /dev/null
+++ b/tests/test_resume_until_timeout.py
@@ -0,0 +1,372 @@
+# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""The explicit resume policy through the production runner and CLI adapter."""
+from __future__ import annotations
+
+import importlib
+import json
+import threading
+from types import SimpleNamespace
+
+import pytest
+
+from sforge.harness.agent.base import Agent
+from sforge.harness.backend.base import ExecResult, StreamingExecResult
+from sforge.harness.config import SForgeConfig
+
+runner = importlib.import_module("sforge.harness.run_agent")
+
+
+class ProbeAgent(Agent):
+    name = "probe"
+    npm_package = "synthetic-probe"
+    default_version = "fixture"
+    api_key_env = "SYNTHETIC_UNUSED_KEY"
+    run_cmd = "initial {prompt_file}"
+    resume_cmd = "resume-same-session"
+    blocked = None
+
+    def install(self, *args):
+        pass
+
+    def resume_block_reason(self, result):
+        return self.blocked
+
+
+class ProbeBackend:
+    backend_name = "fixture"
+    public_dns = ()
+    uses_hostname_egress = True
+
+    def __init__(self, clock):
+        self.clock = clock
+        self.calls = []
+        self.creates = 0
+        self.duration = 1.0
+        self.overhead = 0.0
+        self.exit_code = 0
+        self.after_exec = lambda: None
+
+    def image_exists(self, image):
+        return True
+
+    def remove_container_by_name(self, *args):
+        pass
+
+    def create_container(self, *args, **kwargs):
+        self.creates += 1
+        return SimpleNamespace(id="same-worker")
+
+    def start_container(self, handle):
+        pass
+
+    def cleanup_container(self, *args):
+        pass
+
+    def copy_to_container(self, *args):
+        pass
+
+    def exec_run(self, *args, **kwargs):
+        return ExecResult()
+
+    def exec_run_with_timeout(self, handle, cmd, **kwargs):
+        self.calls.append((handle.id, cmd, kwargs["timeout"], kwargs["log_append"]))
+        self.clock.now += self.duration + self.overhead
+        self.after_exec()
+        return StreamingExecResult("successful fixture", self.exit_code, False, self.duration)
+
+
+@pytest.fixture
+def execution(tmp_path, monkeypatch):
+    clock = SimpleNamespace(now=0.0)
+    monkeypatch.setattr(runner, "time", SimpleNamespace(
+        monotonic=lambda: clock.now, time=lambda: 1800000000 + clock.now))
+    backend = ProbeBackend(clock)
+    config = SForgeConfig(log_dir=tmp_path)
+    agent = ProbeAgent(config)
+    task = SimpleNamespace(task_id="toy", game_mode=False, work_image_key="work",
+        judge_image_key="judge", work=SimpleNamespace(cpu_limit=1, mem_limit="128m", instruction="Write answer.txt"),
+        submit_paths=["answer.txt"], cwd="/work", internet=True)
+    registrations = []
+
+    class Response:
+        def __init__(self, payload):
+            self.payload = payload
+
+        def raise_for_status(self):
+            pass
+
+        def json(self):
+            return self.payload
+
+    def post(url, **kwargs):
+        registrations.append(kwargs["json"])
+        return Response({"token": "synthetic-registration"})
+
+    monkeypatch.setattr(runner.requests, "post", post)
+    monkeypatch.setattr(runner.requests, "get", lambda *a, **k: Response({"entries": [], "agent_submissions": 0, "auto_submissions": 0}))
+    monkeypatch.setattr(runner, "_install_tools", lambda *a, **k: None)
+    monkeypatch.setattr(runner, "_extract_archive_from_container", lambda *a, **k: b"")
+    monkeypatch.setattr(runner, "_sync_agent_sessions", lambda *a, **k: None)
+    monkeypatch.setattr(runner, "_session_sync_loop", lambda *a, **k: None)
+
+    def run(**kwargs):
+        return runner.run_agent(task, agent, config, backend, "probe", timeout=130,
+            disable_auto_eval=True, disable_stop_hook=True, **kwargs)
+
+    return SimpleNamespace(run=run, backend=backend, agent=agent, config=config,
+        task=task, clock=clock, registrations=registrations,
+        trial=tmp_path/"runs/probe/toy")
+
+
+def test_opt_in_crosses_100_without_replacing_worker_or_session(execution):
+    x = execution
+    result = x.run(resume_until_timeout=True, max_submissions=7, submission_cooldown=3600)
+    assert len(x.backend.calls) == 130
+    assert result.resume_count == 129
+    assert x.backend.creates == len(x.registrations) == 1
+    assert x.registrations[0]["submission_cooldown"] == 3600
+    assert x.registrations[0]["max_agent_submissions"] == 7
+    assert x.backend.calls[0][1][-1].startswith("initial ")
+    assert all(c[0] == "same-worker" and c[1][-1] == "resume-same-session" and c[3] for c in x.backend.calls[1:])
+    policy = result.to_dict()["resume_policy"]
+    assert policy["stop_reason"] == "budget_exhausted"
+    assert policy["deadline_unix"] - policy["started_at_unix"] == 130
+    assert policy["remaining_seconds"] == 0
+    assert policy["phase_elapsed_seconds"] == result.runtime_seconds == 130
+    assert json.loads((x.trial/"resume-policy.json").read_text()) == policy
+
+
+def test_default_retains_100_resumes_and_wire(execution):
+    result = execution.run()
+    assert len(execution.backend.calls) == 101
+    assert result.resume_count == 100
+    assert not result.timed_out
+    assert "resume_policy" not in result.to_dict()
+    assert not (execution.trial/"resume-policy.json").exists()
+
+
+def test_deadline_charges_between_segment_overhead(execution):
+    x = execution
+    x.backend.duration, x.backend.overhead = 20, 30
+    result = x.run(resume_until_timeout=True)
+    assert [c[2] for c in x.backend.calls] == [130, 80, 30]
+    assert result.runtime_seconds == 60
+    assert result.resume_policy["phase_elapsed_seconds"] == 150
+    assert result.resume_policy["deadline_unix"] - result.resume_policy["started_at_unix"] == 130
+
+
+@pytest.mark.parametrize("kind,reason", [("stop", "shutdown_requested"), ("failure", "process_failed"), ("provider", "provider_blocked"), ("short", "short_segment")])
+def test_real_stop_and_failure_guards_remain_terminal(execution, kind, reason):
+    x = execution
+    stop = threading.Event()
+    if kind == "stop":
+        x.backend.after_exec = stop.set
+    elif kind == "failure":
+        x.backend.exit_code = 1  # Includes provider quota failures; no prose classifier.
+    elif kind == "provider":
+        x.agent.blocked = "quota_exhausted"
+    else:
+        x.backend.duration = 0.2
+    result = x.run(resume_until_timeout=True, shutdown_event=stop)
+    assert len(x.backend.calls) == 1
+    assert result.resume_count == 0
+    assert result.resume_policy["stop_reason"] == reason
+    assert result.resume_policy["remaining_seconds"] > 0
+
+
+def test_preexisting_stop_dispatches_nothing(execution):
+    stop = threading.Event()
+    stop.set()
+    result = execution.run(resume_until_timeout=True, shutdown_event=stop)
+    assert not execution.backend.calls
+    assert not execution.backend.creates and not execution.registrations
+    assert result.resume_policy["stop_reason"] == "shutdown_requested"
+
+
+@pytest.mark.parametrize("timeout", [None, 0, -1, True, 1.5, float("inf")])
+def test_invalid_or_implicit_budget_rejected_before_registration(execution, timeout):
+    x = execution
+    with pytest.raises(ValueError, match="explicit positive integer"):
+        runner.run_agent(x.task, x.agent, x.config, x.backend, "invalid", timeout=timeout, resume_until_timeout=True)
+    assert not x.registrations and not x.backend.creates
+
+
+def test_disabled_resume_cannot_be_overridden(execution):
+    with pytest.raises(ValueError, match="enabled auto-resume"):
+        execution.run(resume_until_timeout=True, disable_auto_resume=True)
+    assert not execution.registrations
+
+
+@pytest.mark.parametrize("command", [None, "", "  "])
+def test_missing_resume_command_cannot_fall_back_to_cold_start(execution, command):
+    execution.agent.resume_cmd = command
+    with pytest.raises(ValueError, match="Agent resume command"):
+        execution.run(resume_until_timeout=True)
+    assert not execution.registrations
+
+
+def test_explicit_config_timeout_uses_same_owner(execution):
+    x = execution
+    x.config.agent_timeout = 5
+    result = runner.run_agent(x.task, x.agent, x.config, x.backend, "probe",
+        resume_until_timeout=True, disable_auto_eval=True, disable_stop_hook=True)
+    assert len(x.backend.calls) == 5
+    assert result.resume_policy["timeout_seconds"] == 5
+
+
+def test_initial_format_is_separate_setup_cost(execution):
+    x = execution
+    original = x.agent.format_run_cmd
+
+    def format_cmd(*args, resume=False, **kwargs):
+        if not resume:
+            x.clock.now += 5
+        return original(*args, resume=resume, **kwargs)
+
+    x.agent.format_run_cmd = format_cmd
+    result = x.run(resume_until_timeout=True)
+    assert len(x.backend.calls) == 130
+    assert result.resume_policy["phase_elapsed_seconds"] == 130
+    assert result.resume_policy["total_elapsed_seconds"] == 135
+
+
+def test_resume_format_cannot_spend_stale_remaining_timeout(execution):
+    x = execution
+    original = x.agent.format_run_cmd
+
+    def format_cmd(*args, resume=False, **kwargs):
+        if resume:
+            x.clock.now += 130
+        return original(*args, resume=resume, **kwargs)
+
+    x.agent.format_run_cmd = format_cmd
+    result = x.run(resume_until_timeout=True)
+    assert len(x.backend.calls) == 1
+    assert result.resume_count == 0
+    assert result.resume_policy["stop_reason"] == "budget_exhausted"
+
+
+def test_stop_during_format_cannot_enter_backend(execution):
+    x = execution
+    stop = threading.Event()
+    original = x.agent.format_run_cmd
+
+    def format_cmd(*args, **kwargs):
+        stop.set()
+        return original(*args, **kwargs)
+
+    x.agent.format_run_cmd = format_cmd
+    result = x.run(resume_until_timeout=True, shutdown_event=stop)
+    assert not x.backend.calls
+    assert result.resume_policy["phase_elapsed_seconds"] == 0
+
+
+@pytest.mark.parametrize("error", [OSError, KeyboardInterrupt])
+def test_failed_resume_excludes_final_collection_from_phase(execution, monkeypatch, error):
+    x = execution
+    original_format = x.agent.format_run_cmd
+    original_get = runner.requests.get
+
+    def format_cmd(*args, resume=False, **kwargs):
+        if resume:
+            x.clock.now += 2
+            raise error("Synthetic resume preparation failure")
+        return original_format(*args, resume=resume, **kwargs)
+
+    def collect(*args, **kwargs):
+        x.clock.now += 10
+        return original_get(*args, **kwargs)
+
+    x.agent.format_run_cmd = format_cmd
+    monkeypatch.setattr(runner.requests, "get", collect)
+    result = x.run(resume_until_timeout=True)
+    assert len(x.backend.calls) == len(x.registrations) == 1
+    assert result.resume_policy["phase_elapsed_seconds"] == 3
+    assert result.resume_policy["remaining_seconds"] == 127
+    assert result.resume_policy["total_elapsed_seconds"] == 13
+    assert result.runtime_seconds == 1
+
+
+def test_initial_preparation_failure_keeps_phase_unstarted(execution, monkeypatch):
+    x = execution
+    original_get = runner.requests.get
+
+    def format_cmd(*args, **kwargs):
+        x.clock.now += 2
+        raise OSError("Synthetic first preparation failure")
+
+    def collect(*args, **kwargs):
+        x.clock.now += 10
+        return original_get(*args, **kwargs)
+
+    x.agent.format_run_cmd = format_cmd
+    monkeypatch.setattr(runner.requests, "get", collect)
+    result = x.run(resume_until_timeout=True)
+    assert not x.backend.calls
+    assert result.resume_policy["phase_elapsed_seconds"] == 0
+    assert result.resume_policy["remaining_seconds"] == 130
+    assert result.resume_policy["total_elapsed_seconds"] == 12
+
+
+def test_unconfirmed_exec_failure_keeps_conservative_phase_boundary(execution, monkeypatch):
+    x = execution
+    original_get = runner.requests.get
+
+    def uncertain_exec(*args, **kwargs):
+        x.clock.now += 1
+        raise OSError("Synthetic stream failure without confirmed process exit")
+
+    def collect(*args, **kwargs):
+        x.clock.now += 10
+        return original_get(*args, **kwargs)
+
+    monkeypatch.setattr(x.backend, "exec_run_with_timeout", uncertain_exec)
+    monkeypatch.setattr(runner.requests, "get", collect)
+    result = x.run(resume_until_timeout=True)
+    assert len(x.registrations) == 1
+    assert result.resume_policy["phase_elapsed_seconds"] == 11
+    assert result.resume_policy["remaining_seconds"] == 119
+    assert result.resume_policy["total_elapsed_seconds"] == 11
+
+
+def test_old_attempt_cannot_gain_new_deadline(execution):
+    x = execution
+    x.trial.mkdir(parents=True)
+    (x.trial/"started_at").write_text("1800000000\n")
+    with pytest.raises(ValueError, match="fresh run"):
+        x.run(resume_until_timeout=True)
+    assert not x.registrations
+
+
+def test_policy_receipt_prevents_uncertain_invocation_replay(execution):
+    execution.run(resume_until_timeout=True)
+    with pytest.raises((ValueError, FileExistsError)):
+        execution.run(resume_until_timeout=True)
+    assert len(execution.registrations) == 1
+
+
+def test_cli_adapter_forwards_policy_and_reads_result(execution, monkeypatch):
+    from sforge import cli
+    import sforge.harness.agent
+    x = execution
+    x.config.log_dir.joinpath("runs/probe/toy").mkdir(parents=True)
+    monkeypatch.setattr(sforge.harness.agent, "create_agent", lambda *a: x.agent)
+    args = SimpleNamespace(agent="probe", model=None, timeout=130, eval_interval=0,
+        judge_url="http://127.0.0.1:1", resume_until_timeout=True)
+    value = cli._run_single_task(x.task, args, x.config, x.backend, "probe")
+    assert value["resume_policy"]["mode"] == "until_timeout"
+    assert value["resume_count"] == 129
+    assert json.loads((x.trial/"final_result.json").read_text())["resume_policy"] == value["resume_policy"]
diff --git a/tests/test_resume_until_timeout_docker.py b/tests/test_resume_until_timeout_docker.py
new file mode 100644
index 00000000..e5b37766
--- /dev/null
+++ b/tests/test_resume_until_timeout_docker.py
@@ -0,0 +1,399 @@
+# Copyright (c) 2026 ByteDance Ltd. and/or its affiliates
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Explicit opt-in real Docker/process/TCP qualification without model calls."""
+from __future__ import annotations
+
+import importlib
+import inspect
+import io
+import json
+import os
+import shlex
+import socket
+import threading
+import time
+import uuid
+from contextlib import contextmanager
+from pathlib import Path, PurePosixPath
+from types import SimpleNamespace
+
+import docker
+import pytest
+import uvicorn
+
+from sforge import cli
+from sforge.harness import judge_server
+from sforge.harness.agent.base import Agent
+from sforge.harness.backend.docker_backend import DockerBackend
+from sforge.harness.config import SForgeConfig
+from sforge.harness.task_spec import load_task
+
+runner = importlib.import_module("sforge.harness.run_agent")
+pytestmark = pytest.mark.skipif(os.environ.get("SFORGE_RESUME_DOCKER_TEST") != "1",
+    reason="Requires explicit isolated Docker/TCP resume qualification and a cached Python image")
+
+PROBE = r'''
+import io,json,os,sys,tarfile,time,uuid
+from pathlib import Path
+from urllib.request import Request,urlopen
+from urllib.error import HTTPError
+root=Path('/work/research');root.mkdir(exist_ok=True)
+p=root/'session.json';phase=sys.argv[1]
+if phase=='initial':
+    assert not p.exists(), 'cold restart over retained session'
+    state={'session':uuid.uuid4().hex,'visits':0,'cooldown_http':None}
+else:
+    assert p.exists(), 'resume lost original session'
+    state=json.loads(p.read_text())
+state['visits']+=1
+p.write_text(json.dumps(state))
+Path('/work/answer.txt').write_text('1\n')
+def submit():
+    archive=io.BytesIO()
+    with tarfile.open(fileobj=archive,mode='w:gz') as tar:
+        info=tarfile.TarInfo('answer.txt');info.size=2;tar.addfile(info,io.BytesIO(b'1\n'))
+    boundary='fixture-'+uuid.uuid4().hex
+    body=b''
+    for key,value in [('token',os.environ['SFORGE_TOKEN']),('kind','agent')]:
+        body+=f'--{boundary}\r\nContent-Disposition: form-data; name="{key}"\r\n\r\n{value}\r\n'.encode()
+    body+=f'--{boundary}\r\nContent-Disposition: form-data; name="archive"; filename="answer.tar.gz"\r\nContent-Type: application/gzip\r\n\r\n'.encode()+archive.getvalue()+f'\r\n--{boundary}--\r\n'.encode()
+    req=Request(os.environ['SFORGE_JUDGE_URL']+'/api/v1/submit',data=body,headers={'Content-Type':'multipart/form-data; boundary='+boundary})
+    try:
+        with urlopen(req,timeout=10) as r:return r.status,json.load(r)
+    except HTTPError as e:return e.code,None
+if state['visits']==1:
+    status,response=submit();assert status==200,status
+    state['accepted_submission']=response['submission_id']
+elif state['visits']==2:
+    state['cooldown_http'],_=submit();assert state['cooldown_http']==429
+p.write_text(json.dumps(state))
+print(json.dumps({'session':state['session'],'visits':state['visits'],'phase':phase,'cooldown_http':state['cooldown_http']}),flush=True)
+time.sleep(float(os.environ.get('SYNTHETIC_SLEEP','1.05')))
+if os.environ.get('SYNTHETIC_QUOTA_FAILURE')=='1':
+    print(json.dumps({'type':'turn.failed','error':{'code':'insufficient_quota'}}),flush=True)
+    sys.exit(1)
+'''
+
+
+@contextmanager
+def serving(app):
+    sock = socket.socket()
+    sock.bind(("0.0.0.0", 0))
+    server = uvicorn.Server(uvicorn.Config(app, log_level="error", lifespan="off"))
+    thread = threading.Thread(target=server.run, kwargs={"sockets": [sock]}, daemon=True)
+    thread.start()
+    try:
+        until = time.monotonic() + 10
+        while not server.started and time.monotonic() < until:
+            time.sleep(.01)
+        assert server.started
+        yield sock.getsockname()[1]
+    finally:
+        server.should_exit = True
+        thread.join(10)
+        sock.close()
+        assert not thread.is_alive()
+
+
+class ObservedBackend(DockerBackend):
+    def __init__(self, client):
+        super().__init__(client)
+        self.created = []
+        self.processes = []
+        self.on_solver_start = None
+
+    def create_container(self, *args, **kwargs):
+        handle = super().create_container(*args, **kwargs)
+        self.created.append(handle)
+        return handle
+
+    def exec_run_with_timeout(self, handle, cmd, **kwargs):
+        is_solver = getattr(kwargs.get("log_file"), "name", None) == "agent_output.txt"
+        if is_solver and self.on_solver_start is not None:
+            callback, self.on_solver_start = self.on_solver_start, None
+            callback()
+        result = super().exec_run_with_timeout(handle, cmd, **kwargs)
+        if is_solver:
+            self.processes.append({"container_id": handle.id, "command": cmd[-1],
+                "timeout": kwargs["timeout"], "exit_code": result.exit_code,
+                "timed_out": result.timed_out, "elapsed_seconds": result.elapsed_seconds,
+                "output": result.output})
+        return result
+
+
+class ProcessAgent(Agent):
+    name = "process-probe"
+    npm_package = "synthetic-local-probe"
+    default_version = "fixture"
+    api_key_env = "SYNTHETIC_UNUSED_KEY"
+    run_cmd = "python /opt/sforge/probe.py initial"
+    resume_cmd = "python /opt/sforge/probe.py resume"
+
+    def install(self, backend, handle, logger):
+        path = self._config.log_dir/"process-probe.py"
+        path.write_text(PROBE)
+        backend.exec_run(handle, ["mkdir", "-p", "/opt/sforge", "/work"])
+        backend.copy_to_container(handle, path, PurePosixPath("/opt/sforge/probe.py"))
+
+
+@pytest.fixture
+def live(tmp_path, monkeypatch):
+    client = docker.from_env()
+    base = os.environ.get("SFORGE_TEST_DOCKER_BASE", "python:3.12-slim")
+    image = client.images.get(base)  # Cached only; never pull or install a model.
+    root = tmp_path/"benchmark"
+    task_dir = root/"tasks/toy"
+    (task_dir/"environment").mkdir(parents=True)
+    (task_dir/"judge").mkdir()
+    dockerfile = (f"FROM {base}\nRUN mkdir -p /work /opt/sforge\nWORKDIR /work\n"
+        'ENTRYPOINT ["sh","-c","trap \'exit 0\' TERM; while :; do sleep 1; done"]\n')
+    (task_dir/"environment/Dockerfile").write_text(dockerfile)
+    (task_dir/"judge/Dockerfile").write_text(dockerfile)
+    (task_dir/"instruction.md").write_text("Write answer.txt containing one.\n")
+    name = "resume-" + uuid.uuid4().hex[:12]
+    (root/"benchmark.toml").write_text(f'name = "{name}"\n')
+    grade_cmd = "python -c " + shlex.quote(
+        'import json,pathlib; pathlib.Path("/opt/sforge/result.json").write_text('
+        'json.dumps(dict(score=1,valid=True,summary="synthetic")))')
+    (task_dir/"task.toml").write_text(
+        'id = "toy"\nversion = "1.0.0"\n[environment]\ncwd = "/work"\n'
+        f'platform = "{image.attrs["Os"]}/{image.attrs["Architecture"]}"\ncpus = 1\nmemory = "128m"\n'
+        '[submission]\npaths = ["answer.txt"]\n[judge]\n'
+        f'cmd = {json.dumps(grade_cmd)}\n'
+        'timeout_sec = 10\ncpus = 1\nmemory = "128m"\n[score]\ndirection = "maximize"\n')
+    task = load_task(task_dir)
+    tags = [task.work_image_key, task.judge_image_key]
+    built, _ = client.images.build(fileobj=io.BytesIO(dockerfile.encode()),
+        tag=tags[0], rm=True, pull=False, network_mode="none")
+    built.tag(tags[1])
+    config = SForgeConfig(benchmark_dir=root, log_dir=tmp_path/"logs", judge_cpu_limit=1,
+        judge_mem_limit="128m", work_cpu_limit=1, work_mem_limit="128m")
+    backend = ObservedBackend(client)
+    states = []
+    real_state = judge_server.JudgeState
+
+    def observe_state(*args, **kwargs):
+        state = real_state(*args, **kwargs)
+        states.append(state)
+        return state
+
+    monkeypatch.setattr(judge_server, "JudgeState", observe_state)
+    # This test qualifies resume/process/HTTP, not global iptables sweeping.
+    # Never let a fixture touch network rules belonging to active experiments.
+    import sforge.harness.network_isolation
+    monkeypatch.setattr(sforge.harness.network_isolation, "cleanup_stale_chains", lambda *a, **k: None)
+    try:
+        app = judge_server.create_app(config)
+        with serving(app) as port:
+            host = os.environ.get("SFORGE_TEST_JUDGE_HOST", "host.docker.internal")
+            yield SimpleNamespace(config=config, task=task, backend=backend, states=states,
+                host_url=f"http://127.0.0.1:{port}", worker_url=f"http://{host}:{port}", name=name)
+    finally:
+        for handle in backend.created:
+            try:
+                client.containers.get(handle.id).remove(force=True)
+            except docker.errors.NotFound:
+                pass
+        for tag in tags:
+            client.images.remove(tag)
+        client.close()
+
+
+def execute(live, monkeypatch, *, timeout, shutdown_event=None, agent=None, expected_submissions=1):
+    import sforge.harness.agent
+    agent = agent or ProcessAgent(live.config)
+    monkeypatch.setattr(sforge.harness.agent, "create_agent", lambda *a: agent)
+    args = SimpleNamespace(agent=agent.name, model=None, timeout=timeout, eval_interval=0,
+        judge_url=live.worker_url, host_judge_url=live.host_url,
+        disable_auto_eval=True, disable_stop_hook=True, resume_until_timeout=True,
+        max_submissions=3, submission_cooldown=3600, enable_internet=True)
+    run = "probe-" + live.name
+    value = cli._run_single_task(live.task, args, live.config, live.backend, run,
+        shutdown_event=shutdown_event)
+    assert len(live.backend.created) == 1
+    assert len(live.states[0].tokens) == 1
+    info = next(iter(live.states[0].tokens.values()))
+    assert info["next_agent"] == expected_submissions + 1
+    assert info["submission_cooldown"] == 3600
+    assert info["max_agent_submissions"] == 3
+    trial = live.config.log_dir/"runs"/run/"toy"
+    history = json.loads((trial/"run_history.json").read_text())
+    assert history["agent_submissions"] == expected_submissions and history["auto_submissions"] == 0
+    assert all(e["valid"] is True for e in history["entries"])
+    assert len({p["container_id"] for p in live.backend.processes}) == bool(live.backend.processes)
+    assert all(p["command"] == agent.resume_cmd for p in live.backend.processes[1:])
+    events = []
+    for p in live.backend.processes:
+        for line in p["output"].splitlines():
+            try:
+                event = json.loads(line)
+            except ValueError:
+                continue
+            if "session" in event:
+                events.append(event)
+    assert len({e["session"] for e in events}) == bool(live.backend.processes)
+    assert value["usage"]["requests"] == 0
+    cooldown_statuses = sorted({e["cooldown_http"] for e in events if e.get("cooldown_http") is not None})
+    if len(live.backend.processes) > 1:
+        assert cooldown_statuses == [429]
+    summary = {"source": runner.__file__, "supports_policy": "resume_until_timeout" in inspect.signature(runner.run_agent).parameters,
+        "result": value, "actual_processes": len(live.backend.processes),
+        "session_ids": sorted({e["session"] for e in events}),
+        "one_container_and_registration": True, "next_agent": info["next_agent"],
+        "submission_cooldown": info["submission_cooldown"], "cooldown_http_statuses": cooldown_statuses,
+        "model_calls": value["usage"]["requests"]}
+    evidence = os.environ.get("SFORGE_RESUME_TEST_EVIDENCE")
+    if evidence:
+        destination = Path(evidence)
+        destination.mkdir(parents=True, exist_ok=True)
+        (destination/(run+".json")).write_text(json.dumps(summary, indent=2)+"\n")
+    return value, summary, trial
+
+
+def test_same_session_crosses_100_until_one_deadline(live, monkeypatch):
+    value, summary, trial = execute(live, monkeypatch, timeout=180)
+    if os.environ.get("SFORGE_RESUME_BASE_CONTROL") != "1":
+        assert summary["supports_policy"]
+        assert summary["actual_processes"] > 101
+        policy = value["resume_policy"]
+        assert policy["stop_reason"] in ("budget_exhausted", "segment_timeout")
+        assert policy["remaining_seconds"] < 1.1
+        assert policy["deadline_unix"] - policy["started_at_unix"] == 180
+        assert policy["total_elapsed_seconds"] >= policy["phase_elapsed_seconds"] >= value["runtime_seconds"]
+        assert value["resume_count"] == summary["actual_processes"] - 1
+        assert json.loads((trial/"resume-policy.json").read_text()) == policy
+    else:
+        # Counterfactual on the exact base: the CLI ignores the new argument.
+        assert not summary["supports_policy"]
+        assert summary["actual_processes"] == 101 and value["resume_count"] == 100
+        assert not value["timed_out"] and value["runtime_seconds"] < 175
+        assert "resume_policy" not in value
+
+
+@pytest.mark.parametrize("failure", [False, True])
+def test_stop_or_quota_failure_does_not_restart(live, monkeypatch, failure):
+    if os.environ.get("SFORGE_RESUME_BASE_CONTROL") == "1":
+        pytest.skip("Opt-in stop/failure qualification applies to the new head")
+    stop = threading.Event()
+    if failure:
+        live.config.agent_extra_env["SYNTHETIC_QUOTA_FAILURE"] = "1"
+        timer = None
+    else:
+        live.config.agent_extra_env["SYNTHETIC_SLEEP"] = "30"
+        timer = threading.Timer(3, stop.set)
+        live.backend.on_solver_start = timer.start
+    try:
+        value, summary, _ = execute(live, monkeypatch, timeout=30, shutdown_event=stop)
+    finally:
+        if timer:
+            timer.cancel()
+    assert summary["actual_processes"] == 1 and value["resume_count"] == 0
+    assert value["resume_policy"]["stop_reason"] == ("process_failed" if failure else "shutdown_requested")
+    assert value["resume_policy"]["remaining_seconds"] > 0
+
+
+def test_slow_resume_format_cannot_dispatch_with_stale_time(live, monkeypatch):
+    class SlowResume(ProcessAgent):
+        def format_run_cmd(self, *args, resume=False, **kwargs):
+            if resume:
+                time.sleep(3)
+            return super().format_run_cmd(*args, resume=resume, **kwargs)
+
+    value, summary, _ = execute(live, monkeypatch, timeout=3, agent=SlowResume(live.config))
+    assert summary["actual_processes"] == 1
+    assert value["resume_policy"]["stop_reason"] == "budget_exhausted"
+    assert value["resume_policy"]["phase_elapsed_seconds"] > 3
+
+
+def test_initial_format_is_setup_before_solver_deadline(live, monkeypatch):
+    class SlowInitial(ProcessAgent):
+        def format_run_cmd(self, *args, resume=False, **kwargs):
+            if not resume:
+                time.sleep(2)
+            return super().format_run_cmd(*args, resume=resume, **kwargs)
+
+    value, summary, _ = execute(live, monkeypatch, timeout=3, agent=SlowInitial(live.config))
+    assert summary["actual_processes"] >= 2
+    policy = value["resume_policy"]
+    assert policy["total_elapsed_seconds"] - policy["phase_elapsed_seconds"] >= 2
+
+
+def test_stop_during_format_dispatches_no_process(live, monkeypatch):
+    stop = threading.Event()
+
+    class StopInFormat(ProcessAgent):
+        def format_run_cmd(self, *args, **kwargs):
+            time.sleep(.2)
+            stop.set()
+            return super().format_run_cmd(*args, **kwargs)
+
+    value, summary, _ = execute(live, monkeypatch, timeout=3, shutdown_event=stop,
+        agent=StopInFormat(live.config), expected_submissions=0)
+    assert summary["actual_processes"] == 0
+    assert value["resume_policy"]["stop_reason"] == "shutdown_requested"
+    assert value["resume_policy"]["phase_elapsed_seconds"] == 0
+
+
+@pytest.mark.parametrize("error", [OSError, KeyboardInterrupt])
+def test_failed_resume_excludes_real_history_collection(live, monkeypatch, error):
+    times = {}
+
+    class FailedResume(ProcessAgent):
+        def format_run_cmd(self, *args, resume=False, **kwargs):
+            if resume:
+                time.sleep(.2)
+                times["failure"] = time.monotonic()
+                raise error("Synthetic resume preparation failure")
+            return super().format_run_cmd(*args, resume=resume, **kwargs)
+
+    original_exec = live.backend.exec_run_with_timeout
+    original_get = runner.requests.get
+
+    def dispatch(*args, **kwargs):
+        if getattr(kwargs.get("log_file"), "name", None) == "agent_output.txt":
+            times.setdefault("dispatch", time.monotonic())
+        return original_exec(*args, **kwargs)
+
+    def history(url, *args, **kwargs):
+        if "/api/v1/history" not in str(url):
+            return original_get(url, *args, **kwargs)
+        times["history_start"] = time.monotonic()
+        time.sleep(2)
+        response = original_get(url, *args, **kwargs)
+        times["history_end"] = time.monotonic()
+        return response
+
+    monkeypatch.setattr(live.backend, "exec_run_with_timeout", dispatch)
+    monkeypatch.setattr(runner.requests, "get", history)
+    result = runner.run_agent(live.task, FailedResume(live.config), live.config, live.backend,
+        "failed-" + live.name, timeout=30, resume_until_timeout=True,
+        judge_url=live.worker_url, host_judge_url=live.host_url,
+        disable_auto_eval=True, disable_stop_hook=True,
+        max_submissions=3, submission_cooldown=3600)
+    policy = result.resume_policy
+    assert len(live.backend.processes) == len(live.backend.created) == 1
+    assert len(live.states[0].tokens) == 1
+    token, info = next(iter(live.states[0].tokens.items()))
+    assert info["next_agent"] == 2 and info["submission_cooldown"] == 3600
+    history = original_get(live.host_url + "/api/v1/history", params={"token": token}, timeout=10).json()
+    assert history["agent_submissions"] == 1 and history["entries"][0]["valid"] is True
+    assert policy["stop_reason"] == ("interrupted" if error is KeyboardInterrupt else "runner_error")
+    assert times["history_start"] > times["failure"]
+    # The production dispatch receipt precedes the actual process. Retain room
+    # for its IO without admitting this two-second post-failure HTTP delay.
+    assert policy["phase_elapsed_seconds"] <= times["failure"] - times["dispatch"] + .5
+    collection = times["history_end"] - times["history_start"]
+    assert policy["total_elapsed_seconds"] - policy["phase_elapsed_seconds"] >= collection - .5
+    assert policy["phase_elapsed_seconds"] + policy["remaining_seconds"] == pytest.approx(30)