\dvx status\: group by status + \-s\/\-x\ filters

Group output under per-status headers (Stale → Missing → Transitive →
Error → Fresh), each with a count and the status's color. Flatten with
\`-G\`/\`--no-group\`.

Add \`-s\`/\`--status\` (include) and \`-x\`/\`--omit\` (exclude), both
comma-separated and prefix-matched (\`s\` → stale, \`m\` → missing, etc.).
JSON/YAML now respect filters too; the summary line keeps reporting
full unfiltered counts.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: ryan-williams

src/dvx/cli/status.py

@@ -209,19 +209,51 @@ def _mark_transitive_staleness(results: list[dict], target_list: list) -> None:
                 queue.append(dependent)
 
 
+STATUS_NAMES = ["fresh", "stale", "missing", "error", "transitive"]
+GROUP_ORDER = ["stale", "missing", "transitive", "error", "fresh"]
+
+
+def _resolve_status_list(value: str | None) -> set[str] | None:
+    """Resolve a comma-separated list of status names (with prefix matching) to a set.
+
+    Returns None if value is None/empty. Raises click.BadParameter on ambiguous or unknown prefixes.
+    """
+    if not value:
+        return None
+    result = set()
+    for raw in value.split(","):
+        token = raw.strip().lower()
+        if not token:
+            continue
+        matches = [s for s in STATUS_NAMES if s.startswith(token)]
+        if not matches:
+            raise click.BadParameter(f"unknown status {token!r} (expected one of {STATUS_NAMES})")
+        if len(matches) > 1:
+            raise click.BadParameter(f"ambiguous status prefix {token!r}: matches {matches}")
+        result.add(matches[0])
+    return result
+
+
 @click.command()
 @click.argument("targets", nargs=-1)
 @click.option("-d", "--with-deps", is_flag=True, default=True, help="Check upstream dependencies.")
+@click.option("-G", "--no-group", is_flag=True, help="Don't group output by status.")
 @click.option("-j", "--jobs", type=int, default=None, help="Number of parallel workers.")
+@click.option("-N", "--no-transitive", is_flag=True, help="Hide transitively stale stages.")
+@click.option("-s", "--status", "status_filter", default=None, help="Show only these statuses (comma-sep, prefix-matched, e.g. 's,m').")
 @click.option("-v", "--verbose", is_flag=True, help="Show all files including fresh.")
+@click.option("-x", "--omit", default=None, help="Exclude these statuses (comma-sep, prefix-matched, e.g. 'm').")
 @click.option("--json", "as_json", is_flag=True, help="Output results as JSON.")
-@click.option("-N", "--no-transitive", is_flag=True, help="Hide transitively stale stages.")
 @click.option("-y", "--yaml", "as_yaml", is_flag=True, help="Output detailed results as YAML (includes before/after hashes).")
-def status(targets, with_deps, jobs, verbose, as_json, no_transitive, as_yaml):
+def status(targets, with_deps, no_group, jobs, no_transitive, status_filter, verbose, omit, as_json, as_yaml):
     """Check freshness status of artifacts.
 
-    By default, only shows stale/missing files (like git status).
-    Use -v/--verbose to show all files including fresh ones.
+    By default, only shows stale/missing files (like git status), grouped by status.
+    Use -v/--verbose to also include fresh files.
+    Use -s/--status to include only specific statuses (e.g. -s stale,missing).
+    Use -x/--omit to exclude specific statuses (e.g. -x missing).
+    Status names support prefix matching: 's' → stale, 'm' → missing, etc.
+    Use -G/--no-group to flatten output (paths sorted, no per-status sections).
     Use -y/--yaml for detailed output with before/after hashes for changed deps.
 
     Examples:
@@ -231,11 +263,16 @@ def status(targets, with_deps, jobs, verbose, as_json, no_transitive, as_yaml):
         dvx status -j 4              # Use 4 parallel workers
         dvx status --json            # Output as JSON
         dvx status -y                # Detailed YAML with hashes
+        dvx status -x m              # Hide missing files
+        dvx status -s s,t            # Show only stale and transitive
     """
     import json as json_module
     from concurrent.futures import ThreadPoolExecutor, as_completed
     from functools import partial
 
+    include = _resolve_status_list(status_filter)
+    exclude = _resolve_status_list(omit) or set()
+
     # Find targets - expand directories to .dvc files
     if targets:
         target_list = _expand_targets(targets)
@@ -270,50 +307,78 @@ def status(targets, with_deps, jobs, verbose, as_json, no_transitive, as_yaml):
         _mark_transitive_staleness(results, target_list)
 
     results.sort(key=lambda r: r["path"])
-    transitive_count = sum(1 for r in results if r["status"] == "transitive")
-    stale_count = sum(1 for r in results if r["status"] == "stale")
-    missing_count = sum(1 for r in results if r["status"] == "missing")
-    fresh_count = sum(1 for r in results if r["status"] == "fresh")
-    error_count = sum(1 for r in results if r["status"] == "error")
+
+    # Counts from the full, unfiltered set (for the summary line)
+    counts = {s: sum(1 for r in results if r["status"] == s) for s in STATUS_NAMES}
+
+    # Compute the visible set. Precedence: -s overrides default; -v adds fresh to default;
+    # -x always subtracts.
+    if include is not None:
+        visible = set(include)
+    else:
+        visible = set(STATUS_NAMES) if verbose else {"stale", "missing", "error", "transitive"}
+    visible -= exclude
+
+    filtered = [r for r in results if r["status"] in visible]
 
     if as_yaml:
         import yaml
-        # Filter to non-fresh unless verbose
-        if not verbose:
-            results = [r for r in results if r["status"] != "fresh"]
-        # Convert to dict keyed by path for nicer YAML
         yaml_data = {}
-        for r in results:
+        for r in filtered:
             path = r.pop("path")
-            # Remove None values for cleaner output
             yaml_data[path] = {k: v for k, v in r.items() if v is not None}
         click.echo(yaml.dump(yaml_data, default_flow_style=False, sort_keys=False))
-    elif as_json:
-        click.echo(json_module.dumps(results, indent=2))
+        return
+
+    if as_json:
+        click.echo(json_module.dumps(filtered, indent=2))
+        return
+
+    status_style = {
+        "fresh": ("✓", "green"),
+        "stale": ("✗", "red"),
+        "missing": ("?", "magenta"),
+        "error": ("!", "red"),
+        "transitive": ("⚠", "yellow"),
+    }
+
+    def _render(r):
+        icon, color = status_style.get(r["status"], ("?", "red"))
+        styled_icon = click.style(icon, fg=color)
+        line = f"{styled_icon} {r['path']}"
+        if r.get("reason"):
+            line += click.style(f" ({r['reason']})", fg="bright_black")
+        return line
+
+    if no_group:
+        for r in filtered:
+            click.echo(_render(r))
     else:
-        # By default, only show non-fresh files (like git status)
-        status_style = {
-            "fresh": ("✓", "green"),
-            "stale": ("✗", "red"),
-            "missing": ("?", "magenta"),
-            "error": ("!", "red"),
-            "transitive": ("⚠", "yellow"),
-        }
-        for r in results:
-            if r["status"] == "fresh" and not verbose:
+        first = True
+        for s in GROUP_ORDER:
+            if s not in visible:
+                continue
+            group = [r for r in filtered if r["status"] == s]
+            if not group:
                 continue
-            icon, color = status_style.get(r["status"], ("?", "red"))
-            styled_icon = click.style(icon, fg=color)
-            line = f"{styled_icon} {r['path']}"
-            if r.get("reason"):
-                line += click.style(f" ({r['reason']})", fg="bright_black")
-            click.echo(line)
-
-        # Summary line
-        parts = [f"Fresh: {fresh_count}", f"Stale: {stale_count}"]
-        if transitive_count:
-            parts.append(f"Transitively stale: {transitive_count}")
-        click.echo(f"\n{', '.join(parts)}")
+            if not first:
+                click.echo()
+            first = False
+            _, color = status_style[s]
+            header = click.style(f"{s.capitalize()} ({len(group)}):", fg=color, bold=True)
+            click.echo(header)
+            for r in group:
+                click.echo(f"  {_render(r)}")
+
+    # Summary line (always reflects the full set, not filtered)
+    parts = [f"Fresh: {counts['fresh']}", f"Stale: {counts['stale']}"]
+    if counts["missing"]:
+        parts.append(f"Missing: {counts['missing']}")
+    if counts["transitive"]:
+        parts.append(f"Transitively stale: {counts['transitive']}")
+    if counts["error"]:
+        parts.append(f"Error: {counts['error']}")
+    click.echo(f"\n{', '.join(parts)}")
 
 
 # Export the command

tests/test_cli.py

@@ -340,7 +340,7 @@ def test_status_shows_fresh_and_stale(runner, temp_dvc_repo):
     lines = result.output.strip().split("\n")
 
     # Find the stale.txt line - should have ✗ and show data changed
-    stale_lines = [l for l in lines if "stale.txt" in l]
+    stale_lines = [l.lstrip() for l in lines if "stale.txt" in l]
     assert len(stale_lines) == 1
     stale_line = stale_lines[0]
     assert stale_line.startswith("✗")
@@ -365,7 +365,7 @@ def test_status_json_output(runner, temp_dvc_repo):
             "outs": [{"md5": file_hash, "size": 10, "hash": "md5", "path": "data.txt"}]
         }, f)
 
-    result = runner.invoke(cli, ["status", "--json"])
+    result = runner.invoke(cli, ["status", "--json", "-v"])
     assert result.exit_code == 0
 
     assert json.loads(result.output) == [
@@ -415,7 +415,7 @@ def test_status_dep_changed(runner, temp_dvc_repo):
     lines = result.output.strip().split("\n")
 
     # Find output.txt line - should show dep changed, not data changed
-    output_lines = [l for l in lines if "output.txt" in l]
+    output_lines = [l.lstrip() for l in lines if "output.txt" in l]
     assert len(output_lines) == 1
     output_line = output_lines[0]
     assert output_line.startswith("✗")
@@ -511,3 +511,119 @@ def test_status_transitive_staleness(runner, tmp_path):
     assert "⚠" in result.output
     assert "step_b" in result.output
     assert "upstream stale" in result.output
+
+
+@pytest.fixture
+def mixed_status_repo(tmp_path):
+    """Repo with one stale, one missing, one fresh .dvc file."""
+    os.chdir(tmp_path)
+    (tmp_path / ".dvc").mkdir()
+
+    from dvx.run.hash import compute_md5
+
+    # Fresh
+    f = tmp_path / "fresh.txt"
+    f.write_text("fresh\n")
+    with open(tmp_path / "fresh.txt.dvc", "w") as fp:
+        yaml.dump({"outs": [{"md5": compute_md5(f), "size": f.stat().st_size, "path": "fresh.txt"}]}, fp)
+
+    # Stale
+    s = tmp_path / "stale.txt"
+    s.write_text("stale\n")
+    with open(tmp_path / "stale.txt.dvc", "w") as fp:
+        yaml.dump({"outs": [{"md5": "0" * 32, "size": 5, "path": "stale.txt"}]}, fp)
+
+    # Missing
+    with open(tmp_path / "missing.txt.dvc", "w") as fp:
+        yaml.dump({"outs": [{"md5": "1" * 32, "size": 10, "path": "missing.txt"}]}, fp)
+
+    return tmp_path
+
+
+def test_status_grouped_by_default(runner, mixed_status_repo):
+    """Default output groups stale / missing under headers."""
+    result = runner.invoke(cli, ["status"])
+    assert result.exit_code == 0
+
+    out = result.output
+    stale_idx = out.index("Stale (1)")
+    missing_idx = out.index("Missing (1)")
+    # Stale group appears before missing per GROUP_ORDER
+    assert stale_idx < missing_idx
+    assert "stale.txt.dvc" in out
+    assert "missing.txt.dvc" in out
+    # Fresh hidden by default
+    assert "Fresh (" not in out
+    assert "fresh.txt.dvc" not in out
+
+
+def test_status_no_group(runner, mixed_status_repo):
+    """-G disables grouping; no headers."""
+    result = runner.invoke(cli, ["status", "-G"])
+    assert result.exit_code == 0
+    assert "Stale (" not in result.output
+    assert "Missing (" not in result.output
+    assert "stale.txt.dvc" in result.output
+    assert "missing.txt.dvc" in result.output
+
+
+def test_status_omit_missing(runner, mixed_status_repo):
+    """-x missing hides missing paths."""
+    result = runner.invoke(cli, ["status", "-x", "missing"])
+    assert result.exit_code == 0
+    assert "stale.txt.dvc" in result.output
+    assert "missing.txt.dvc" not in result.output
+    assert "Missing (" not in result.output
+
+
+def test_status_omit_prefix(runner, mixed_status_repo):
+    """-x m (prefix) also hides missing."""
+    result = runner.invoke(cli, ["status", "-x", "m"])
+    assert result.exit_code == 0
+    assert "missing.txt.dvc" not in result.output
+
+
+def test_status_include_only(runner, mixed_status_repo):
+    """-s stale shows only stale, hides missing even though not omitted."""
+    result = runner.invoke(cli, ["status", "-s", "stale"])
+    assert result.exit_code == 0
+    assert "stale.txt.dvc" in result.output
+    assert "missing.txt.dvc" not in result.output
+    assert "fresh.txt.dvc" not in result.output
+
+
+def test_status_include_prefix_comma_sep(runner, mixed_status_repo):
+    """-s s,m accepts comma-separated prefixes."""
+    result = runner.invoke(cli, ["status", "-s", "s,m"])
+    assert result.exit_code == 0
+    assert "stale.txt.dvc" in result.output
+    assert "missing.txt.dvc" in result.output
+    assert "fresh.txt.dvc" not in result.output
+
+
+def test_status_unknown_status(runner, mixed_status_repo):
+    """Unknown status name is rejected."""
+    result = runner.invoke(cli, ["status", "-s", "bogus"])
+    assert result.exit_code != 0
+    assert "unknown status" in result.output
+
+
+def test_status_json_respects_filter(runner, mixed_status_repo):
+    """JSON output respects -s filter."""
+    import json
+    result = runner.invoke(cli, ["status", "-s", "stale", "--json"])
+    assert result.exit_code == 0
+    data = json.loads(result.output)
+    statuses = {r["status"] for r in data}
+    assert statuses == {"stale"}
+
+
+def test_status_summary_includes_all_counts(runner, mixed_status_repo):
+    """Summary line reflects full unfiltered set even when filtered."""
+    result = runner.invoke(cli, ["status", "-s", "stale"])
+    assert result.exit_code == 0
+    summary = result.output.strip().split("\n")[-1]
+    # Full counts: 1 fresh, 1 stale, 1 missing
+    assert "Fresh: 1" in summary
+    assert "Stale: 1" in summary
+    assert "Missing: 1" in summary