@@ -325,7 +325,7 @@ Emitted immediately before the SDK call.
 | `message_count` | int | Number of messages in the conversation |
 | `tool_count` | int | Number of tools available |
 | `streaming` | bool | Whether streaming is enabled (default: `true`) |
-| `timeout` | float | Request timeout in seconds |
+| `timeout` | float or null | Explicit request timeout in seconds; null waits for completion, cancellation or failure |
 
 ### `llm:response`
 
 
@@ -32,7 +32,8 @@
     "defaults": {
         "model": "auto",
         "max_tokens": 4096,
-        "timeout": 3600,
+        # Healthy generation has no implicit elapsed-time deadline.
+        "timeout": None,
         "context_window": 128000,
         "max_output_tokens": 16384,
     },
 
@@ -187,6 +187,10 @@ class SessionProtectionConfig:
     # Contract: streaming-contract:abort-on-capture:MUST:1
     idle_timeout_seconds: float = 30.0
 
+    # Public SDK ping cadence while waiting for completion. This is not a timeout:
+    # a slow ping or long generation is never failed because of elapsed time.
+    connection_check_interval_seconds: float = 1.0
+
     # Timeout for session.disconnect() call.
     # Prevents indefinite hang on stubborn sessions during cleanup.
     # Contract: sdk-protection:Session:MUST:3
 
