fix: show statement and step progress rows only with --debug

src/sqlbuild/cli/commands/_helpers/entry/observability.py

@@ -90,6 +90,7 @@ def cli_observability_scope(*, args: CliNamespace, project_dir: Path) -> Iterato
         if machine_output or args.debug or args.command == DEBUG_COMMAND
         else sys.stdout,
         use_color=not machine_output and not args.no_color and supports_color(),
+        debug=args.debug,
     )
     unsubscribe_progress: Unsubscribe = dispatcher.subscribe_lifecycle(
         subscriber=projector.consume,

src/sqlbuild/cli/commands/_helpers/seed/execution.py

@@ -4,6 +4,7 @@
 
 import time
 from collections.abc import Callable
+from dataclasses import replace
 from typing import TextIO
 
 from sqlbuild.cli.commands.models import (
@@ -13,6 +14,10 @@
 )
 from sqlbuild.cli.output.classes.execution_event_writer import ExecutionEventWriter
 from sqlbuild.cli.progress.classes.connection_progress_reporter import ConnectionProgressReporter
+from sqlbuild.cli.progress.classes.native_progress_projector import (
+    NativeProgressProjector,
+    current_native_progress_projector,
+)
 from sqlbuild.compiler.planner.models import PlanOutput
 from sqlbuild.executor.build.models import SeedExecutionResult
 from sqlbuild.executor.build.types import ExecutionStatus
@@ -82,7 +87,17 @@ def _build_on_complete(
     event_writer: ExecutionEventWriter,
     plan: PlanOutput,
 ) -> Callable[[SeedExecutionResult], None]:
+    projector: NativeProgressProjector | None = current_native_progress_projector()
+    if projector is not None:
+        projector.configure_resources(ordinals=seed_order, total=total_count)
+
     def _on_complete(result: SeedExecutionResult) -> None:
+        if projector is not None:
+            canonical_duration_ms: float | None = projector.consume_resource_terminal(
+                resource_name=result.seed_name, resource_id=f"seed:{result.seed_name}"
+            )
+            if canonical_duration_ms is not None:
+                result = replace(result, duration_ms=int(canonical_duration_ms))
         status_text: str = "OK" if result.status == ExecutionStatus.SUCCESS else "FAIL"
         style: CliStyle = CliStyle(use_color=use_color)
         status: str = style.status(status=status_text)

src/sqlbuild/cli/progress/classes/native_progress_projector.py

@@ -90,8 +90,9 @@
 class NativeProgressProjector:
     """Project canonical starts and correlate enriched terminal presentation."""
 
-    def __init__(self, *, stream: TextIO, use_color: bool) -> None:
+    def __init__(self, *, stream: TextIO, use_color: bool, debug: bool = False) -> None:
         self._stream: TextIO = stream
+        self._debug: bool = debug
         self._lines: TransientLineCoordinator = shared_transient_line_coordinator()
         self._style: CliStyle = CliStyle(use_color=use_color)
         self._is_tty: bool = hasattr(stream, "isatty") and stream.isatty()
@@ -104,6 +105,7 @@ def __init__(self, *, stream: TextIO, use_color: bool) -> None:
         self._resource_ids_by_name: defaultdict[str, set[str]] = defaultdict(set)
         self._rendered_terminal_ids: set[str] = set()
         self._operation_starts: dict[str, LifecycleEvent] = {}
+        self._quiet_operation_ids: set[str] = set()
         self._enriched_resource_names: set[str] = set()
         self._resource_names: dict[str, str] = {}
         self._statement_starts: dict[str, LifecycleEvent] = {}
@@ -249,7 +251,7 @@ def _consume_resource_start(self, event: LifecycleEvent) -> None:
         self._resource_names[resource_id] = resource_name
         self._render_prior_attempts(resource_id=resource_id, resource_name=resource_name)
         self._attempts_by_id[attempt_id] = event
-        if self._is_tty and resource_name in self._enriched_resource_names:
+        if self._is_tty and (resource_name in self._enriched_resource_names or not self._debug):
             return
         ordinal: int | None = self._resource_ordinals.get(resource_name)
         counter: str = ""
@@ -367,6 +369,9 @@ def _consume_operation_start(self, event: LifecycleEvent) -> None:
         ):
             return
         self._operation_starts[operation_id] = event
