fix: no traceback after failed builds, and aligned test statuses

Makefile

@@ -92,6 +92,7 @@ E2E_DUCKDB_BUILD_CORE_PATHS := \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_selector_surface.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_table_function_dependency.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_template_expressions.py \
+	tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py \
 	tests/e2e/src/sqlbuild/cli/commands/main/build/test_validation_failures.py
 
 E2E_DUCKDB_BUILD_INCREMENTAL_PATHS := \

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

@@ -23,6 +23,7 @@
     SkillMaintenanceResult,
 )
 from sqlbuild.lint.exceptions import LintError
+from sqlbuild.presentation.main.transient_line_coordinator import shared_transient_line_coordinator
 from sqlbuild.rule_engine.exceptions import RulesError
 
 
@@ -37,7 +38,7 @@ def dispatch_and_handle_errors(
 
     _report_skill_freshness(invocation=invocation)
     try:
-        return dispatch_with_observability(args=args, handlers=handlers)
+        return _dispatch_clearing_transient_line(args=args, handlers=handlers)
     except SystemExit as error:
         return error.code if isinstance(error.code, int) else 1
     except (CliUserError, RulesError) as error:
@@ -80,6 +81,17 @@ def dispatch_and_handle_errors(
         return 1
 
 
+def _dispatch_clearing_transient_line(
+    *, args: CliNamespace, handlers: CliEntrypointHandlers
+) -> int:
+    """Stop a status line a failed command left running before its error is reported."""
+
+    try:
+        return dispatch_with_observability(args=args, handlers=handlers)
+    finally:
+        shared_transient_line_coordinator().clear_active()
+
+
 def _expected_command_failure_types() -> tuple[type[Exception], ...]:
     """Load command-specific failures only after a command raises."""
 

src/sqlbuild/cli/commands/_helpers/test/sql_progress.py

@@ -5,6 +5,7 @@
 import json
 from pathlib import Path
 
+from sqlbuild.cli.progress.constants import NESTED_CHILD_NAME_INDENT
 from sqlbuild.cli.progress.main._expectation_detail import format_expectation_detail
 from sqlbuild.cli.progress.main._expectation_name import format_expectation_name
 from sqlbuild.cli.progress.models import NestedProgressChildRow
@@ -21,6 +22,7 @@ def resolve_test_name_width(test_entries: tuple[SqlTestPlanEntry, ...]) -> int:
     """Resolve one shared width for test rows and nested expectation rows."""
 
     names: list[str] = []
+    child_names: list[str] = []
     entry: SqlTestPlanEntry
     for entry in test_entries:
         names.append(
@@ -31,9 +33,15 @@ def resolve_test_name_width(test_entries: tuple[SqlTestPlanEntry, ...]) -> int:
                 parameter_values=entry.parameter_values,
             )
         )
-        names.extend(f"expected {step.model_name}" for step in entry.chain if step.expected_cte_sql)
-        names.extend(f"assertion {assertion.name}" for assertion in entry.assertions)
-    return resolve_name_column_width(names=names, min_width=50)
+        child_names.extend(
+            f"expected {step.model_name}" for step in entry.chain if step.expected_cte_sql
+        )
+        child_names.extend(f"assertion {assertion.name}" for assertion in entry.assertions)
+    child_name_length: int = max((len(name) for name in child_names), default=0)
+    return max(
+        resolve_name_column_width(names=names, min_width=50),
+        child_name_length + NESTED_CHILD_NAME_INDENT,
+    )
 
 
 def build_test_expectation_rows(

src/sqlbuild/cli/commands/classes/build_progress_callbacks.py

@@ -67,6 +67,7 @@
 _MIN_NAME_WIDTH: int = 20
 _NAME_PADDING: int = 2
 _SUB_INDENT: int = 2
+_TREE_CHILD_NAME_INDENT: int = _SUB_INDENT + 4
 _HOOK_PHASE_WIDTH: int = 10
 _HOOK_TYPE_WIDTH: int = 8
 _HOOK_MIN_LABEL_WIDTH: int = 24
@@ -242,13 +243,17 @@ def __init__(
                 if getattr(chain_step, "expected_cte_sql", None):
                     max_name_len = max(
                         max_name_len,
-                        len(format_expectation_name(str(getattr(chain_step, "model_name", "")))),
+                        _tree_child_name_length(
+                            format_expectation_name(str(getattr(chain_step, "model_name", "")))
+                        ),
                     )
             assertion: object
             for assertion in getattr(test_entry, "assertions", ()):
                 max_name_len = max(
                     max_name_len,
-                    len(format_expectation_name(f"assertion {getattr(assertion, 'name', '')}")),
+                    _tree_child_name_length(
+                        format_expectation_name(f"assertion {getattr(assertion, 'name', '')}")
+                    ),
                 )
         self._name_width: int = max(max_name_len + _NAME_PADDING, _MIN_NAME_WIDTH)
 
@@ -704,7 +709,7 @@ def _write_failed_test_result(self, *, test_result: SqlTestExecutionResult) -> N
         )
         expectation_pad: str = " " * (self._prefix_width + _SUB_INDENT + 2)
         expectation_type_width: int = _TYPE_WIDTH - 2
-        sub_nw: int = self._name_width - _SUB_INDENT
+        tree_child_nw: int = self._name_width - _TREE_CHILD_NAME_INDENT
         step_result: StepResult
         for index, step_result in enumerate(test_result.step_results):
             expectation_status: str = self._style.status(
@@ -719,7 +724,7 @@ def _write_failed_test_result(self, *, test_result: SqlTestExecutionResult) -> N
             self._stream.write(
                 f"{expectation_pad}{connector} "
                 f"{self._style.muted(f'{"expect":<{expectation_type_width}}')}"
-                f"{expectation_name:<{sub_nw}} {expectation_status}{expectation_detail}\n"
+                f"{expectation_name:<{tree_child_nw}} {expectation_status}{expectation_detail}\n"
             )
         if test_result.error_message:
             self._write_error_detail(
@@ -784,7 +789,7 @@ def _write_error_detail(
         pad: str = " " * (self._prefix_width + child_offset)
         connector: str = tree_connector(style=self._style, last=True) if tree_child else ""
         connector_suffix: str = f"{connector} " if tree_child else ""
-        label_width: int = _TYPE_WIDTH - 4 if tree_child else _TYPE_WIDTH - 1
+        label_width: int = _TYPE_WIDTH - 3 if tree_child else _TYPE_WIDTH - 1
         label_padding: str = " " * max(0, label_width - len("error"))
         label: str = f"{self._style.error_muted('error')}{label_padding}"
         plain_prefix_width: int = len(pad) + (4 if tree_child else 0) + label_width + 1
@@ -1495,6 +1500,12 @@ def _format_display_sql(sql: str) -> str:
     return f"{stripped};"
 
 
+def _tree_child_name_length(name: str) -> int:
+    """Measure a tree child name so its status still fits the shared status column."""
+
+    return len(name) + _TREE_CHILD_NAME_INDENT - _NAME_PADDING
+
+
 def _truncate_name(*, name: str, width: int) -> str:
     """Truncate a display name to fit within the given width."""
 

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

@@ -9,6 +9,7 @@
     NativeProgressProjector,
     current_native_progress_projector,
 )
+from sqlbuild.cli.progress.constants import NESTED_CHILD_NAME_INDENT
 from sqlbuild.cli.progress.models import NestedProgressChildRow
 from sqlbuild.presentation.classes.cli_style import CliStyle
 from sqlbuild.presentation.classes.transient_line_coordinator import TransientLineCoordinator
@@ -155,14 +156,15 @@ def _write_completed_item(
         self._stream.write(
             f"    {self._label:<{_LABEL_WIDTH}}{item_name:<{self._name_width}} {status}{detail}\n"
         )
+        child_name_width: int = self._name_width - NESTED_CHILD_NAME_INDENT
         child_row: NestedProgressChildRow
         for index, child_row in enumerate(child_rows):
             child_status: str = self._style.status(status=child_row.status_text)
             last: bool = index == len(child_rows) - 1 and error_message is None
             connector: str = tree_connector(style=self._style, last=last)
             self._stream.write(
                 f"      {connector} {child_row.label:<{_LABEL_WIDTH - 2}}"
-                f"{child_row.name:<{self._name_width}} "
+                f"{child_row.name:<{child_name_width}} "
                 f"{child_status}{child_row.detail}\n"
             )
         if error_message is not None:

src/sqlbuild/cli/progress/constants.py

@@ -0,0 +1,3 @@
+"""Shared layout constants for CLI progress rendering."""
+
+NESTED_CHILD_NAME_INDENT: int = 4

src/sqlbuild/presentation/classes/transient_line_coordinator.py

@@ -28,6 +28,16 @@ def release(self, *, owner: TransientLineOwner) -> None:
             if self._active is not None and self._active[1] is owner:
                 self._active = None
 
+    def clear_active(self) -> None:
+        """Erase and forget any live transient line, so nothing redraws it afterwards."""
+
+        with self.lock:
+            if self._active is None:
+                return
+            owner: TransientLineOwner = self._active[1]
+            self._active = None
+            owner.clear_transient_line()
+
     def write_persistent(self, *, stream: TextIO, text: str) -> None:
         """Write lasting text, clearing and then redrawing any live transient line."""
 

src/sqlbuild/runtime/compute_logs/classes/binary_tee.py

@@ -35,6 +35,7 @@ def __init__(
         self._before_write: Callable[[], None] | None = before_write
         self._failure_callback: Callable[[Exception], None] | None = failure_callback
         self._storage_failed: bool = False
+        self._detached: bool = False
 
     def write(self, data: bytes | bytearray) -> int:
         """Preserve sink behavior while isolating capture append failures."""
@@ -53,7 +54,7 @@ def write(self, data: bytes | bytearray) -> int:
                 : accepted_write_count(result=result, offered_count=len(text))
             ]
             accepted = accepted_text.encode(self._encoding, self._errors)
-        if accepted and not self._storage_failed:
+        if accepted and not self._storage_failed and not self._detached:
             try:
                 self._storage.append(
                     invocation_id=self._invocation_id, stream=self._stream, data=accepted
@@ -64,6 +65,11 @@ def write(self, data: bytes | bytearray) -> int:
                     self._failure_callback(error)
         return result if isinstance(result, int) else len(offered)
 
+    def detach(self) -> None:
+        """Stop retaining writes once capture ends; late writers reach only the sink."""
+
+        self._detached = True
+
     def flush(self) -> None:
         """Flush the original sink without closing it."""
 

src/sqlbuild/runtime/compute_logs/classes/scoped_compute_log_capture.py

@@ -5,7 +5,6 @@
 import sys
 from collections.abc import Callable
 from datetime import UTC, datetime
-from functools import partial
 from typing import Any
 
 from sqlbuild.diagnostics.classes.invocation_diagnostic_routing import InvocationDiagnosticRouting
@@ -73,7 +72,7 @@ def run(self, *, operation: Callable[[], int]) -> int:
         return operation_result
 
     def _install(self) -> None:
-        write_failure: Callable[[Exception], None] = partial(self._report, channel="capture_write")
+        write_failure: Callable[[Exception], None] = self._report_write_failure
         self._stdout_tee = TextComputeLogTee(
             sink=self._original_stdout,
             storage=self._storage,
@@ -110,6 +109,9 @@ def _cleanup(self) -> bool:
                 self._report(error=error, channel="capture_flush")
         sys.stdout = self._original_stdout
         sys.stderr = self._original_stderr
+        for tee in (self._stdout_tee, self._stderr_tee):
+            if tee is not None:
+                tee.detach()
         if self._routing is not None:
             try:
                 self._routing.__exit__(None, None, None)
@@ -151,6 +153,9 @@ def _close_storage(self) -> None:
         except Exception as error:
             self._report(error=error, channel="capture_close")
 
+    def _report_write_failure(self, error: Exception) -> None:
+        self._report(error=error, channel="capture_write")
+
     def _report(self, *, error: Exception, channel: str) -> None:
         try:
             self._failure_callback(error, channel)

src/sqlbuild/runtime/compute_logs/classes/text_tee.py

@@ -30,6 +30,7 @@ def __init__(
         self._errors: str = getattr(sink, "errors", None) or "strict"
         self._failure_callback: Callable[[Exception], None] | None = failure_callback
         self._storage_failed: bool = False
+        self._detached: bool = False
         binary_sink: Any = getattr(sink, "buffer", sink)
         binary_uses_text_sink: bool = binary_sink is sink
         self._buffer: BinaryComputeLogTee = BinaryComputeLogTee(
@@ -49,7 +50,7 @@ def write(self, text: str) -> int:
 
         result: object = self._sink.write(text)
         accepted_count: int = accepted_write_count(result=result, offered_count=len(text))
-        if accepted_count and not self._storage_failed:
+        if accepted_count and not self._storage_failed and not self._detached:
             try:
                 emitted: bytes = text[:accepted_count].encode(self._encoding, self._errors)
                 self._storage.append(
@@ -61,6 +62,12 @@ def write(self, text: str) -> int:
                     self._failure_callback(error)
         return result if isinstance(result, int) else len(text)
 
+    def detach(self) -> None:
+        """Stop retaining writes once capture ends; late writers reach only the sink."""
+
+        self._detached = True
+        self._buffer.detach()
+
     def flush(self) -> None:
         """Flush the existing text sink without changing its ownership."""
 

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

@@ -1206,3 +1206,12 @@ class QueryFingerprintEditE2ETestCase:
     format_flags: tuple[str, ...]
     expected_format_summary: str
     expected_reasons: dict[str, str]
+
+
+@dataclass(frozen=True)
+class TerminalBuildFailureE2ETestCase:
+    description: str
+    command: tuple[str, ...]
+    model_replacements: tuple[tuple[str, str, str], ...]
+    expected_screen_fragments: tuple[str, ...]
+    columns: int = 120

tests/e2e/src/sqlbuild/cli/commands/main/build/test_terminal_failure.py

@@ -0,0 +1,74 @@
+"""E2E coverage for early build failures in a real terminal."""
+
+from __future__ import annotations
+
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from tests.e2e.src.sqlbuild.cli.commands.main.build._test_types import (
+    TerminalBuildFailureE2ETestCase,
+)
+from tests.e2e.src.sqlbuild.cli.commands.shared.helpers import (
+    SPINNER_GLYPHS,
+    prepare_waffle_shop,
+    render_terminal_screen,
+    run_sqb_with_pty,
+)
+
+
+@pytest.mark.parametrize(
+    "test_case",
+    [
+        TerminalBuildFailureE2ETestCase(
+            description="selected model with unbuilt upstreams fails during planning",
+            command=("build", "--select", "daily_revenue"),
+            model_replacements=(),
+            expected_screen_fragments=(
+                "Inspected warehouse state.",
+                "error[S301]: cannot build selected scope: 2 missing upstream dependencies",
+            ),
+        ),
+        TerminalBuildFailureE2ETestCase(
+            description="compile error fails before connecting",
+            command=("build",),
+            model_replacements=(
+                (
+                    "models/marts/daily_revenue.sql",
+                    '__ref("stg_payments")',
+                    '__ref("missing_payments")',
+                ),
+            ),
+            expected_screen_fragments=("error[", "missing_payments"),
+        ),
+    ],
+    ids=lambda case: case.description,
+)
+def test_given_early_build_failure_when_running_in_terminal_then_exits_with_error_only(
+    tmp_path: Path, test_case: TerminalBuildFailureE2ETestCase
+) -> None:
+    project_dir: Path = prepare_waffle_shop(tmp_path)
+    for relative_path, old, new in test_case.model_replacements:
+        model_path: Path = project_dir / relative_path
+        model_path.write_text(
+            model_path.read_text(encoding="utf-8").replace(old, new), encoding="utf-8"
+        )
+
+    result: subprocess.CompletedProcess[str] = run_sqb_with_pty(
+        command=test_case.command, project_dir=project_dir, columns=test_case.columns
+    )
+    screen: tuple[str, ...] = render_terminal_screen(
+        output=result.stdout, columns=test_case.columns
+    )
+    rendered: str = "\n".join(screen)
+
+    assert result.returncode == 1, rendered
+    assert "Traceback" not in result.stdout, rendered
+    assert "Exception ignored" not in result.stdout, rendered
+    assert not any(glyph in rendered for glyph in SPINNER_GLYPHS), 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/commands/shared/_helpers/_test_types.py

@@ -176,3 +176,12 @@ class NestedProgressConcurrentTestCase:
     description: str
     concurrency: int
     expected_output: str
+
+
+@dataclass(frozen=True)
+class SqlTestStatusAlignmentTestCase:
+    description: str
+    test_name: str
+    expected_models: tuple[str, ...]
+    assertion_names: tuple[str, ...]
+    expected_row_count: int