@@ -22,6 +22,7 @@
 ## New additions
 * Streamlit entities support `include_project_definition: true` to opt in to uploading the project-root `snowflake.yml` with `snow streamlit deploy`. When that file is uploaded, deploy emits a warning because stage readers can retrieve it.
 * `snow upgrade` is now generally available. It upgrades a direct install of Snowflake CLI and is listed in `snow --help` without a feature flag. Other install channels refuse and print the direct-install command.
+* `snow sql` accepts `!spool <file>` and `!spool off`. Subsequent query results are appended to the file and still printed on the terminal. Until `!spool off`, TABLE is a static pipe table on both the terminal and the file: no Live refresh and no wrapping to the window. Format follows `--format` / `!set output_format`.
 
 ## Fixes and improvements
 * `snow streamlit deploy --replace` now applies `main_file` changes to existing versioned apps in place, preserving the app, grants and version history. Files are uploaded before the entrypoint changes; `--prune` runs only after the change succeeds. Accounts that reject the update receive an actionable error.
 
@@ -266,13 +266,16 @@ def __init__(
         bold_header: bool,
         flush_chunks: bool,
         plain: bool = False,
+        stream: object | None = None,
     ) -> None:
         self._columns = columns
         self._widths = widths
         self._wrap = wrap
         self._bold_header = bold_header
         self._plain = plain
