fix: correct sqb mv plurals and plain-language plan bounds

src/sqlbuild/cli/commands/_helpers/refactor/command.py

@@ -32,6 +32,7 @@
 from sqlbuild.compiler.refactoring.models import RefactorPlan, RefactorRequest
 from sqlbuild.compiler.refactoring.types import RefactorOperation, RefactorStatus
 from sqlbuild.presentation.classes.transient_status_reporter import TransientStatusReporter
+from sqlbuild.presentation.main.count_noun import format_count_noun
 from sqlbuild.presentation.main.supports_color import supports_color
 
 _STAGING_PREFIX: str = "sqb-refactor-"
@@ -75,7 +76,7 @@ def _run(
         if refactor_request.operation == RefactorOperation.RENAME_COLUMN
         else plan_model_refactor(project=before.project, request=refactor_request)
     )
-    status.complete(message=f"Planned edits to {len(plan.changes)} files.")
+    status.complete(message=f"Planned edits to {_files(plan)}.")
     if plan.blocking or plan.manual:
         status.error("Refused: some references cannot be rewritten safely.")
         return _finish(
@@ -135,9 +136,9 @@ def _run(
             diagnostics=(),
             use_color=use_color,
         )
-    status.start(f"Writing {len(plan.changes)} files...")
+    status.start(f"Writing {_files(plan)}...")
     commit_refactor_plan(project_dir=project_dir, originals=originals, plan=plan)
-    status.complete(message=f"Wrote {len(plan.changes)} files.")
+    status.complete(message=f"Wrote {_files(plan)}.")
     return _finish(
         request=request,
         plan=plan,
@@ -166,6 +167,10 @@ def _finish(
     return 0 if status in {RefactorStatus.APPLIED, RefactorStatus.DRY_RUN} else 1
 
 
+def _files(plan: RefactorPlan) -> str:
+    return format_count_noun(count=len(plan.changes), singular="file")
+
+
 def _refactor_request(*, request: RefactorCommandRequest) -> RefactorRequest:
     target: str = request.target.strip()
     if request.command == CliCommand.MV:

src/sqlbuild/cli/commands/_helpers/refactor/output.py

@@ -15,6 +15,7 @@
 )
 from sqlbuild.compiler.refactoring.types import RefactorOperation, RefactorStatus
 from sqlbuild.presentation.classes.cli_style import CliStyle
+from sqlbuild.presentation.main.count_noun import format_count_noun
 from sqlbuild.presentation.main.tree_connector import tree_connector
 
 _CONTINUE: str = "\u2502   "
@@ -197,8 +198,7 @@ def _diagnostic_lines(*, style: CliStyle, diagnostics: tuple[CompilerDiagnostic,
 
 
 def _status_line(*, style: CliStyle, plan: RefactorPlan, status: RefactorStatus) -> str:
-    count: int = len(plan.changes)
-    files: str = f"{count} file{'s' if count != 1 else ''}"
+    files: str = format_count_noun(count=len(plan.changes), singular="file")
     if status == RefactorStatus.APPLIED:
         return style.success_strong(f"Compiled: ok, {files} changed")
     if status == RefactorStatus.DRY_RUN:

src/sqlbuild/cli/output/_helpers/cursor_plan.py

@@ -12,7 +12,12 @@
 from sqlbuild.compiler.planner.main.execution.effective_microbatch_batch_size import (
     resolve_effective_microbatch_batch_size,
 )
-from sqlbuild.compiler.planner.models import CursorBounds, Duration, ModelPlanEntry
+from sqlbuild.compiler.planner.models import (
+    CursorBounds,
+    CursorInputRelation,
+    Duration,
+    ModelPlanEntry,
+)
 from sqlbuild.compiler.planner.types import (
     CursorGrain,
     CursorType,
@@ -136,6 +141,27 @@ def _count_batches(*, bounds: CursorBounds, batch_size: str, cursor_type: str) -
     return None
 
 
+def runtime_cursor_inputs_clause(*, entry: ModelPlanEntry) -> str:
+    """Name the cursor inputs this run brings up to date before the model's range is known."""
+
+    labels: list[str] = []
+    relation: CursorInputRelation
+    for relation in entry.cursor_input_relations:
+        if not relation.is_runtime_owned:
+            continue
+        name: str = relation.input_name or relation.relation
+        label: str = (
+            f"{relation.input_kind.value} {name}" if relation.input_kind is not None else name
+        )
+        if label not in labels:
+            labels.append(label)
+    if not labels:
+        return "once its inputs are up to date"
+    if len(labels) == 1:
+        return f"once {labels[0]} is up to date"
+    return f"once {', '.join(labels[:-1])} and {labels[-1]} are up to date"
+
+
 def append_microbatch_plan_detail(
     *, lines: list[str], details: CursorPlanDetails, entry: ModelPlanEntry
 ) -> list[str]:
@@ -160,7 +186,9 @@ def append_microbatch_plan_detail(
     if details.planned_batch_count is not None and details.effective_batch_size is not None:
         lines.append(f"    batches: {details.planned_batch_count} x {details.effective_batch_size}")
     if details.resolution_status == CursorResolutionStatus.DEFERRED:
-        lines.append("    batches: resolved at runtime after upstream models complete")
+        lines.append(
+            f"    batches: counted at run time, {runtime_cursor_inputs_clause(entry=entry)}"
+        )
     lines.append(f"    batch concurrency: {entry.batch_concurrency}")
     if entry.unaccounted_partition_policy is not None:
         lines.append(f"    unaccounted partition policy: {entry.unaccounted_partition_policy}")

src/sqlbuild/cli/output/_helpers/plan_text.py

@@ -11,6 +11,7 @@
 from sqlbuild.cli.output._helpers.cursor_plan import (
     append_microbatch_plan_detail,
     build_cursor_plan_details,
+    runtime_cursor_inputs_clause,
 )
 from sqlbuild.cli.output._helpers.selection_diagnostics import direct_selection_diagnostics_enabled
 from sqlbuild.cli.output.models import CursorPlanDetails
@@ -65,7 +66,6 @@
 _LEAF_KEY_SEPARATOR: str = ": "
 _LEAF_ROW_PREFIX: str = "    "
 _NESTED_ROW_PREFIX: str = "      "
-_RUNTIME_PROMINENT_MARKERS: tuple[str, ...] = ("runtime", "deferred")
 _STALE_INPUT_WARNING_TITLE: str = "Stale inputs detected"
 _TREE_EXEMPT_CHARS: frozenset[str] = frozenset({" ", "."})
 
@@ -949,7 +949,9 @@ def _append_cursor_detail(
     if entry.incremental_mode == IncrementalMode.MICROBATCH:
         lines = append_microbatch_plan_detail(lines=lines, details=details, entry=entry)
     if details.bounds_owner == CursorBoundsOwner.RUNTIME:
-        lines.append("    bounds: runtime-owned (model-backed cursor input)")
+        lines.append(
+            f"    bounds: computed at run time, {runtime_cursor_inputs_clause(entry=entry)}"
+        )
     elif details.resolution_status == CursorResolutionStatus.RESOLVED:
         lines.append("    bounds: planner-resolved")
     return lines
@@ -1645,10 +1647,7 @@ def _treeify_plan_lines(lines: list[str]) -> list[str]:
             )
             key, separator, value = content.partition(_LEAF_KEY_SEPARATOR)
             if separator:
-                rendered_value: str = value
-                if any(marker in value for marker in _RUNTIME_PROMINENT_MARKERS):
-                    rendered_value = style.section(value)
-                out.append(f"{_LEAF_ROW_PREFIX}{connector} {style.muted(key)}  {rendered_value}")
+                out.append(f"{_LEAF_ROW_PREFIX}{connector} {style.muted(key)}  {value}")
             elif content.endswith(":"):
                 out.append(f"{_LEAF_ROW_PREFIX}{connector} {style.muted(content)}")
             else:

src/sqlbuild/compiler/planner/_helpers/output/plan_entry.py

@@ -125,6 +125,7 @@
     ContractPolicy,
     CursorGrain,
     CursorType,
+    GraphResourceKind,
     IncrementalMode,
     IncrementalStrategy,
     MaterializationType,
@@ -1582,6 +1583,8 @@ def _build_cursor_input_relations(
                         ref.ref_kind == SqlReferenceKind.REF and ref.ref_name in model_locations
                     ),
                     is_runtime_produced=ref.ref_name in runtime_cursor_producer_names,
+                    input_name=ref.ref_name,
+                    input_kind=_cursor_input_kind(ref=ref, model_locations=model_locations),
                     terminal_cursor_start=(
                         parse(
                             raw=get_config_cursor_bound(
@@ -1764,6 +1767,22 @@ def _resolve_cursor_input_relation(
     return None
 
 
+def _cursor_input_kind(
+    *, ref: CompileSqlReference, model_locations: dict[str, CompiledRelationLocation]
+) -> GraphResourceKind | None:
+    """Return the resource kind a resolved cursor input reference reads."""
+
+    if ref.ref_kind == SqlReferenceKind.REF:
+        return (
+            GraphResourceKind.MODEL if ref.ref_name in model_locations else GraphResourceKind.SEED
+        )
+    if ref.ref_kind == SqlReferenceKind.SEED:
+        return GraphResourceKind.SEED
+    if ref.ref_kind == SqlReferenceKind.SOURCE:
+        return GraphResourceKind.SOURCE
+    return None
+
+
 def _resolve_cursor_input_grain(
     *,
     ref: CompileSqlReference,

src/sqlbuild/compiler/planner/models.py

@@ -550,6 +550,8 @@ class CursorInputRelation:
     is_runtime_produced: bool = False
     terminal_cursor_start: CursorScalar | None = None
     terminal_cursor_end: CursorScalar | None = None
+    input_name: str | None = None
+    input_kind: GraphResourceKind | None = None
 
     def __init__(
         self,
@@ -561,6 +563,8 @@ def __init__(
         is_runtime_produced: bool = False,
         terminal_cursor_start: CursorScalar | str | None = None,
         terminal_cursor_end: CursorScalar | str | None = None,
+        input_name: str | None = None,
+        input_kind: GraphResourceKind | None = None,
     ) -> None:
         object.__setattr__(self, "relation", relation)
         object.__setattr__(self, "cursor_column", cursor_column)
@@ -569,6 +573,8 @@ def __init__(
         object.__setattr__(self, "is_runtime_produced", is_runtime_produced)
         object.__setattr__(self, "terminal_cursor_start", terminal_cursor_start)
         object.__setattr__(self, "terminal_cursor_end", terminal_cursor_end)
+        object.__setattr__(self, "input_name", input_name)
+        object.__setattr__(self, "input_kind", input_kind)
         self.__post_init__()
 
     def __post_init__(self) -> None:

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

@@ -72,6 +72,7 @@ class SeedWatermarkBuildE2ETestCase:
     changed_seed: str
     expected_initial_rows: tuple[tuple[object, ...], ...]
     expected_changed_rows: tuple[tuple[object, ...], ...]
+    expected_plan_bounds: str
 
 
 @dataclass(frozen=True)
@@ -81,6 +82,7 @@ class LoaderWatermarkBuildE2ETestCase:
     changed_maximum: int
     expected_initial_rows: tuple[tuple[object, ...], ...]
     expected_changed_rows: tuple[tuple[object, ...], ...]
+    expected_plan_bounds: str
 
 
 @dataclass(frozen=True)

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

@@ -27,6 +27,9 @@
             changed_maximum=3,
             expected_initial_rows=((1, 100), (2, 200)),
             expected_changed_rows=((1, 100), (2, 200), (3, 300)),
+            expected_plan_bounds=(
+                "bounds  computed at run time, once source raw_orders is up to date"
+            ),
         )
     ],
     ids=lambda case: case.description,
@@ -105,6 +108,13 @@ def test_given_managed_loader_watermark_when_reloaded_then_incremental_consumes_
         sql="SELECT id, amount FROM main.raw_orders_incremental ORDER BY id",
     ) == list(test_case.expected_initial_rows)
 
+    plan_result: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "plan", "--select", "+raw_orders_incremental"),
+        project_dir=project_dir,
+    )
+    assert plan_result.returncode == 0, plan_result.stdout + plan_result.stderr
+    assert test_case.expected_plan_bounds in plan_result.stdout, plan_result.stdout
+
     maximum_path.write_text(str(test_case.changed_maximum), encoding="utf-8")
     changed_result: subprocess.CompletedProcess[str] = run_sqb(
         command=("--no-color", "build", "--reload", "--select", "+raw_orders_incremental"),

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

@@ -191,7 +191,7 @@
             rebuild_command=(),
             expected_plan_fragments=(
                 "full rebuild",
-                "bounds  runtime-owned (model-backed cursor input)",
+                "bounds  computed at run time, once model stg_orders is up to date",
                 "policy  replay_on_change=full",
             ),
         ),

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

@@ -27,6 +27,9 @@
             changed_seed="id,amount\n1,100\n2,200\n3,300\n",
             expected_initial_rows=((1, 100), (2, 200)),
             expected_changed_rows=((1, 100), (2, 200), (3, 300)),
+            expected_plan_bounds=(
+                "bounds  computed at run time, once seed order_events is up to date"
+            ),
         )
     ],
     ids=lambda case: case.description,
@@ -92,6 +95,12 @@ def test_given_seed_watermark_when_seed_changes_then_incremental_consumes_new_hi
         sql="SELECT id, amount FROM main.order_events_incremental ORDER BY id",
     ) == list(test_case.expected_initial_rows)
 
+    plan_result: subprocess.CompletedProcess[str] = run_sqb(
+        command=("--no-color", "plan"), project_dir=project_dir
+    )
+    assert plan_result.returncode == 0, plan_result.stdout + plan_result.stderr
+    assert test_case.expected_plan_bounds in plan_result.stdout, plan_result.stdout
+
     (project_dir / "seeds" / "order_events.csv").write_text(
         test_case.changed_seed, encoding="utf-8"
     )

tests/integration/src/sqlbuild/cli/commands/_helpers/refactor/__init__.py

@@ -0,0 +1 @@
+"""Refactor command integration tests."""

tests/integration/src/sqlbuild/cli/commands/_helpers/refactor/_test_types.py

@@ -0,0 +1,15 @@
+"""Test case types for refactor command integration tests."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass(frozen=True)
+class RefactorStatusMessagesTestCase:
+    description: str
+    command: str
+    target: str
+    new_name: str | None
+    destination: str | None
+    expected_lines: tuple[str, ...]