@@ -4,10 +4,11 @@ SQLMesh can add structured metadata to warehouse statements so a query-history
 tool can connect a statement to the plan or run, model, and evaluation that issued
 it. The metadata is a versioned JSON object in a `QUERY_CONTEXT` SQL comment.
 
-Enable `query_execution_context` on a connection and set a provider on its
-engine adapter. The provider returns a `QueryExecutionContext`. The flag defaults
-to `false`. For YAML configuration, set the option under the gateway's
-`connection`:
+Set `query_execution_context` to `true` on a connection to enable bounded query
+context comments on supported adapters. The flag defaults to `false`. With this
+flag enabled, `Context.apply()` and built-in scheduler runs add model context to
+SQL on supported adapters without a custom provider. For YAML configuration,
+set the option under the gateway's `connection`:
 
 ```yaml
 gateways:
@@ -17,20 +18,59 @@ gateways:
       query_execution_context: true
 ```
 
-The flag enables context comments but does not infer model context. Without a
-provider, SQLMesh adds parent metadata only when a plan or run correlation ID is
-available. SQLMesh calls the provider after rendering each statement and before
-executing it. The call runs in the current worker's context, so concurrent model
-evaluations can return different IDs for a shared adapter.
+## Automatic model context
 
-To build a provider from native observer events, see the
-[query execution context example](../examples/query_execution_context.md).
+For each model batch, SQLMesh creates a context with a new operation ID, the
+model's canonical name, an opaque snapshot ID, and the batch interval and index
+when available. Intervals use UTC and an exclusive end.
 
-## Configure a provider
+The context stays active for SQL inside the batch, including nested audits.
+Automatic context uses the broad phases `materialize`, `audit`, and `ddl`. The
+`role` field does not identify the engine operation for each SQL statement.
 