+        if event.resource_attempt_id is not None and not self._debug:
+            self._quiet_operation_ids.add(operation_id)
+            return
         status: str = self._style.status(status="START")
         prefix: str = "    " if event.resource_attempt_id is not None else ""
         attempt: str = _operation_attempt_label(event)
@@ -378,9 +383,14 @@ def _consume_operation_terminal(self, event: LifecycleEvent) -> None:
             return
         start: LifecycleEvent = self._operation_starts.pop(operation_id)
         operation_name: object = start.payload.get("operation_name")
+        failed: bool = event.event_type.endswith("failed")
+        if operation_id in self._quiet_operation_ids:
+            self._quiet_operation_ids.discard(operation_id)
+            if not failed:
+                return
         if not isinstance(operation_name, str):
             return
-        status_text: str = "FAIL" if event.event_type.endswith("failed") else "OK"
+        status_text: str = "FAIL" if failed else "OK"
         status: str = self._style.status(status=status_text)
         duration: object = event.payload.get("duration_ms")
         elapsed: str = (
@@ -413,7 +423,8 @@ def _consume_statement_start(self, event: LifecycleEvent) -> None:
         if statement_id is None or event.resource_id not in self._resource_names:
             return
         self._statement_starts[statement_id] = event
-        self._write(f"    statement  {self._statement_context(event)}  START")
+        if self._debug:
+            self._write(f"    statement  {self._statement_context(event)}  START")
 
     def _consume_statement_submitted(self, event: LifecycleEvent) -> None:
         statement_id: str | None = event.statement_id
@@ -427,6 +438,8 @@ def _consume_statement_submitted(self, event: LifecycleEvent) -> None:
         if self._statement_query_ids.get(statement_id) == query_id:
             return
         self._statement_query_ids[statement_id] = query_id
+        if not self._debug:
+            return
         start: LifecycleEvent = self._statement_starts.get(statement_id, event)
         self._write(f"    statement  {self._statement_context(start)}  query_id={query_id}")
 

tests/e2e/src/sqlbuild/cli/commands/main/test/_test_types.py

@@ -199,6 +199,14 @@ class TerminalTestProgressE2ETestCase:
     expected_summary: str
 
 
+@dataclass(frozen=True)
+class TerminalStatementRowsE2ETestCase:
+    description: str
+    columns: int
+    command: tuple[str, ...]
+    expected_statement_rows_visible: bool
+
+
 @dataclass(frozen=True)
 class ChainCteScopeTestCase:
     description: str
@@ -222,3 +230,13 @@ class CommentedFixtureE2ETestCase:
     test_files: dict[str, str]
     expected_exit_code: int
     expected_output_fragments: tuple[str, ...]
+
+
+@dataclass(frozen=True)
+class TerminalStepRowsE2ETestCase:
+    description: str
+    columns: int
+    command: tuple[str, ...]
+    expected_absent_fragments: tuple[str, ...]
+    expected_absent_line_prefixes: tuple[str, ...]
+    expected_screen_fragments: tuple[str, ...]

tests/e2e/src/sqlbuild/cli/commands/main/test/test_terminal_progress.py

@@ -8,6 +8,8 @@
 import pytest
 
 from tests.e2e.src.sqlbuild.cli.commands.main.test._test_types import (
+    TerminalStatementRowsE2ETestCase,
+    TerminalStepRowsE2ETestCase,
     TerminalTestProgressE2ETestCase,
 )
 from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import (
@@ -19,6 +21,15 @@
     run_sqb_with_pty,
 )
 
+_STEP_ROW_FRAGMENTS: tuple[str, ...] = (
+    "START",
+    "Create staging relation",
+    "Promote relation",
+    "loader callable",
+    "Audit evaluation",
+    "SQL test assertion",
+)
+
 
 @pytest.mark.parametrize(
     "test_case",
@@ -83,5 +94,129 @@ def test_given_built_project_when_testing_in_terminal_then_no_spinner_row_surviv
         assert test_case.expected_summary in rendered
 
 
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        TerminalStatementRowsE2ETestCase(
+            description="test without debug shows no statement rows",
+            columns=120,
+            command=("test",),
+            expected_statement_rows_visible=False,
+        ),
+        TerminalStatementRowsE2ETestCase(
+            description="build without debug shows no statement rows",
+            columns=120,
+            command=("build",),
+            expected_statement_rows_visible=False,
+        ),
+        TerminalStatementRowsE2ETestCase(
+            description="test with debug shows statement start rows",
+            columns=120,
+            command=("--debug", "test"),
+            expected_statement_rows_visible=True,
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_built_project_when_running_in_terminal_then_statement_rows_follow_debug_flag(
+    tmp_path: Path, test_case: TerminalStatementRowsE2ETestCase
+) -> None:
+    project_dir: Path = prepare_waffle_shop(tmp_path)
+    built: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "build"), project_dir=project_dir
+    )
+
+    result: subprocess.CompletedProcess[str] = run_sqb_with_pty(
+        command=test_case.command,
+        project_dir=project_dir,
+        columns=test_case.columns,
+        timeout_seconds=180.0,
+    )
+
+    assert built.returncode == 0, built.stdout + built.stderr
+    screen: tuple[str, ...] = render_terminal_screen(
+        output=result.stdout, columns=test_case.columns
+    )
+    rendered: str = "\n".join(screen)
+    assert result.returncode == 0, rendered
+    statement_row_count: int = sum(line.lstrip().startswith("statement  model=") for line in screen)
+    assert (statement_row_count > 0) is test_case.expected_statement_rows_visible, rendered
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        TerminalStepRowsE2ETestCase(
+            description="passing build shows result rows without start or step rows",
+            columns=120,
+            command=("build",),
+            expected_absent_fragments=_STEP_ROW_FRAGMENTS,
+            expected_absent_line_prefixes=(),
+            expected_screen_fragments=(
+                "seed      waffle_types",
+                "✓ Completed successfully",
+            ),
+        ),
+        TerminalStepRowsE2ETestCase(
+            description="passing seed shows one result row per seed",
+            columns=120,
+            command=("seed",),
+            expected_absent_fragments=_STEP_ROW_FRAGMENTS,
+            expected_absent_line_prefixes=("  seed      waffle_types",),
+            expected_screen_fragments=("1/1  seed      waffle_types", "✓ Completed successfully"),
+        ),
+        TerminalStepRowsE2ETestCase(
+            description="passing test shows no start or step rows",
+            columns=120,
+            command=("test",),
+            expected_absent_fragments=_STEP_ROW_FRAGMENTS,
+            expected_absent_line_prefixes=(),
+            expected_screen_fragments=("PASS=",),
+        ),
+        TerminalStepRowsE2ETestCase(
+            description="debug build shows start and step rows",
+            columns=120,
+            command=("--debug", "build"),
+            expected_absent_fragments=(),
+            expected_absent_line_prefixes=(),
+            expected_screen_fragments=(
+                "Create staging relation  START",
+                "Create staging relation  OK",
+                "loader callable  START",
+            ),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_built_project_when_running_in_terminal_then_step_rows_follow_debug_flag(
+    tmp_path: Path, test_case: TerminalStepRowsE2ETestCase
+) -> None:
+    project_dir: Path = prepare_waffle_shop(tmp_path)
+    built: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "build"), project_dir=project_dir
+    )
+
+    result: subprocess.CompletedProcess[str] = run_sqb_with_pty(
+        command=test_case.command,
+        project_dir=project_dir,
+        columns=test_case.columns,
+        timeout_seconds=180.0,
+    )
+
+    assert built.returncode == 0, built.stdout + built.stderr
+    screen: tuple[str, ...] = render_terminal_screen(
+        output=result.stdout, columns=test_case.columns
+    )
+    rendered: str = "\n".join(screen)
+    assert result.returncode == 0, rendered
+    assert not any(fragment in rendered for fragment in test_case.expected_absent_fragments), (
+        rendered
+    )
+    assert not any(line.startswith(test_case.expected_absent_line_prefixes) for line in screen), (
+        rendered
+    )
+    assert all(fragment in rendered for fragment in test_case.expected_screen_fragments), rendered
+
+
 if __name__ == "__main__":
     pytest.main([__file__, "-vv"])

tests/unit/src/sqlbuild/cli/progress/classes/_test_types.py

@@ -47,6 +47,7 @@ class StatementProgressCase:
     adapter: str
     query_id: str | None
     expected_context: str
+    debug: bool
 
 
 @dataclass(frozen=True)
@@ -71,3 +72,11 @@ class FreshnessMetadataProjectionCase:
     terminal_event_type: str
     terminal_payload: dict[str, object]
     expected_terminal_line: str
+
+
+@dataclass(frozen=True)
+class StepProgressCase:
+    description: str
+    tty: bool
+    debug: bool
+    expected_line_prefixes: tuple[str, ...]