-        self._out = _ChunkWriter(sys.stdout, flush_chunks=flush_chunks)
+        self._out = _ChunkWriter(
+            stream if stream is not None else sys.stdout, flush_chunks=flush_chunks
+        )
 
     def write_table(
         self,
@@ -378,6 +381,7 @@ def stream_probed_table(
     is_tty: bool,
     on_success: Callable[[], None],
     plain: bool = False,
+    stream: object | None = None,
 ) -> None:
     preferred = preferred_widths(probed.columns, probed.rows)
     if is_tty:
@@ -394,6 +398,7 @@ def stream_probed_table(
         bold_header=bold_header,
         flush_chunks=is_tty,
         plain=plain,
+        stream=stream,
     )
     writer.write_table(probed.rows, probed.pending, probed.rest, on_success)
 
 
@@ -49,13 +49,77 @@
 
 NO_ITEMS_FOUND: str = "No data"
 
+# Rich falls back to width=80 when stdout is not a terminal. With many columns
+# that squeezes each cell below readability, producing a series of "|"
+# separators with no content (see GH#2725). Use an effectively unlimited width
+# for non-terminal destinations so Rich picks each column's natural width.
+_NON_TERMINAL_RENDER_WIDTH = 1_000_000
+
 # ensure we do not break URLs that wrap lines
 get_console().soft_wrap = True
 
 # Disable markup to avoid escaping errors
 get_console()._markup = False  # noqa: SLF001
 
 
+class TeeTextIO:
+    """Write the same text to stdout and an open spool file. Never a TTY."""
+
+    def __init__(self, primary: TextIO, secondary: TextIO) -> None:
+        self._primary = primary
+        self._secondary = secondary
+
+    def write(self, s: str) -> int:
+        written = self._primary.write(s)
+        self._secondary.write(s)
+        return written
+
+    def flush(self) -> None:
+        self._primary.flush()
+        self._secondary.flush()
+
+    def isatty(self) -> bool:
+        return False
+
+    @property
+    def encoding(self) -> str:
+        return getattr(self._primary, "encoding", None) or "utf-8"
+
+    @property
+    def errors(self) -> str | None:
+        return getattr(self._primary, "errors", None)
+
+
+def _result_stream() -> TextIO | TeeTextIO:
+    spool = get_cli_context().spool_stream
+    if spool is None:
+        return sys.stdout
+    return TeeTextIO(sys.stdout, spool)
+
+
+def _print(*args, **kwargs) -> None:
+    kwargs.setdefault("file", _result_stream())
+    print(*args, **kwargs)
+
+
+def _rich_print(*args, **kwargs) -> None:
+    stream = _result_stream()
+    if stream is sys.stdout:
+        rich_print(*args, **kwargs)
+        return
+    flush = kwargs.pop("flush", False)
+    Console(
+        file=stream,
+        force_terminal=False,
+        markup=False,
+        soft_wrap=True,
+        width=_NON_TERMINAL_RENDER_WIDTH,
+        height=_NON_TERMINAL_RENDER_WIDTH,
+    ).print(*args, **kwargs)
+    if flush:
+        stream.flush()
+
+
 class CustomJSONEncoder(JSONEncoder):
     """Custom JSON encoder handling serialization of non-standard types"""
 
@@ -107,11 +171,16 @@ def _print_json_item_with_array_indentation(item: Any, indent: int):
         indented_lines = indented_output.split("\n")
         for i, line in enumerate(indented_lines):
             if i == 0:
-                print(" " * indent + line, end="")
+                _print(" " * indent + line, end="")
             else:
-                print("\n" + " " * indent + line, end="")
+                _print("\n" + " " * indent + line, end="")
     else:
-        json.dump(item, sys.stdout, cls=StreamingJSONEncoder, separators=(",", ":"))
+        json.dump(
+            item,
+            _result_stream(),
+            cls=StreamingJSONEncoder,
+            separators=(",", ":"),
+        )
 
 
 def _stream_collection_as_json(result: CollectionResult, indent: int = 4):
@@ -120,18 +189,18 @@ def _stream_collection_as_json(result: CollectionResult, indent: int = 4):
     try:
         first_item = next(items)
     except StopIteration:
-        print("[]", end="")
+        _print("[]", end="")
         return
 
-    print("[")
+    _print("[")
 
     _print_json_item_with_array_indentation(first_item, indent)
 
     for item in items:
-        print(",")
+        _print(",")
         _print_json_item_with_array_indentation(item, indent)
 
-    print("\n]", end="")
+    _print("\n]", end="")
 
 
 def _stream_collection_as_csv(result: CollectionResult):
@@ -146,7 +215,9 @@ def _stream_collection_as_csv(result: CollectionResult):
     if not isinstance(first_item, dict):
         raise TypeError("CSV output requires dictionary items")
 
-    writer = csv.DictWriter(sys.stdout, fieldnames=fieldnames, lineterminator="\n")
+    writer = csv.DictWriter(
+        _result_stream(), fieldnames=fieldnames, lineterminator="\n"
+    )
     writer.writeheader()
     _write_csv_row(writer, first_item)
 
@@ -194,16 +265,13 @@ def _get_table(output_format: OutputFormat | None = None):
     return Table(show_header=True, box=box.ASCII)
 
 
-# Rich falls back to width=80 when stdout is not a terminal. With many columns
-# that squeezes each cell below readability, producing a series of "|"
-# separators with no content (see GH#2725). Use an effectively unlimited width
-# for non-terminal destinations so Rich picks each column's natural width.
-_NON_TERMINAL_RENDER_WIDTH = 1_000_000
-
-
 def _stdout_is_terminal() -> bool:
     # Current stdout, not get_console().is_terminal: Rich caches the latter on
     # the stream it first saw, and FORCE_COLOR (even "0") makes that True.
+    # While spooling, both sinks get the non-TTY layout so the file does not
+    # pick up Rich ANSI / Live (see GH#2725).
+    if get_cli_context().spool_stream is not None:
+        return False
     isatty = getattr(sys.stdout, "isatty", None)
     try:
         return False if isatty is None else bool(isatty())
@@ -213,9 +281,11 @@ def _stdout_is_terminal() -> bool:
 
 
 def _render_console_for_table() -> Console:
+    stream = _result_stream()
     if _stdout_is_terminal():
         default_console = get_console()
         return Console(
+            file=stream,
             width=default_console.width,
             height=default_console.height or 25,
             soft_wrap=True,
@@ -227,6 +297,7 @@ def _render_console_for_table() -> Console:
     # size falls back to 80x25 and many-column tables collapse. Both dimensions
     # are required for the unlimited width to actually apply (GH#2725).
     return Console(
+        file=stream,
         width=_NON_TERMINAL_RENDER_WIDTH,
         height=_NON_TERMINAL_RENDER_WIDTH,
         soft_wrap=True,
@@ -242,7 +313,7 @@ def _print_multiple_table_results(
     try:
         first_item = next(items)
     except StopIteration:
-        rich_print(NO_ITEMS_FOUND, end="\n\n")
+        _rich_print(NO_ITEMS_FOUND, end="\n\n")
         return
     probed = probe_rows(first_item, items)
     if probed.use_rich:
@@ -257,15 +328,16 @@ def _trailer() -> None:
             sys.stdout.flush()
             return
         if console.is_terminal:
-            sys.stdout.write("\n")
-        rich_print(flush=True)
+            _result_stream().write("\n")
+        _rich_print(flush=True)
 
     stream_probed_table(
         probed,
         console=console,
         is_tty=is_tty,
         on_success=_trailer,
         plain=_is_plain_format(output_format),
+        stream=_result_stream(),
     )
 
 
@@ -281,7 +353,7 @@ def _print_rich_table(
     with Live(table, console=console, refresh_per_second=4):
         for row in rows:
             table.add_row(*row)
-    rich_print(flush=True)
+    _rich_print(flush=True)
 
 
 def __to_str(val):
@@ -304,7 +376,7 @@ def print_structured(
         if output_format == OutputFormat.CSV:
             for command_result in result.result:
                 _print_csv_result_streaming(command_result)
-                print(flush=True)
+                _print(flush=True)
             printed_end_line = True
         else:
             _stream_json(result)
@@ -315,8 +387,8 @@ def print_structured(
             if output_format == OutputFormat.CSV:
                 _print_csv_result_streaming(r)
             else:
-                json.dump(r, sys.stdout, cls=StreamingJSONEncoder)
-            print(flush=True)
+                json.dump(r, _result_stream(), cls=StreamingJSONEncoder)
+            _print(flush=True)
             printed_end_line = True
     else:
         if output_format == OutputFormat.CSV:
@@ -327,17 +399,17 @@ def print_structured(
 
     # Adds empty line at the end
     if not printed_end_line:
-        print(flush=True)
+        _print(flush=True)
 
 
 def _print_json_result_streaming(result: CommandResult):
     """Print a single CommandResult as JSON with streaming support"""
     if isinstance(result, CollectionResult):
         _stream_collection_as_json(result, indent=4)
     elif isinstance(result, (ObjectResult, MessageResult)):
-        json.dump(result, sys.stdout, cls=StreamingJSONEncoder, indent=4)
+        json.dump(result, _result_stream(), cls=StreamingJSONEncoder, indent=4)
     else:
-        json.dump(result, sys.stdout, cls=StreamingJSONEncoder, indent=4)
+        json.dump(result, _result_stream(), cls=StreamingJSONEncoder, indent=4)
 
 
 def _print_object_result_as_csv(result: ObjectResult):
@@ -349,7 +421,7 @@ def _print_object_result_as_csv(result: ObjectResult):
     data = result.result
     if isinstance(data, dict):
         writer = csv.DictWriter(
-            sys.stdout, fieldnames=list(data.keys()), lineterminator="\n"
+            _result_stream(), fieldnames=list(data.keys()), lineterminator="\n"
         )
         writer.writeheader()
         _write_csv_row(writer, data)
@@ -361,7 +433,9 @@ def _print_message_result_as_csv(result: MessageResult):
     Creates a simple CSV structure with one column named 'message'
     containing the sanitized message text.
     """
-    writer = csv.DictWriter(sys.stdout, fieldnames=["message"], lineterminator="\n")
+    writer = csv.DictWriter(
+        _result_stream(), fieldnames=["message"], lineterminator="\n"
+    )
     writer.writeheader()
     writer.writerow({"message": sanitize_for_terminal(result.message)})
 
@@ -381,32 +455,37 @@ def _stream_json(result):
     indent_size = 2
 
     class _Indented:
-        def __init__(self, stream: TextIO):
+        def __init__(self, stream: TextIO | TeeTextIO):
             self._stream = stream
 
         def write(self, text: str):
             return self._stream.write(indent(text, " " * indent_size))
 
-    print("[")
+    _print("[")
     results = result.result
     res = next(results, None)
     while res:
-        json.dump(res, _Indented(sys.stdout), cls=CustomJSONEncoder, indent=indent_size)  # type: ignore
+        json.dump(
+            res,
+            _Indented(_result_stream()),
+            cls=CustomJSONEncoder,
+            indent=indent_size,
+        )  # type: ignore
         if res := next(results, None):
-            print(",")
-    print("\n]")
+            _print(",")
+    _print("\n]")
 
 
 def print_unstructured(
     obj: CommandResult | None, output_format: OutputFormat | None = None
 ):
     """Handles outputs like table, plain text and other unstructured types."""
     if not obj:
-        rich_print("Done", flush=True)
+        _rich_print("Done", flush=True)
     elif not obj.result:
-        rich_print("No data", flush=True)
+        _rich_print("No data", flush=True)
     elif isinstance(obj, MessageResult):
-        rich_print(sanitize_for_terminal(obj.message), flush=True)
+        _rich_print(sanitize_for_terminal(obj.message), flush=True)
     else:
         if isinstance(obj, ObjectResult):
             _print_single_table(obj, output_format)
@@ -424,7 +503,7 @@ def _print_single_table(obj, output_format: OutputFormat | None = None):
         table.add_row(
             sanitize_for_terminal(str(key)), sanitize_for_terminal(__to_str(value))
         )
-    rich_print(table, flush=True)
+    _rich_print(table, flush=True)
 
 
 def print_result(cmd_result: CommandResult, output_format: OutputFormat | None = None):
@@ -442,3 +521,4 @@ def print_result(cmd_result: CommandResult, output_format: OutputFormat | None =
             print_unstructured(cmd_result, resolved_format)
         case _:
             raise ValueError(f"Unexpected type {type(cmd_result)}")
+    _result_stream().flush()
@@ -22,6 +22,7 @@
 
 import typer
 from click import UsageError
+from snowflake.cli._app.printing import print_result
 from snowflake.cli._plugins.sql.client_query_span import sql_client_query_span
 from snowflake.cli._plugins.sql.manager import SqlManager
 from snowflake.cli._plugins.sql.prompt_format import (
@@ -400,8 +401,12 @@ def execute_sql(
             list(cursors)
             return EmptyResult()
         if expected_results_cnt == 1:
-            results = list(cursors)
-            if not results:
+            cursors = iter(cursors)
+            try:
+                first = next(cursors)
+            except StopIteration:
                 return EmptyResult()
-            return QueryResult(results[0])
+            print_result(QueryResult(first))
+            list(cursors)
+            return EmptyResult()
         return MultipleResults((QueryResult(c) for c in cursors))