-Set the provider on warehouse adapters before loading a Python `Context`. When
-`Context` creates its snapshot evaluator, it clones those adapters and preserves
-the provider and flag.
+Outside a model batch, SQLMesh can add parent metadata when a plan or run
+correlation ID is available. Use a custom provider when you need caller-defined
+context for other statements.
+
+## Read the context from an observer
+
+SQLMesh establishes the model context before it calls an observer's `start`
+callback. It restores the context after the `finish` callback. Both callbacks can
+read the active immutable value with `get_query_execution_context()`:
+
+```python
+from sqlmesh.core.query_execution import get_query_execution_context
+
+
+def observe(event, facts, error):
+    context = get_query_execution_context()
+    if context is not None and context.evaluation is not None:
+        print(context.evaluation.evaluation_id)
+```
+
+For console observer registration and callback behavior, see
+[Execution observations](../reference/execution_observations.md).
+
+SQLMesh forwards each observer event to the configured console observer once. If
+the console has no observer, model context still works. Nested observer scopes
+restore the outer observer and context on exit. Use `observer_scope(None)` to
+disable the built-in observer for its block.
+
+An observer callback can read the active context with
+`get_query_execution_context()`. SQL issued inside the callback does not use that
+automatic context. A custom adapter provider remains authoritative. Without one,
+an active plan or run correlation ID can still produce parent-only metadata.
+
+## Configure a custom provider
+
+Use a custom provider when you need caller-defined operation IDs or context
+outside a model batch. Set it on warehouse adapters before loading a Python
+`Context`. When `Context` creates its snapshot evaluator, it clones those
+adapters and preserves the provider and flag.
+
+SQLMesh calls the provider after rendering each statement and before executing
+it. The provider runs in the current worker's context, so concurrent workers can
+return different values for a shared adapter.
 
 ```python
 from sqlmesh.core.context import Context
@@ -53,23 +93,32 @@ with observer_scope(observer):
 
 `QueryContextObserver` returns the immutable context active in its worker, or
 `None` when no context applies. It assigns an evaluation ID when SQLMesh reports
-a model batch and restores the previous context when that event finishes. It also
+a model batch and restores the previous context when that event finishes. It
 forwards the same event, facts, and error to the observer returned by the console.
 
+A custom provider replaces the built-in provider for its adapter. If it returns
+`None`, SQLMesh does not use automatic model context for that statement. SQLMesh
+can still add parent-only metadata when the adapter has a correlation ID.
+
 Adapters created with `with_settings` inherit the flag and provider. Pass
 `query_execution_context_provider=None` to clear the inherited provider. Setting
 the flag to `false` disables provider calls and comment output.
 
 When `state_connection` is omitted, the built-in scheduler uses the warehouse
-connection for its state adapter and disables query-context comments on that
-adapter. To add context to state SQL, configure `state_connection` explicitly
-and set a provider on its adapter.
+connection for its state adapter and disables query context on that adapter. An
+explicitly configured `state_connection` uses its own `query_execution_context`
+setting. The automatic observer activates only if a supported warehouse adapter
+has `query_execution_context` enabled and no custom provider. An enabled state
+adapter without a custom provider can use that context during a model batch.
+Setting the flag only on `state_connection` does not activate the observer. Use a
+custom provider for state SQL outside a model batch.
 
 ## Read the payload
 
 The version-one payload uses the SQLMesh model's canonical name and opaque
-snapshot identifier. Operation intervals use UTC half-open boundaries. Fields
-that are unavailable are omitted.
+snapshot identifier. Operation intervals use UTC half-open boundaries, and the
+payload includes `batch_index` when available. Fields that are unavailable are
+omitted.
 
 ```json
 {
@@ -93,9 +142,9 @@ that are unavailable are omitted.
 }
 ```
 
-The provider assigns an operation ID to the statements for one model evaluation,
-including nested audits. A later evaluation gets a new ID. The adapter preserves
-the provider's ID. Consumers can compare it with the execution observer's record.
+Each model batch gets one operation ID, which nested audits reuse. The built-in
+provider creates this ID. A custom provider supplies its own. The adapter preserves
+the provider's ID when it adds the context comment.
 
 The adapter's correlation ID sets `parent` and replaces any parent supplied by
 the provider. If the adapter has no correlation ID, SQLMesh clears the provider's
 
@@ -212,7 +212,7 @@ Most parameters are specific to the connection engine `type` - see [below](#engi
 | `register_comments` | Whether SQLMesh should register model comments with the SQL engine (if the engine supports it). (Default: `true`.)                                                      | bool |    N     |
 | `pre_ping`          | Whether or not to pre-ping the connection before starting a new transaction to ensure it is still alive. This can only be enabled for engines with transaction support. | bool |    N     |
 | `pretty_sql`        | If SQL should be formatted before being executed, not recommended in a production setting. (Default: `false`.)                                                          | bool |    N     |
-| `query_execution_context` | Whether an opted-in engine adapter adds bounded execution metadata to SQL comments. (Default: `false`.) See [query execution context](../concepts/query_execution_context.md). | bool | N |
+| `query_execution_context` | Enables bounded query-context comments on supported adapters (Default: `false`). During `Context.apply()` and built-in scheduler runs, SQLMesh adds context for each model batch to enabled adapters without a custom provider. See [query execution context](../concepts/query_execution_context.md). | bool | N |
 
 #### Engine-specific
 
 
@@ -2706,7 +2706,14 @@ def _attach_query_execution_context(self, sql: str) -> str:
 
             provider = self.query_execution_context_provider
             try:
-                context = provider() if provider is not None else None
+                if provider is None:
+                    from sqlmesh.core.query_execution_context_observer import (
+                        get_automatic_query_execution_context,
+                    )
+
+                    context = get_automatic_query_execution_context()
+                else:
+                    context = provider()
             except Exception:
                 self._warn_query_execution_context_once(
                     "provider",
 
@@ -119,14 +119,33 @@ def execution_context_factory() -> t.Optional[t.Callable[[], contextvars.Context
 
 
 @contextlib.contextmanager
-def console_observer_scope(console: t.Any) -> t.Iterator[None]:
+def console_observer_scope(
+    console: t.Any, *, query_execution_context: bool = False
+) -> t.Iterator[None]:
     """Resolve once at the outer boundary; an explicit None scope disables hooks."""
     if _scoped.get():
+        observer = _observer.get()
+        if query_execution_context and observer is not None:
+            from sqlmesh.core.query_execution_context_observer import (
+                with_automatic_query_execution_observer,
+            )
+
+            automatic_observer = with_automatic_query_execution_observer(observer)
+            if automatic_observer is not observer:
+                with observer_scope(automatic_observer):
+                    yield
+                return
         yield
         return
     try:
         observer = console.get_execution_observer()
     except Exception:
         observer = None
+    if query_execution_context:
+        from sqlmesh.core.query_execution_context_observer import (
+            with_automatic_query_execution_observer,
+        )
+
+        observer = with_automatic_query_execution_observer(observer)
     with observer_scope(observer):
         yield
@@ -19,6 +19,9 @@
 import typing as t
 from sqlmesh.core import analytics
 from sqlmesh.core.execution_observation import action, console_observer_scope
+from sqlmesh.core.query_execution_context_observer import (
+    should_enable_automatic_query_execution_context,
+)
 from sqlmesh.core import constants as c
 from sqlmesh.core.console import Console, get_console
 from sqlmesh.core.environment import EnvironmentNamingInfo, execute_environment_statements
@@ -103,7 +106,12 @@ def evaluate(
 
         try:
             plan_stages = stages.build_plan_stages(plan, self.state_sync, self.default_catalog)
-            with console_observer_scope(self.console):
+            with console_observer_scope(
+                self.console,
+                query_execution_context=should_enable_automatic_query_execution_context(
+                    self.snapshot_evaluator.adapters.values()
+                ),
+            ):
                 self._evaluate_stages(plan_stages, plan)
         except Exception as e:
             analytics.collector.on_plan_apply_end(plan_id=plan.plan_id, error=e)
 
@@ -170,10 +170,10 @@ def _query_execution_context_transport_scope() -> t.Iterator[None]:
 
 
 @contextlib.contextmanager
-def query_execution_scope(context: QueryExecutionContext) -> t.Iterator[None]:
-    """Set a query execution context for this scope and restore it on exit."""
-    if not isinstance(context, QueryExecutionContext):
-        raise TypeError("context must be a QueryExecutionContext")
+def query_execution_scope(context: t.Optional[QueryExecutionContext]) -> t.Iterator[None]:
+    """Set or clear the query execution context for this scope."""
+    if context is not None and not isinstance(context, QueryExecutionContext):
+        raise TypeError("context must be a QueryExecutionContext or None")
     token = _current_query_execution_context.set(context)
     try:
         yield