Reject a bad --summary-format before running the command

CLAUDE.md

@@ -32,6 +32,8 @@ This file provides guidance to Claude Code (claude.ai/code) when working with co
 - Types: Use strict typing with annotations; `from __future__ import annotations`
 - Naming: Classes (PascalCase), functions/variables (snake_case), constants (UPPER_SNAKE_CASE)
 - Error handling: Use explicit try/except with informative messages; log exceptions
+- Fail fast on bad user input: validate before the command runs and exit non-zero, rather than silently falling back
+- After the command has run, a duct failure must not cost the user the run: prefer loud fallbacks (log the error, keep the logs, preserve the command's exit code) over raising, since the user should not have to rerun a potentially expensive execution
 - prefer `Pathlib` over `os.path`
 - Documentation: Use indented docstrings with complete parameter descriptions
 

src/con_duct/_duct_main.py

@@ -193,5 +193,21 @@ def execute(
         )
         remove_files(log_paths)
     else:
-        lgr.info(report.execution_summary_formatted)
+        try:
+            summary = report.execution_summary_formatted
+        except Exception as e:
+            # The command has already run: keep its exit code and logs
+            lgr.error(
+                "Could not render the summary with %r (%s: %s)%s",
+                summary_format,
+                type(e).__name__,
+                e,
+                (
+                    f"; the execution summary is in {log_paths.info}"
+                    if record_types.has_system_summary()
+                    else ""
+                ),
+            )
+        else:
+            lgr.info(summary)
     return report.process.returncode

src/con_duct/_formatter.py

@@ -2,13 +2,9 @@
 
 from __future__ import annotations
 from datetime import datetime
-import logging
 import string
 from typing import Any
 
-lgr = logging.getLogger("con-duct")
-
-
 # Decimal (SI) byte units shared by SummaryFormatter.naturalsize (run
 # summary) and the plot axis formatter, so "kB" means the same thing in
 # both places.
@@ -165,15 +161,49 @@ def format_field(self, value: Any, format_spec: str) -> Any:
             format_spec, conversion = format_spec.split("!", 1)
         else:
             conversion = None
-        try:
-            value_ = super().format_field(value, format_spec)
-        except ValueError as exc:
-            lgr.warning(
-                "Falling back to `str` formatting for %r due to exception: %s",
-                value,
-                exc,
-            )
-            return str(value)
+        value_ = super().format_field(value, format_spec)
         if conversion:
             return self.convert_field(value_, conversion)
         return value_
+
+
+# One non-None value per Report.execution_summary key, typed as a finished run
+# with 2+ samples produces them, so a format that renders these renders a run.
+VALIDATE_SUMMARY_PLACEHOLDERS: dict[str, Any] = {
+    "exit_code": 0,
+    "command": "cmd arg",
+    "logs_prefix": ".duct/logs/prefix_",
+    "wall_clock_time": 1.0,
+    "peak_rss": 1,
+    "average_rss": 1.0,
+    "peak_vsz": 1,
+    "average_vsz": 1.0,
+    "peak_pmem": 1.0,
+    "average_pmem": 1.0,
+    "peak_pcpu": 1.0,
+    "average_pcpu": 1.0,
+    "num_samples": 2,
+    "num_reports": 1,
+    "start_time": 1.0,
+    "end_time": 2.0,
+    "working_directory": "/",
+}
+
+
+def validate_summary_format(summary_format: str) -> str:
+    """Render ``summary_format`` against placeholder values, raising if it is invalid.
+
+    Renders with colors both off and on: the custom conversions return colored
+    strings, so e.g. ``{wall_clock_time!X:.3f}`` fails only with colors.
+
+    Args:
+        summary_format (str): Template as given to --summary-format.
+
+    Returns:
+        str: ``summary_format`` unchanged, so this can serve as an argparse ``type``.
+    """
+    for enable_colors in (False, True):
+        SummaryFormatter(enable_colors=enable_colors).format(
+            summary_format, **VALIDATE_SUMMARY_PLACEHOLDERS
+        )
+    return summary_format

src/con_duct/cli.py

@@ -9,6 +9,7 @@
 from con_duct import __version__
 from con_duct._duct_main import DUCT_OUTPUT_PREFIX, EXECUTION_SUMMARY_FORMAT
 from con_duct._duct_main import execute as duct_execute
+from con_duct._formatter import validate_summary_format
 from con_duct._models import Outputs, RecordTypes, SessionMode
 from con_duct.ls import LS_FIELD_CHOICES, ls
 from con_duct.plot import CPU_MODE_PS_PCPU, CPU_MODES, matplotlib_plot
@@ -247,6 +248,32 @@ def setup_logging(args: argparse.Namespace) -> None:
         )
 
 
