diff --git a/CHANGELOG.md b/CHANGELOG.md
index f2a2220..e5fbe33 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -4,6 +4,22 @@ All notable changes to `xgen-agent-runtime` are recorded here. The format
 follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/) and
 this project adheres to [Semantic Versioning](https://semver.org/).
 
+## [4.102.0] - 2026-10-24
+
+### Fixed: 일하는 턴은 시계로 끊기지 않는다 (아무 일도 없을 때만 끝난다)
+
+DEV 에서 긴 에이전트 턴(긴 빌드·연이은 도구 호출)이 일하는 중에 끊겼다. 시간 한도가 모두 **시작부터 센 고정값**이었다.
+
+- `RunBudget(max_idle_seconds=…)`: 유휴 모드. 아무 일도 없이(모델 호출·도구 호출이 없고 도는 것도 없이) 그만큼 지나야
+  끝난다. 도는 도구(`reserve`~`release`)와 도는 모델 호출(`begin_model_call`~`end_model_call`)은 활동이다.
+  `max_wall_seconds` 는 상한이 된다. 기본(유휴 모드 아님)은 예전과 같다. `expired()`, `idle_seconds()`, `touch()`.
+- `max_total_result_bytes`: 턴 전체 결과 바이트 상한을 결과 하나의 상한(`max_result_bytes`)과 따로 둔다.
+- Stage 6 이 모델 호출을 도는 동안 호스트에 알린다(`_host_model_activity`, `bind_run_guard` 가 건다).
+- 라우터의 임시 취소 범위는 마감을 한 번 읽지 않고 예산에 매번 묻는다(유휴 모드의 마감은 활동으로 밀린다).
+- CLI(claude_code·codex): `timeout_s` 가 **유휴 한도**가 됐다. 출력도 도구 활동도 없이 그만큼 지나야 멈춘다. 브릿지
+  도구가 도는 동안 CLI 는 아무것도 출력하지 않으므로 턴의 도구 표면이 활동을 알린다(`TurnToolSurface.idle_seconds`,
+  `set_activity_probe`). 예전에는 시작부터 1시간이면 일하는 중에도 끊었다.
+
 ## [4.101.0] - 2026-10-23
 
 ### Fixed: 2026-10-23 전체 흐름 검토 (창, assignment 메모리, 정보 유실)
diff --git a/pyproject.toml b/pyproject.toml
index 1197d87..456e4e1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "xgen-agent-runtime"
-version = "4.101.0"
+version = "4.102.0"
 description = "Harness-engineered agent pipeline library with 21-stage dual-abstraction architecture, built on the Anthropic API"
 readme = "README.md"
 license = "Apache-2.0"
diff --git a/src/xgen_agent_runtime/host/execution_scope.py b/src/xgen_agent_runtime/host/execution_scope.py
index 2cb1f2e..c622cd1 100644
--- a/src/xgen_agent_runtime/host/execution_scope.py
+++ b/src/xgen_agent_runtime/host/execution_scope.py
@@ -16,6 +16,18 @@ class ExecutionBoundaryError(RuntimeError):
 
 
 class RunBudget:
+    """One run's tool calls, result bytes, usage and time.
+
+    Time is either a fixed wall budget (``max_wall_seconds`` from creation, the default) or, with
+    ``max_idle_seconds`` set, an **idle** budget: the run ends only after that long with nothing
+    happening (no model call, no tool call, nothing in flight); ``max_wall_seconds`` is then the hard
+    ceiling. A host serving long agent turns uses idle mode. A fixed five minutes cut a turn that was
+    working, a long build or a run of tool calls, in the middle (2026-10-24 review).
+
+    ``max_result_bytes`` bounds one result; ``max_total_result_bytes`` bounds all of the run's results
+    together (the same value when not given, as before).
+    """
+
     def __init__(
         self,
         *,
@@ -25,11 +37,15 @@ def __init__(
         max_wall_seconds: float = 300,
         max_cost: float = 0,
         max_tokens: int = 0,
+        max_idle_seconds: float = 0,
+        max_total_result_bytes: Optional[int] = None,
     ) -> None:
         if (
             min(max_calls, max_active, max_result_bytes) <= 0
             or not math.isfinite(max_wall_seconds)
             or max_wall_seconds <= 0
+            or not math.isfinite(max_idle_seconds)
+            or max_idle_seconds < 0
         ):
             raise ValueError("invalid run budget")
         self.max_calls, self.max_active, self.max_result_bytes = (
@@ -37,7 +53,16 @@ def __init__(
             max_active,
             max_result_bytes,
         )
-        self.deadline = time.monotonic() + max_wall_seconds
+        self.max_total_result_bytes = (
+            max_result_bytes if max_total_result_bytes is None else int(max_total_result_bytes)
+        )
+        if self.max_total_result_bytes <= 0:
+            raise ValueError("invalid run budget")
+        now = time.monotonic()
+        self._hard_deadline = now + max_wall_seconds
+        self.max_idle_seconds = float(max_idle_seconds)
+        self._last_activity = now
+        self.models_active = 0
         if (
             not math.isfinite(max_cost)
             or max_cost < 0
@@ -52,25 +77,62 @@ def __init__(
         self._usage_seen: dict[str, tuple[int, float]] = {}
         self._lock = threading.Lock()
 
+    @property
+    def idle_mode(self) -> bool:
+        return self.max_idle_seconds > 0
+
+    @property
+    def deadline(self) -> float:
+        """When the run ends if nothing more happens. Work in flight keeps an idle-mode run open until
+        the hard ceiling."""
+        if not self.idle_mode:
+            return self._hard_deadline
+        if self.active > 0 or self.models_active > 0:
+            return self._hard_deadline
+        return min(self._hard_deadline, self._last_activity + self.max_idle_seconds)
+
+    def expired(self) -> bool:
+        return time.monotonic() >= self.deadline
+
+    def idle_seconds(self) -> float:
+        """Seconds since anything happened in this run (0 while a tool or model call is in flight)."""
+        if self.active > 0 or self.models_active > 0:
+            return 0.0
+        return max(0.0, time.monotonic() - self._last_activity)
+
+    def touch(self) -> None:
+        """Record activity (an idle-mode run stays open)."""
+        self._last_activity = time.monotonic()
+
     def before_model_call(self) -> None:
         """Host port invoked before every provider dispatch, including retries."""
-        if time.monotonic() >= self.deadline:
+        if self.expired():
             raise ExecutionBoundaryError("run_budget_exceeded")
+        self.touch()
+
+    def begin_model_call(self) -> None:
+        """A provider call is in flight (a long answer or a CLI turn is activity, not idleness)."""
+        with self._lock:
+            self.models_active += 1
+            self._last_activity = time.monotonic()
+
+    def end_model_call(self) -> None:
+        with self._lock:
+            self.models_active = max(0, self.models_active - 1)
+            self._last_activity = time.monotonic()
 
     def reserve(self) -> None:
         with self._lock:
-            if (
-                time.monotonic() >= self.deadline
-                or self.calls >= self.max_calls
-                or self.active >= self.max_active
-            ):
+            if self.expired() or self.calls >= self.max_calls or self.active >= self.max_active:
                 raise ExecutionBoundaryError("run_budget_exceeded")
             self.calls += 1
             self.active += 1
+            self._last_activity = time.monotonic()
 
     def release(self) -> None:
         with self._lock:
             self.active = max(0, self.active - 1)
+            self._last_activity = time.monotonic()
 
     def charge_result(self, value: Any) -> None:
         # A bounded traversal prevents cyclic/deep results becoming a bypass.
@@ -83,7 +145,8 @@ def charge_result(self, value: Any) -> None:
             raise ExecutionBoundaryError("tool_result_invalid") from None
         with self._lock:
             self.result_bytes += size
-            if self.result_bytes > self.max_result_bytes or time.monotonic() >= self.deadline:
+            self._last_activity = time.monotonic()
+            if self.result_bytes > self.max_total_result_bytes or self.expired():
                 raise ExecutionBoundaryError("run_budget_exceeded")
 
     def charge_usage(self, *, cost: float = 0, tokens: int = 0) -> None:
@@ -98,6 +161,7 @@ def charge_usage(self, *, cost: float = 0, tokens: int = 0) -> None:
         with self._lock:
             self.cost += cost
             self.tokens += tokens
+            self._last_activity = time.monotonic()
             if (self.max_cost > 0 and self.cost > self.max_cost) or (
                 self.max_tokens > 0 and self.tokens > self.max_tokens
             ):
@@ -114,6 +178,7 @@ def reconcile_usage(self, key: str, *, cost: float, tokens: int) -> None:
             self.tokens += tokens - old_tokens
             self.cost += cost - old_cost
             self._usage_seen[key] = (tokens, cost)
+            self._last_activity = time.monotonic()
             if (self.max_cost > 0 and self.cost >= self.max_cost) or (
                 self.max_tokens > 0 and self.tokens >= self.max_tokens
             ):
@@ -131,7 +196,8 @@ def enforce(state: Any) -> None:
         if context.cancellation_scope is not None:
             context.cancellation_scope.check()
         if context.run_budget is not None:
-            if time.monotonic() >= context.run_budget.deadline:
+            expired = getattr(context.run_budget, "expired", None)
+            if expired() if callable(expired) else time.monotonic() >= context.run_budget.deadline:
                 raise ExecutionBoundaryError("run_budget_exceeded")
             usage = getattr(state, "turn_token_usage", []) or []
             tokens = sum(
@@ -176,6 +242,11 @@ def before_model_call() -> None:
                 context.run_budget.before_model_call()
 
         api_stage._host_model_boundary = before_model_call
+        budget = context.run_budget
+        if budget is not None and callable(getattr(budget, "begin_model_call", None)):
+            # A provider call in flight is activity: an idle-mode run must not end under a long answer
+            # or a CLI turn that is working (its tool calls come back through the bridge meanwhile).
+            api_stage._host_model_activity = (budget.begin_model_call, budget.end_model_call)
 
 
 class CancellationScope:
diff --git a/src/xgen_agent_runtime/host/tool_surface.py b/src/xgen_agent_runtime/host/tool_surface.py
index 3617d62..d409139 100644
--- a/src/xgen_agent_runtime/host/tool_surface.py
+++ b/src/xgen_agent_runtime/host/tool_surface.py
@@ -25,6 +25,8 @@
 import asyncio
 import json
 import logging
+import threading
+import time
 import uuid
 from typing import Any, Dict, List, Optional, Tuple
 
@@ -57,9 +59,26 @@ def __init__(
         #: CLI 는 자기 도구 사건을 스트림으로 알린다 — 여기서 파이프라인 사건으로 또 내면 화면에 두 번
         #: 뜬다. 진단용으로만 붙잡아 둔다(최근 것 몇 개).
         self.events: List[Tuple[str, Dict[str, Any]]] = []
+        #: 도구 활동. CLI 는 브릿지 도구가 도는 동안 아무것도 출력하지 않으므로, CLI 의 "조용함" 만으로는
+        #: 멈춘 것과 일하는 것을 가를 수 없다. 이 표면이 그 사실을 안다(:meth:`idle_seconds`).
+        self._calls_in_flight = 0
+        self._last_activity = time.monotonic()
+        self._activity_lock = threading.Lock()
 
     # ── 수명 ─────────────────────────────────────────────────────────
 
+    def idle_seconds(self) -> float:
+        """마지막 도구 활동 뒤 흐른 시간(초). 도구가 도는 중이면 0."""
+        with self._activity_lock:
+            if self._calls_in_flight > 0:
+                return 0.0
+            return max(0.0, time.monotonic() - self._last_activity)
+
+    def _activity(self, delta: int) -> None:
+        with self._activity_lock:
+            self._calls_in_flight = max(0, self._calls_in_flight + delta)
+            self._last_activity = time.monotonic()
+
     def bind_loop(self, loop: Optional[asyncio.AbstractEventLoop]) -> None:
         """턴 루프를 알린다(러너가 루프를 만들 때 부르고, 닫기 전에 None 으로 부른다)."""
         self._loop = loop
@@ -164,10 +183,14 @@ async def call(self, name: str, arguments: Dict[str, Any]) -> Dict[str, Any]:
             running = asyncio.get_running_loop()
         except RuntimeError:
             running = None
-        if running is loop:
-            return await self._dispatch(name, arguments)
-        future = asyncio.run_coroutine_threadsafe(self._dispatch(name, arguments), loop)
-        return await asyncio.wrap_future(future)
+        self._activity(+1)
+        try:
+            if running is loop:
+                return await self._dispatch(name, arguments)
+            future = asyncio.run_coroutine_threadsafe(self._dispatch(name, arguments), loop)
+            return await asyncio.wrap_future(future)
+        finally:
+            self._activity(-1)
 
 
 def to_mcp_result(result: Dict[str, Any]) -> Dict[str, Any]:
diff --git a/src/xgen_agent_runtime/host/turn_executor.py b/src/xgen_agent_runtime/host/turn_executor.py
index fbec62d..b38d98a 100644
--- a/src/xgen_agent_runtime/host/turn_executor.py
+++ b/src/xgen_agent_runtime/host/turn_executor.py
@@ -927,6 +927,13 @@ def run_dir_cleanup(_d: str = run_dir) -> None:
                     "codex",
                     kwargs,
                 )
+            if _tool_surface is not None and callable(
+                getattr(llm_client, "set_activity_probe", None)
+            ):
+                # The CLI prints nothing while a bridged tool runs: its time limit is idle time, and the
+                # surface says when the tools were last busy. A working turn is no longer stopped at
+                # the limit; a silent one with nothing running still is (2026-10-24 review).
+                llm_client.set_activity_probe(_tool_surface.idle_seconds)
 
             # ── 작은 워크스페이스 fast path (관리자/노드 opt-in) ──────────
             # 전체 입력을 확실히 실을 수 있을 때만 탐색 왕복을 없앤다. 과제명·확장자·
diff --git a/src/xgen_agent_runtime/llm_client/_cli_runtime.py b/src/xgen_agent_runtime/llm_client/_cli_runtime.py
index 0a21f87..2b3888b 100644
--- a/src/xgen_agent_runtime/llm_client/_cli_runtime.py
+++ b/src/xgen_agent_runtime/llm_client/_cli_runtime.py
@@ -51,6 +51,7 @@
     Any,
     AsyncGenerator,
     AsyncIterator,
+    Callable,
     Iterable,
     Mapping,
     NamedTuple,
@@ -343,6 +344,13 @@ class CLIProcessRunner:
     #: the pipe buffer, which drain immediately. This grace is generous
     #: for that and short enough that a leaked FD costs seconds.
     exit_drain_grace_s: float = 5.0
+    #: ``timeout_s`` is an **idle** limit for :meth:`stream`: the CLI is stopped after that long with no
+    #: output and no tool activity. A tool the CLI runs through a host bridge prints nothing while it
+    #: works, so the host says how long its tools have been idle (0 while one is running). Before
+    #: (2026-10-24) the limit counted from the start: a turn that was working was killed at the hour.
+    idle_probe: Optional[Callable[[], float]] = None
+    #: Optional hard ceiling for one stream, from its start (0 = none).
+    max_wall_s: float = 0.0
 
     def __post_init__(self) -> None:
         if not self.binary:
@@ -429,6 +437,8 @@ async def stream(
                 start_t=t0,
                 proc=proc,
                 drain_grace_s=self.exit_drain_grace_s,
+                idle_probe=self.idle_probe,
+                max_wall_s=self.max_wall_s,
             ):
                 yield line
             rc = await self._reap(proc)
@@ -575,9 +585,15 @@ async def _aiter_lines(
     start_t: float,
     proc: Optional[asyncio.subprocess.Process] = None,
     drain_grace_s: float = 5.0,
+    idle_probe: Optional[Callable[[], float]] = None,
+    max_wall_s: float = 0.0,
 ) -> AsyncIterator[bytes]:
     """Yield stdout lines until EOF — or until the child is gone and quiet.
 
+    ``timeout_s`` is an idle limit: it is measured from the last line, and while ``idle_probe`` says a
+    tool is running or ran recently (its idle seconds are below the limit) the stream keeps waiting.
+    ``max_wall_s`` (when set) is a hard ceiling from ``start_t``.
+
     Once the child has exited, EOF is no longer guaranteed to arrive at
     all: an inherited write end can keep the pipe open indefinitely. So
     the read budget collapses to ``drain_grace_s`` from the moment the
@@ -589,12 +605,25 @@ async def _aiter_lines(
         return
     read_task: Optional[asyncio.Task[bytes]] = None
     died_at: Optional[float] = None
+    last_line_t = start_t
     try:
         while True:
             now = time.monotonic()
-            remaining = timeout_s - (now - start_t)
+            if max_wall_s > 0 and now - start_t >= max_wall_s:
+                raise CLITimeout(f"stream exceeded its {max_wall_s:.0f}s ceiling")
+            quiet = now - last_line_t
+            if idle_probe is not None and quiet >= timeout_s:
+                try:
+                    quiet = min(quiet, float(idle_probe()))
+                except Exception:  # noqa: BLE001 (a broken probe must not keep a dead CLI alive)
+                    pass
+            remaining = timeout_s - quiet
             if remaining <= 0:
-                raise CLITimeout(f"stream timeout after {timeout_s:.1f}s")
+                raise CLITimeout(
+                    f"no output and no tool activity for {timeout_s:.1f}s"
+                    if idle_probe is not None
+                    else f"stream timeout: no output for {timeout_s:.1f}s"
+                )
 
             # One read task, carried across iterations: a readline that
             # loses the race must NOT be discarded, or the bytes it is
@@ -629,11 +658,8 @@ async def _aiter_lines(
             )
 
             if read_task not in done:
-                if died_at is not None or budget < remaining:
-                    # Either draining after exit, or just a poll tick with
-                    # budget left — go round again.
-                    continue
-                raise CLITimeout(f"stream readline timeout after {timeout_s:.1f}s")
+                # Draining after exit, or a poll tick: go round again (the idle check is at the top).
+                continue
 
             try:
                 line = read_task.result()
@@ -657,6 +683,7 @@ async def _aiter_lines(
                     read_task = None
             if not line:
                 return
+            last_line_t = time.monotonic()
             yield line
     finally:
         if read_task is not None and not read_task.done():
diff --git a/src/xgen_agent_runtime/llm_client/claude_code.py b/src/xgen_agent_runtime/llm_client/claude_code.py
index 7327e43..d9a9ab7 100644
--- a/src/xgen_agent_runtime/llm_client/claude_code.py
+++ b/src/xgen_agent_runtime/llm_client/claude_code.py
@@ -329,6 +329,10 @@ def __init__(
         self._disallow_tools = tuple(disallow_tools)
         self._extra_args = tuple(extra_args)
         self._timeout_s = timeout_s
+        #: How long the turn's tools have been idle (seconds, 0 while one runs): set by the host for a
+        #: turn whose tools run through a bridge. ``timeout_s`` is an idle limit (no output and no tool
+        #: activity), so a CLI that waits on a long tool is not stopped (:class:`CLIProcessRunner`).
+        self._activity_probe: Optional[Callable[[], float]] = None
         self._extra_env: Dict[str, str] = dict(env_extras) if env_extras else {}
         self._strict_wire = strict_wire
         self._runner_factory = runner_factory
@@ -535,6 +539,10 @@ def _env_extras(self) -> Dict[str, str]:
     def _truthy(value: Any) -> bool:
         return str(value or "").strip().lower() in ("1", "true", "yes", "on")
 
+    def set_activity_probe(self, probe: Optional[Callable[[], float]]) -> None:
+        """Tell the client how long the turn's tools have been idle (see ``_activity_probe``)."""
+        self._activity_probe = probe
+
     def _make_runner(
         self,
         *,
@@ -564,6 +572,7 @@ def _make_runner(
         return CLIProcessRunner(
             binary=self._binary,
             cwd=self._workspace_dir,
+            idle_probe=self._activity_probe,
             env_extras=env_extras,
             timeout_s=effective_timeout,
         )
diff --git a/src/xgen_agent_runtime/llm_client/codex.py b/src/xgen_agent_runtime/llm_client/codex.py
index e83147a..edaf03b 100644
--- a/src/xgen_agent_runtime/llm_client/codex.py
+++ b/src/xgen_agent_runtime/llm_client/codex.py
@@ -182,6 +182,10 @@ def __init__(
         self._mcp_config = mcp_config
         self._extra_args = tuple(extra_args)
         self._timeout_s = timeout_s
+        #: How long the turn's tools have been idle (seconds, 0 while one runs): set by the host for a
+        #: turn whose tools run through a bridge. ``timeout_s`` is an idle limit (no output and no tool
+        #: activity), so a CLI that waits on a long tool is not stopped (:class:`CLIProcessRunner`).
+        self._activity_probe: Optional[Callable[[], float]] = None
         self._extra_env: Dict[str, str] = dict(env_extras) if env_extras else {}
         self._strict_wire = strict_wire
         self._runner_factory = runner_factory
@@ -205,6 +209,10 @@ def _env_extras(self) -> Dict[str, str]:
             extras["OPENAI_API_KEY"] = self._api_key
         return extras
 
+    def set_activity_probe(self, probe: Optional[Callable[[], float]]) -> None:
+        """Tell the client how long the turn's tools have been idle (see ``_activity_probe``)."""
+        self._activity_probe = probe
+
     def _make_runner(self, *, timeout_s: Optional[float] = None) -> CLIProcessRunner:
         effective_timeout = self._timeout_s if timeout_s is None else timeout_s
         if self._runner_factory is not None:
@@ -222,6 +230,7 @@ def _make_runner(self, *, timeout_s: Optional[float] = None) -> CLIProcessRunner
         return CLIProcessRunner(
             binary=self._binary,
             cwd=self._workspace_dir,
+            idle_probe=self._activity_probe,
             env_extras=self._env_extras(),
             timeout_s=effective_timeout,
         )
diff --git a/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py b/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py
index f685dc6..f83c47c 100644
--- a/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py
+++ b/src/xgen_agent_runtime/stages/s06_api/artifact/default/stage.py
@@ -621,6 +621,22 @@ async def _call_provider() -> APIResponse:
 
         return response
 
+    async def _while_model_active(self, call: Any) -> Any:
+        """Run one provider call with the host told it is in flight (``_host_model_activity``).
+
+        A host that ends a run after a stretch of inactivity must not count a long answer, or a CLI
+        turn whose tools are running, as inactivity (2026-10-24: a working turn was cut at five minutes).
+        """
+        hooks = getattr(self, "_host_model_activity", None)
+        if not hooks:
+            return await call
+        begin, end = hooks
+        begin()
+        try:
+            return await call
+        finally:
+            end()
+
     async def _recover_overflow(
         self, state: PipelineState, error: APIError, *, client: Any = None, model: str = ""
     ) -> bool:
@@ -935,7 +951,9 @@ async def _call_with_retry(
             try:
                 limit = self._request_timeout_s()
                 try:
-                    return await asyncio.wait_for(client.create_message(**kwargs), timeout=limit)
+                    return await self._while_model_active(
+                        asyncio.wait_for(client.create_message(**kwargs), timeout=limit)
+                    )
                 except asyncio.TimeoutError:
                     raise APIError(
                         f"모델이 {limit:g}초 안에 답하지 않았습니다",
@@ -1000,7 +1018,9 @@ async def _call_streaming_with_retry(
             if boundary is not None:
                 boundary()
             try:
-                return await self._call_streaming(client, cfg, state, extra_messages=extra_messages)
+                return await self._while_model_active(
+                    self._call_streaming(client, cfg, state, extra_messages=extra_messages)
+                )
             except APIError as e:
                 last_error = e
                 if e.category == ErrorCategory.TIMEOUT:
diff --git a/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py b/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py
index e1d4242..d19498f 100644
--- a/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py
+++ b/src/xgen_agent_runtime/stages/s10_tool/artifact/default/routers.py
@@ -459,7 +459,14 @@ async def decide(inp):
         timeout_s = float(getattr(caps, "timeout_s", 0.0) or 0.0)
         effective_scope = scope
         if effective_scope is None and budget is not None:
-            effective_scope = CancellationScope(deadline=budget.deadline)
+            expired = getattr(budget, "expired", None)
+            # Ask the budget each tick: an idle-mode budget's deadline moves with activity, and this
+            # very call is activity (a deadline read once here would cut a long tool that is working).
+            effective_scope = (
+                CancellationScope(cancel_check=expired)
+                if callable(expired)
+                else CancellationScope(deadline=budget.deadline)
+            )
         execution_failed = False
         try:
             if effective_scope is not None:
diff --git a/tests/unit/test_long_turn_alive.py b/tests/unit/test_long_turn_alive.py
new file mode 100644
index 0000000..c75f0cc
--- /dev/null
+++ b/tests/unit/test_long_turn_alive.py
@@ -0,0 +1,120 @@
+"""A turn that is working is not cut by a clock; one with nothing happening is (2026-10-24, 4.102.0).
+
+A fixed five-minute run budget and a CLI limit counted from the start stopped agent turns in the
+middle of a long build or a run of tool calls. Idle mode counts time with nothing happening instead.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import sys
+import time
+
+import pytest
+
+from xgen_agent_runtime.host.execution_scope import ExecutionBoundaryError, RunBudget
+
+
+def test_an_idle_budget_stays_open_while_a_tool_or_a_model_call_is_in_flight():
+    budget = RunBudget(max_wall_seconds=30, max_idle_seconds=0.2)
+    budget.reserve()  # a tool starts
+    time.sleep(0.3)
+    assert not budget.expired(), "a running tool is activity"
+    budget.release()
+    budget.begin_model_call()
+    time.sleep(0.3)
+    assert not budget.expired(), "a provider call in flight is activity"
+    budget.end_model_call()
+    assert not budget.expired()
+    time.sleep(0.3)
+    assert budget.expired(), "nothing happening for the idle limit ends the run"
+    with pytest.raises(ExecutionBoundaryError):
+        budget.before_model_call()
+
+
+def test_activity_keeps_an_idle_budget_open_past_the_old_wall_and_the_ceiling_still_holds():
+    budget = RunBudget(max_wall_seconds=0.6, max_idle_seconds=0.25)
+    for _ in range(3):
+        time.sleep(0.15)
+        budget.before_model_call()
+    assert not budget.expired(), "0.45s of steady activity is more than one idle limit"
+    time.sleep(0.2)
+    budget.touch()
+    assert budget.expired(), "the hard ceiling still ends the run"
+
+
+def test_the_default_budget_is_still_a_fixed_wall_and_total_results_have_their_own_cap():
+    fixed = RunBudget(max_wall_seconds=0.1)
+    fixed.reserve()
+    time.sleep(0.15)
+    assert fixed.expired(), "without idle mode nothing changes"
+
+    budget = RunBudget(max_result_bytes=64, max_total_result_bytes=200)
+    for _ in range(4):
+        budget.charge_result("x" * 40)  # each result is within the per-result cap
+    with pytest.raises(ExecutionBoundaryError):
+        budget.charge_result("x" * 40)
+
+
+def _spawn(code: str):
+    return asyncio.create_subprocess_exec(
+        sys.executable, "-c", code, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE
+    )
+
+
+async def _lines(code: str, **kw):
+    from xgen_agent_runtime.llm_client._cli_runtime import _aiter_lines
+
+    proc = await _spawn(code)
+    out = []
+    try:
+        async for line in _aiter_lines(proc.stdout, start_t=time.monotonic(), proc=proc, **kw):
+            out.append(line.strip())
+    finally:
+        if proc.returncode is None:
+            proc.kill()
+        await proc.wait()
+    return out
+
+
+async def test_a_cli_that_keeps_printing_runs_past_its_time_limit():
+    code = "import time\nfor i in range(6):\n    print(i, flush=True); time.sleep(0.15)"
+    assert await _lines(code, timeout_s=0.5) == [b"0", b"1", b"2", b"3", b"4", b"5"]
+
+
+async def test_a_quiet_cli_waits_for_a_running_tool_and_stops_when_nothing_happens():
+    from xgen_agent_runtime.llm_client._cli_runtime import CLITimeout
+
+    code = "import time\nprint('a', flush=True); time.sleep(0.8); print('b', flush=True)"
+    assert await _lines(code, timeout_s=0.4, idle_probe=lambda: 0.0) == [b"a", b"b"]
+    with pytest.raises(CLITimeout, match="no output"):
+        await _lines(code, timeout_s=0.4)
+
+
+def test_the_tool_surface_knows_when_its_tools_are_busy():
+    from xgen_agent_runtime.host.tool_surface import TurnToolSurface
+    from xgen_agent_runtime.tools import ToolRegistry
+
+    surface = TurnToolSurface(registry=ToolRegistry(), tool_context=None, state=None)
+    surface._activity(+1)
+    time.sleep(0.05)
+    assert surface.idle_seconds() == 0.0
+    surface._activity(-1)
+    time.sleep(0.05)
+    assert surface.idle_seconds() >= 0.05
+
+
+def test_the_run_guard_tells_the_model_stage_when_a_call_is_in_flight():
+    from types import SimpleNamespace
+
+    from xgen_agent_runtime.host.execution_scope import bind_run_guard
+
+    budget = RunBudget(max_idle_seconds=60)
+    api_stage = SimpleNamespace()
+    pipeline = SimpleNamespace(get_stage=lambda order: api_stage if order == 6 else None)
+    bind_run_guard(pipeline, SimpleNamespace(run_budget=budget, cancellation_scope=None))
+    begin, end = api_stage._host_model_activity
+    begin()
+    assert budget.models_active == 1 and budget.idle_seconds() == 0.0
+    end()
+    assert budget.models_active == 0
diff --git a/uv.lock b/uv.lock
index fcb3fe2..3757d61 100644
--- a/uv.lock
+++ b/uv.lock
@@ -3471,7 +3471,7 @@ wheels = [
 
 [[package]]
 name = "xgen-agent-runtime"
-version = "4.101.0"
+version = "4.102.0"
 source = { editable = "." }
 dependencies = [
     { name = "anthropic", extra = ["bedrock"] },