@@ -156,7 +156,7 @@ async def emit_request(
         message_count: int,
         tool_count: int,
         streaming: bool,
-        timeout: float,
+        timeout: float | None,
         raw_request: dict[str, Any] | None = None,
     ) -> None:
         """Emit llm:request event.
 
@@ -25,7 +25,9 @@
 import logging
 import time
 import uuid
+from contextlib import suppress
 from dataclasses import dataclass, field
+from inspect import iscoroutinefunction
 from typing import Any, cast
 
 from amplifier_core import (
@@ -586,6 +588,8 @@ def get_info(self) -> ProviderInfo:
         # are already corrected for the effective model just below.
         # Contract: provider-protocol:get_info:MUST:2
         defaults["model"] = self._effective_default_model
+        configured_timeout = self.config.get("timeout", defaults["timeout"])
+        defaults["timeout"] = None if configured_timeout is None else float(configured_timeout)
         # Report the default model's tier-selected PROMPT budget so Amplifier's
         # compaction math matches the active context tier. Cold cache (model not
         # yet discovered) keeps the static fallback from config/_models.py.
@@ -800,10 +804,11 @@ async def complete(
         async with llm_lifecycle(self.coordinator, model, raw=self._raw) as ctx:
             # Real SDK path: use client wrapper with STREAMING
             # Three-Medium: timeout from YAML config
-            timeout_seconds: float = kwargs.get(
+            configured_timeout = kwargs.get(
                 "_timeout_seconds",
-                float(self._provider_config.defaults["timeout"]),
+                self.config.get("timeout", self._provider_config.defaults["timeout"]),
             )
+            timeout_seconds = None if configured_timeout is None else float(configured_timeout)
 
             retry_config = self._retry_config
 
@@ -973,11 +978,6 @@ async def complete(
                 )
                 accumulator = StreamingAccumulator()
 
-                # Three-Medium: timeout from YAML config
-                timeout_seconds_retry: float = kwargs.get(
-                    "_timeout_seconds",
-                    float(self._provider_config.defaults["timeout"]),
-                )
                 try:
                     # Note: attachments=None for correction - the model already saw the image
                     # Correction responses only need to produce a structured tool call —
@@ -992,7 +992,7 @@ async def complete(
                         client=self._client,
                         model=model,
                         prompt=corrected_prompt,
-                        timeout=timeout_seconds_retry,
+                        timeout=timeout_seconds,
                         event_config=event_config,
                         accumulator=accumulator,
                         tools=internal_request.tools or None,
@@ -1151,7 +1151,7 @@ async def _execute_sdk_completion(
         client: CopilotClientWrapper,
         model: str,
         prompt: str,
-        timeout: float,
+        timeout: float | None,
         event_config: EventConfig,
         accumulator: StreamingAccumulator,
         tools: list[Any] | None = None,
@@ -1178,7 +1178,7 @@ async def _execute_sdk_completion(
         """
         # DEBUG: Log entry point with key parameters
         logger.debug(
-            "[SDK_COMPLETION] Starting: model=%s, prompt_len=%d, timeout=%.1f, "
+            "[SDK_COMPLETION] Starting: model=%s, prompt_len=%d, timeout=%s, "
             "tools=%d, attachments=%d, idle_events=%s, system_message_len=%d",
             model,
             len(prompt),
@@ -1261,6 +1261,21 @@ async def _execute_sdk_completion(
                 )
 
                 unsubscribe = sdk_session.on(event_handler)
+                # The SDK's public send() acknowledges submission, so pipe failure
+                # afterwards may not emit session.error. A separate public ping
+                # watcher detects real connection failure without a generation deadline.
+                connection_task: asyncio.Task[None] | None = None
+                watch_connection = getattr(sdk_session, "wait_for_disconnect", None)
+                if iscoroutinefunction(watch_connection):
+                    async def watch() -> None:
+                        try:
+                            await watch_connection()
+                        except Exception as exc:
+                            if not idle_event.is_set():
+                                error_holder.append(exc)
+                                idle_event.set()
+
+                    connection_task = asyncio.create_task(watch())
                 try:
                     # Record TTFT start time before send
                     # Contract: behaviors:Streaming:MUST:1
@@ -1391,8 +1406,24 @@ async def _execute_sdk_completion(
                         stream_ctx.signal_done()
                         if consumer_task is not None:
                             await consumer_task
+                except asyncio.CancelledError:
+                    # User cancellation and explicit caller deadlines terminate
+                    # native work too; cleanup bounds do not limit healthy generation.
+                    try:
+                        await asyncio.wait_for(
+                            sdk_session.abort(),
+                            timeout=sdk_protection.session.abort_timeout_seconds,
+                        )
+                    except Exception:
+                        logger.debug("[provider] Abort during cancellation failed; disconnecting")
+                    raise
                 finally:
                     unsubscribe()
+                    for task in (connection_task, consumer_task):
+                        if task is not None:
+                            task.cancel()
+                            with suppress(asyncio.CancelledError):
+                                await task
 
     # =========================================================================
     # Progressive Streaming Emission
 
@@ -702,7 +702,14 @@ async def session(
             from .types import SessionHandle
 
             session_id = str(getattr(sdk_session, "session_id", "unknown"))
-            yield SessionHandle(sdk_session, session_id)
+            yield SessionHandle(
+                sdk_session,
+                session_id,
+                ping=getattr(client, "ping", None),
+                connection_check_interval=(
+                    load_sdk_protection_config().session.connection_check_interval_seconds
+                ),
+            )
         finally:
             if sdk_session is not None:
                 try:
 
@@ -6,6 +6,7 @@
 Contract: contracts/sdk-boundary.md
 """
 
+import asyncio
 import logging
 from collections.abc import Awaitable, Callable
 from dataclasses import dataclass, field
@@ -71,9 +72,16 @@ class SessionHandle:
         session_id: Opaque string identifier for the session
     """
 
-    __slots__ = ("_raw_session", "session_id")
+    __slots__ = ("_raw_session", "session_id", "_ping", "_connection_check_interval")
 
-    def __init__(self, raw_session: Any, session_id: str | None = None) -> None:
+    def __init__(
+        self,
+        raw_session: Any,
+        session_id: str | None = None,
+        *,
+        ping: Callable[[], Awaitable[Any]] | None = None,
+        connection_check_interval: float = 1.0,
+    ) -> None:
         """Create a session handle wrapping a raw SDK session.
 
         Args:
@@ -82,6 +90,23 @@ def __init__(self, raw_session: Any, session_id: str | None = None) -> None:
         """
         self._raw_session = raw_session
         self.session_id = session_id or getattr(raw_session, "session_id", "unknown")
+        self._ping = ping
+        self._connection_check_interval = connection_check_interval
+
+    async def wait_for_disconnect(self) -> None:
+        """Watch real SDK connection errors without imposing any response deadline.
+
+        Copilot's public ping uses the JSON-RPC client's unlimited default wait.
+        A slow ping is not treated as failure. Pipe closure or a failed write
+        raises; a successful ping merely schedules the next check. Cancellation
+        removes any pending SDK request through its normal finally cleanup.
+        """
+        if self._ping is None:
+            await asyncio.Event().wait()
+            return
+        while True:
+            await self._ping()
+            await asyncio.sleep(self._connection_check_interval)
 
     def on(self, handler: Callable[[Any], None]) -> Callable[[], None]:
         """Subscribe to session events.
 
@@ -155,7 +155,7 @@ class SessionWrapper:
 ```python
 # ✓ Façade hides the SDK session; raw reference is private
 class SessionHandle:
-    __slots__ = ("_raw_session", "session_id")
+    __slots__ = ("_raw_session", "session_id", "_ping", "_connection_check_interval")
 
     def __init__(self, raw_session: Any, session_id: str | None = None) -> None:
         self._raw_session = raw_session  # private — never exposed
@@ -189,7 +189,7 @@ class SessionHandle:
 ```
 
 - `Types:MUST:4` — **MUST** keep `_raw_session` private: no public attribute, no accessor returning it (`sdk_adapter/types.py:74,83`).
-- `Types:MUST:5` — **MUST** expose only `on(handler) → unsubscribe`, `await send(prompt, *, attachments=None)`, `await abort()`, plus the attribute `session_id: str` (`sdk_adapter/types.py:86-120`).
+- `Types:MUST:5` — **MUST** expose only `on(handler) → unsubscribe`, `await send(prompt, *, attachments=None)`, `await abort()`, `await wait_for_disconnect()`, plus the attribute `session_id: str`. The connection waiter uses an injected public SDK `ping` callable and raises only for real transport failures; it must never convert elapsed time into failure or expose the raw client/session.
 - `Types:MUST:6` — **MUST NOT** expose lifecycle (`connect` / `disconnect` / `destroy` / `close`) on the handle; lifecycle is owned by the `client.session()` async context manager in `sdk_adapter/client.py`.
 - `Types:MUST:7` — **MUST** assign `session_id` exactly once in `__init__` from the raw SDK session (`copilot.session.CopilotSession.session_id`); provider code MUST NOT reassign it. Callers **SHOULD** treat it as read-only — `__slots__` constrains attribute names but does not block reassignment of declared slots; a `__setattr__` guard or `frozen` dataclass is tracked as a code-level follow-up.
 
 
@@ -65,11 +65,11 @@ The provider MUST bound the abort call by `session.abort_timeout_seconds`. If ab
 
 ### SHOULD-2: Idle Timeout (Safety Bound Only)
 
-The provider SHOULD NOT use `session.idle_timeout_seconds` for the main idle wait. SDK API calls can take 60+ seconds for complex operations (e.g., agent delegation). Use the caller's request timeout instead.
+The provider MUST NOT impose a default elapsed-time deadline on healthy generation. `defaults.timeout` is `None`; an explicit per-call `_timeout_seconds` or provider `timeout` value still sets a deadline. An explicit per-call `None` disables an inherited deadline. The provider MUST NOT use `session.idle_timeout_seconds` for the main idle wait.
 
 The `idle_timeout_seconds` config is retained for abort operations only.
 
-**Implementation:** `asyncio.wait_for(idle_event.wait(), timeout=timeout)` — uses caller's timeout
+**Implementation:** one outer `asyncio.timeout(timeout)` with `None` by default; await `idle_event.wait()` directly. The SDK `send()` RPC has a genuine unlimited default. A public SDK `ping()` watcher detects actual connection errors after submission without assigning a ping deadline. Slow pings are not failures. On explicit cancellation or deadline expiration, abort native work, unsubscribe handlers, cancel/join the connection watcher and stream consumer, then disconnect. Existing bounded abort/disconnect/close cleanup remains in force.
 
 ---
 
 
@@ -178,8 +178,8 @@ async def send(
 class TestTimeoutConfigValue:
     """Tests for timeout configuration values."""
 
-    def test_yaml_timeout_is_3600(self) -> None:
-        """YAML timeout value is 3600 (1 hour for reasoning models).
+    def test_default_completion_timeout_is_unlimited(self) -> None:
+        """Healthy generation has no default elapsed-time cutoff.
 
         Contract: behaviors:Config:MUST:2
         Three-Medium: YAML is authoritative source.
@@ -190,4 +190,4 @@ def test_yaml_timeout_is_3600(self) -> None:
 
         config = load_models_config()
         timeout = config.defaults["timeout"]
-        assert timeout == 3600
+        assert timeout is None