+# Python's {field!conv:spec} order; duct's conversions only compose as {field:spec!conv}
+CONVERSION_BEFORE_SPEC = re.compile(r"\{(\w+)!(\w):([^}]*)\}")
+
+
+def _summary_format_type(summary_format: str) -> str:
+    """argparse ``type`` for --summary-format: reject a bad template before the run.
+
+    argparse also applies this to the string default, so a bad
+    DUCT_SUMMARY_FORMAT (from the environment or a .env file) is caught too.
+    """
+    try:
+        return validate_summary_format(summary_format)
+    except Exception as e:
+        # Any failure to render known-good placeholder values is a bad template
+        msg = f"{summary_format!r}: {type(e).__name__}: {e}"
+        if m := CONVERSION_BEFORE_SPEC.search(summary_format):
+            field, conversion, spec = m.groups()
+            msg += (
+                "; duct applies conversions after the spec, "
+                f"try {{{field}:{spec}!{conversion}}}"
+            )
+        raise argparse.ArgumentTypeError(
+            f"{msg} (set via --summary-format, DUCT_SUMMARY_FORMAT, or a .env file)"
+        ) from e
+
+
 def _create_run_parser() -> argparse.ArgumentParser:
     """Create and configure the argument parser for the 'run' command."""
     parser = argparse.ArgumentParser(
@@ -276,7 +303,7 @@ def _create_run_parser() -> argparse.ArgumentParser:
     )
     parser.add_argument(
         "--summary-format",
-        type=str,
+        type=_summary_format_type,
         default=os.getenv("DUCT_SUMMARY_FORMAT", EXECUTION_SUMMARY_FORMAT),
         help="Output template to use when printing the summary following execution. "
         "Accepts custom conversion flags: "

test/duct_main/test_execution.py

@@ -12,7 +12,7 @@
 from utils import assert_files, run_duct_command
 from con_duct import _duct_main
 from con_duct._constants import SUFFIXES
-from con_duct._models import Outputs
+from con_duct._models import Outputs, RecordTypes
 
 
 def test_sample_less_than_report_interval(temp_output_dir: str) -> None:
@@ -122,6 +122,34 @@ def test_sanity_red(
     assert_expected_files(temp_output_dir)
 
 
+@pytest.mark.parametrize(
+    "record_types", [RecordTypes.ALL, RecordTypes.PROCESSES_SAMPLES]
+)
+def test_bad_summary_format_after_run_keeps_exit_code(
+    caplog: pytest.LogCaptureFixture, record_types: RecordTypes, temp_output_dir: str
+) -> None:
+    """execute() skips argparse validation; a render failure must not mask the run."""
+    caplog.set_level("INFO")
+    assert (
+        run_duct_command(
+            ["sh", "-c", "exit 3"],
+            output_prefix=temp_output_dir,
+            fail_time=0,  # keep log files regardless of exit code
+            summary_format="{nonexistent}",
+            record_types=record_types,
+        )
+        == 3
+    )
+    last = caplog.records[-1]
+    assert last.levelname == "ERROR"
+    assert "'{nonexistent}' (KeyError: 'nonexistent')" in last.message
+    has_info = record_types.has_system_summary()
+    assert ("the execution summary is in" in last.message) == has_info
+    # prepare_paths creates every log file; info.json is only filled with a system summary
+    assert (Path(temp_output_dir, SUFFIXES["info"]).stat().st_size > 0) == has_info
+    assert_files(temp_output_dir, [SUFFIXES["stdout"], SUFFIXES["stderr"]])
+
+
 def test_outputs_full(temp_output_dir: str) -> None:
     script_path = str(TEST_SCRIPT_DIR / "test_script.py")
     assert (

test/test_cli.py

@@ -272,3 +272,63 @@ def test_broken_pipe_survives_a_stdout_without_fileno() -> None:
             cli._exit_broken_pipe()
 
     assert excinfo.value.code == EXIT_BROKEN_PIPE
+
+
+@pytest.mark.parametrize(
+    "summary_format",
+    [
+        "{nonexistent}",  # unknown key
+        "{wall_clock_time:.2f",  # unmatched brace
+        "{exit_code!Q}",  # unknown conversion
+        "{0}",  # positional field
+        "{}",  # auto-numbered positional field
+        "{peak_rss:.2d}",  # spec invalid for an int
+        "{wall_clock_time:d}",  # spec invalid for a float
+        "{start_time:%Y}",  # spec invalid for a float
+        "{average_rss:,d}",  # float on real runs, so d is invalid
+        "{peak_rss.foo}",  # AttributeError
+        "{peak_rss[0]}",  # TypeError
+        "{wall_clock_time!X:.3f}",  # fails only with colors
+    ],
+)
+def test_summary_format_rejected_at_parse_time(
+    summary_format: str, capsys: pytest.CaptureFixture
+) -> None:
+    parser = _create_run_parser()
+    with pytest.raises(SystemExit) as excinfo:
+        parser.parse_args(["--summary-format", summary_format, "echo"])
+    assert excinfo.value.code == 2
+    err = capsys.readouterr().err
+    assert "argument --summary-format" in err
+    assert repr(summary_format) in err
+    assert "Traceback" not in err
+
+
+def test_summary_format_from_env_rejected(capsys: pytest.CaptureFixture) -> None:
+    with mock.patch.dict(os.environ, {"DUCT_SUMMARY_FORMAT": "{nonexistent}"}):
+        parser = _create_run_parser()
+        with pytest.raises(SystemExit) as excinfo:
+            parser.parse_args(["echo"])
+    assert excinfo.value.code == 2
+    assert "DUCT_SUMMARY_FORMAT" in capsys.readouterr().err
+
+
+def test_summary_format_valid_passes_through() -> None:
+    parser = _create_run_parser()
+    summary_format = "{exit_code!E} {peak_rss:.2f} {wall_clock_time!T}"
+    args = parser.parse_args(["--summary-format", summary_format, "echo"])
+    assert args.summary_format == summary_format
+
+
+def test_summary_format_conversion_order_hint(capsys: pytest.CaptureFixture) -> None:
+    parser = _create_run_parser()
+    with pytest.raises(SystemExit):
+        parser.parse_args(["--summary-format", "{wall_clock_time!X:.3f}", "echo"])
+    assert "try {wall_clock_time:.3f!X}" in capsys.readouterr().err
+
+
+def test_summary_format_conversion_order_when_valid() -> None:
+    """Python's conversion-first order is fine when the spec suits a str."""
+    parser = _create_run_parser()
+    args = parser.parse_args(["--summary-format", "{peak_rss!S:>10}", "echo"])
+    assert args.summary_format == "{peak_rss!S:>10}"

test/test_formatter.py

@@ -1,7 +1,14 @@
+from collections import Counter
 import os
 from unittest import mock
 import pytest
-from con_duct._formatter import SummaryFormatter
+from con_duct._duct_main import EXECUTION_SUMMARY_FORMAT
+from con_duct._formatter import (
+    VALIDATE_SUMMARY_PLACEHOLDERS,
+    SummaryFormatter,
+    validate_summary_format,
+)
+from con_duct._models import ProcessStats, Sample
 from con_duct._tracker import Report
 
 GREEN_START = SummaryFormatter.COLOR_SEQ % SummaryFormatter.GREEN
@@ -101,6 +108,14 @@ def test_summary_formatter_none_replacement() -> None:
     assert out == "test -"
 
 
+def test_summary_formatter_spec_mismatch_raises() -> None:
+    formatter = SummaryFormatter()
+    with pytest.raises(ValueError):
+        formatter.format("{value:.2d}", value=123456)
+    # None never reaches the spec, so it still renders
+    assert formatter.format("{value:.2d}", value=None) == "-"
+
+
 def test_summary_formatter_S_e2e() -> None:
     formatter = SummaryFormatter()
     one_arg = {"big_num": 100000}
@@ -346,7 +361,8 @@ def test_execution_summary_formatted_wall_clock_time_nowvalid(
     )
     assert f"Rendering: {GREEN}nan{STOP}" == report.execution_summary_formatted
 
-    # or if we really provide bad formatting, e.g. the opposite order of conversion and formatting
+    # Python's {field!conv:spec} order: the colored conversion yields a str,
+    # which the float spec cannot format
     report = Report(
         "_cmd",
         [],
@@ -356,7 +372,11 @@ def test_execution_summary_formatted_wall_clock_time_nowvalid(
         clobber=False,
         colors=colors,
     )
-    assert f"Rendering: {GREEN}nan{STOP}" == report.execution_summary_formatted
+    if colors:
+        with pytest.raises(ValueError):
+            _ = report.execution_summary_formatted
+    else:
+        assert "Rendering: nan" == report.execution_summary_formatted
 
 
 def test_summary_formatter_P_e2e() -> None:
@@ -494,3 +514,45 @@ def test_summary_formatter_T_duration_formats(duration: float, expected: str) ->
     formatter = SummaryFormatter()
     result = formatter.format("{duration!T}", duration=duration)
     assert result == expected
+
+
+def test_validate_summary_format_accepts_default() -> None:
+    assert validate_summary_format(EXECUTION_SUMMARY_FORMAT) == EXECUTION_SUMMARY_FORMAT
+
+
+@mock.patch("con_duct._duct_main.LogPaths")
+def test_validate_summary_placeholders_match_a_real_summary(
+    mock_log_paths: mock.MagicMock,
+) -> None:
+    """Placeholders must have the keys and types of a finished run's summary.
+
+    Otherwise validation is looser or stricter than the post-run render,
+    e.g. an int placeholder where runs produce floats would accept ``{x:d}``.
+    """
+    mock_log_paths.prefix = "mock_prefix"
+    report = Report("_cmd", ["arg"], mock_log_paths, "", os.getcwd())
+    report.process = mock.MagicMock(returncode=0)
+    report.start_time = 1727221840.0
+    report.end_time = report.start_time + 2.5
+    for pcpu, rss in ((10.0, 1000), (30.0, 3000)):
+        sample = Sample()
+        sample.add_pid(
+            1,
+            ProcessStats(
+                pcpu=pcpu,
+                pmem=1.5,
+                rss=rss,
+                vsz=rss * 2,
+                timestamp="2024-06-11T10:09:37-04:00",
+                etime="00:01",
+                stat=Counter(["S"]),
+                cmd="_cmd arg",
+            ),
+        )
+        report.update_from_sample(sample)
+
+    summary = report.execution_summary
+    assert summary.keys() == VALIDATE_SUMMARY_PLACEHOLDERS.keys()
+    assert {k: type(v) for k, v in summary.items()} == {
+        k: type(v) for k, v in VALIDATE_SUMMARY_PLACEHOLDERS.items()
+    }