feat: group preflight warnings into one tree block per check (#848)

* test: pin aggregated preflight warning output

A run with N unguaranteed fields emits N near-identical multi-line
warnings, each repeating the full remedy paragraph. Pin the intended
contract: one warning record per check group, fields grouped per
action on one line, remedy stated once, counts in the header.

Signed-off-by: Muizz Lateef <lateefmuizz@gmail.com>

* test: capture preflight warnings via propagation-forcing helper

The grouping tests must force agent_actions logger propagation back on
for the capture window — the logging bridge disables it — matching the
established wiring-test idiom.

Signed-off-by: Muizz Lateef <lateefmuizz@gmail.com>

* feat: group preflight warnings into one tree block per check

A run against a real project emitted 30+ near-identical multi-line
warnings — one full paragraph per unguaranteed field, each repeating
the same remedy. The warning wall buried both the signal and the two
non-dag-fit findings mixed into it.

The dag-fit scanner now returns missing fields grouped per consumer
action, and the preflight service renders each warn-check's findings
as a single tree-formatted warning: a header with field/action counts,
one line per action, the remedy stated once. The prompt-contract,
tool-passthrough, and resolution warnings route through the same
renderer, so every preflight group reads as one scannable block.

Signed-off-by: Muizz Lateef <lateefmuizz@gmail.com>

---------

Signed-off-by: Muizz Lateef <lateefmuizz@gmail.com>

Author: Muizzkolapo

.changes/unreleased/Enhancement or New Feature-20260808-013000.yaml

@@ -0,0 +1,3 @@
+kind: Enhancement or New Feature
+body: "Preflight warnings now print as one grouped, tree-formatted block per check (dag-fit, prompt-contract, tool-passthrough, resolution) with counts in the header, fields grouped per action, and the remedy stated once — instead of one full paragraph per field. A run that previously emitted 30+ near-identical warning paragraphs now shows a dozen scannable lines."
+time: 2026-08-08T01:30:00Z

agent_actions/services/preflight_service.py

@@ -19,6 +19,7 @@
 from agent_actions.utils.constants import NON_PROMPT_ACTION_KINDS
 from agent_actions.utils.udf_management.registry import get_udf_metadata
 from agent_actions.validation.dag_schema_fit_validator import (
+    DAG_FIT_REMEDY,
     find_dag_schema_compatibility_gaps,
 )
 from agent_actions.validation.dep_observe_validator import find_missing_observe_deps
@@ -113,14 +114,17 @@ def validate(self) -> None:
         ).resolve_all()
         resolution_result.raise_if_invalid()
 
-        for warning in resolution_result.warnings:
-            logger.warning("Pre-flight: %s", warning.message)
+        self._warn_findings(
+            "resolution", [warning.message for warning in resolution_result.warnings]
+        )
 
         # 6. Cross-check prompt refs against each producer's required set
-        for finding in find_unguarded_required_refs(
-            self._collect_prompts(), self._collect_producing_schemas()
-        ):
-            logger.warning("Pre-flight: %s", finding)
+        self._warn_findings(
+            "prompt-contract",
+            find_unguarded_required_refs(
+                self._collect_prompts(), self._collect_producing_schemas()
+            ),
+        )
 
         # 7. Cross-check kind:tool passthrough UDFs against strict output schemas
         self._warn_tool_passthrough_risks()
@@ -198,8 +202,10 @@ def _collect_tool_passthrough_inputs(self) -> dict[str, dict[str, Any]]:
 
     def _warn_tool_passthrough_risks(self) -> None:
         """Warn when a kind:tool UDF passes upstream dicts through a strict schema."""
-        for finding in find_passthrough_schema_risks(self._collect_tool_passthrough_inputs()):
-            logger.warning("Pre-flight: %s", finding)
+        self._warn_findings(
+            "tool-passthrough",
+            find_passthrough_schema_risks(self._collect_tool_passthrough_inputs()),
+        )
 
     def _collect_tool_required_field_inputs(self) -> dict[str, dict[str, Any]]:
         """UDF source + compiled required list per kind:tool action.
@@ -247,5 +253,43 @@ def _check_tool_conditional_required_field_risks(self) -> None:
 
     def _warn_dag_schema_compatibility_gaps(self) -> None:
         """Warn when a tool consumer's required output field is neither guaranteed by an upstream producer nor declared as synthesized via `defaults:`."""
-        for finding in find_dag_schema_compatibility_gaps(self.action_configs):
-            logger.warning("Pre-flight: %s", finding)
+        gaps = find_dag_schema_compatibility_gaps(self.action_configs)
+        if not gaps:
+            return
+        total = sum(len(fields) for fields in gaps.values())
+        self._warn_findings(
+            "dag-fit",
+            [f"{action}: {', '.join(fields)}" for action, fields in gaps.items()],
+            header=(
+                f"dag-fit — {total} required field(s) with no upstream guarantee "
+                f"across {len(gaps)} action(s)"
+            ),
+            remedy=DAG_FIT_REMEDY,
+        )
+
+    @staticmethod
+    def _warn_findings(
+        label: str,
+        findings: list[str],
+        header: str | None = None,
+        remedy: str | None = None,
+    ) -> None:
+        """Emit one grouped warning for a check's findings instead of one per finding.
+
+        Renders a tree so a run's warning wall reads as a few scannable blocks:
+
+            Pre-flight: dag-fit — 5 required field(s) ... across 2 action(s)
+              ├─ flatten: category, key, steps
+              ├─ assemble: id, items
+              └─ Fix: mark the field optional in the consumer schema, ...
+        """
+        if not findings:
+            return
+        items = list(findings)
+        if remedy:
+            items.append(f"Fix: {remedy}")
+        body = "\n".join(
+            f"  {'└─' if i == len(items) - 1 else '├─'} {item}" for i, item in enumerate(items)
+        )
+        head = header or f"{label} — {len(findings)} warning(s)"
+        logger.warning("Pre-flight: %s\n%s", head, body)

agent_actions/validation/_MANIFEST.md

@@ -24,7 +24,7 @@ decoders to schema validators and preflight checks.
 | `base_validator.py` | Module | `BaseValidator` base class with helper assertions for validators. | `validation` |
 | `bus_namespace_validator.py` | Module | `find_unknown_bus_namespaces`: AST scan flagging tool-UDF reads of `data.get("X")` / `data["X"]` where X is not a runtime bus namespace (action name or framework key), catching silent namespace typos at `validate-udfs`. | `validation` |
 | `config_validator.py` | Module | Central config parser/validator used across startup flows. | `configuration`, `validation` |
-| `dag_schema_fit_validator.py` | Module | `find_dag_schema_compatibility_gaps`: per producer/consumer edge in the workflow DAG, warns when a tool consumer's required output field is neither guaranteed at any position in an upstream producer's compiled schema nor declared as synthesized via `defaults:` on the action. Symmetric two-level descent on both sides (root, `field.`, `field[].`) — no UDF source reading. Wired at `PreflightService._warn_dag_schema_compatibility_gaps` (spec 592 Phase 2, warn-only). | `validation` |
+| `dag_schema_fit_validator.py` | Module | `find_dag_schema_compatibility_gaps`: per producer/consumer edge in the workflow DAG, reports `{consumer: [missing fields]}` where a tool consumer's required output field is neither guaranteed at any position in an upstream producer's compiled schema nor declared as synthesized via `defaults:` on the action; `DAG_FIT_REMEDY` carries the shared fix text. Symmetric two-level descent on both sides (root, `field.`, `field[].`) — no UDF source reading. Wired at `PreflightService._warn_dag_schema_compatibility_gaps` (spec 592 Phase 2, warn-only). | `validation` |
 | `dep_observe_validator.py` | Module | `find_missing_observe_deps`: preflight mirror of the fatal runtime check that every declared dependency has an observe/passthrough field reference. | `validation` |
 | `path_validator.py` | Module | Path validation utilities conforming to BaseValidator interface. | `validation` |
 | `prompt_ast.py` | Module | Jinja2 AST parser for extracting template variables. | `prompt_generation` |

agent_actions/validation/dag_schema_fit_validator.py

@@ -82,17 +82,26 @@ def _upstream_edges(
     return resolved
 
 
+DAG_FIT_REMEDY = (
+    "mark the field optional in the consumer schema, mark it required at the "
+    "producing position upstream, or declare it with `defaults:` on the consuming "
+    "action — records missing a required field are rejected at runtime."
+)
+
+
 def find_dag_schema_compatibility_gaps(
     action_configs: dict[str, dict[str, Any]],
-) -> list[str]:
-    """One finding per required output field on a tool consumer that is neither
-    guaranteed by an upstream producer nor declared as synthesized via
-    `defaults:` on the action.
+) -> dict[str, list[str]]:
+    """Per tool consumer, the required output fields that are neither guaranteed
+    by an upstream producer nor declared as synthesized via `defaults:`.
+
+    Returns ``{consumer_action: [sorted missing fields]}`` for the caller to
+    render as one grouped warning (remedy: ``DAG_FIT_REMEDY``).
 
     Warn-only for Phase 2 of spec 592. Phase 4 flips this fatal and removes
     the sibling `find_conditional_required_field_risks` scanner it subsumes.
     """
-    findings: list[str] = []
+    gaps: dict[str, list[str]] = {}
     for consumer_name, consumer in action_configs.items():
         if consumer.get("kind") != "tool":
             continue
@@ -118,15 +127,8 @@ def find_dag_schema_compatibility_gaps(
             if isinstance(producer_schema, dict):
                 guaranteed |= _required_fields_two_level(producer_schema)
 
-        for field in sorted(implicit_inputs - guaranteed):
-            findings.append(
-                f"dag-fit: {consumer_name} output requires '{field}' but no upstream "
-                f"producer guarantees it (declared in properties without being listed "
-                f"as required), and no `defaults:` entry declares it synthesized on "
-                f"this action. Runtime schema validation will reject records where the "
-                f"UDF does not emit it. Fix one of: mark '{field}' optional in the "
-                f"consumer schema; mark it required at the producing position upstream; "
-                f"or add `defaults: {{{field}: <value>}}` on this action."
-            )
-
-    return findings
+        missing = sorted(implicit_inputs - guaranteed)
+        if missing:
+            gaps[consumer_name] = missing
+
+    return gaps

tests/services/test_preflight_warning_grouping.py

@@ -0,0 +1,92 @@
+"""Preflight warnings must aggregate per check, not print one paragraph per field.
+
+A run with N unguaranteed fields used to emit N near-identical multi-line
+warnings, each repeating the full remedy. The contract here: one warning
+record per check group, fields grouped by action, remedy stated once.
+"""
+
+import logging
+
+from agent_actions.services.preflight_service import PreflightService
+
+_LOGGER = "agent_actions.services.preflight_service"
+
+
+def _capture_dagfit(configs, caplog):
+    """Run the dag-fit warn path with root propagation forced on for capture."""
+    aa_logger = logging.getLogger("agent_actions")
+    original = aa_logger.propagate
+    aa_logger.propagate = True
+    try:
+        with caplog.at_level(logging.WARNING, logger=_LOGGER):
+            _service(configs)._warn_dag_schema_compatibility_gaps()
+    finally:
+        aa_logger.propagate = original
+    return [r for r in caplog.records if "dag-fit" in r.getMessage()]
+
+
+def _service(action_configs):
+    return PreflightService(
+        agent_name="wf",
+        action_configs=action_configs,
+        project_root=None,
+        workflow_config_path="wf.yml",
+        verify_keys=False,
+    )
+
+
+def _tool(required, properties=None):
+    props = properties or {f: {"type": "string"} for f in required}
+    return {
+        "kind": "tool",
+        "dependencies": [],
+        "json_output_schema": {
+            "type": "object",
+            "properties": props,
+            "required": list(required),
+        },
+    }
+
+
+class TestDagFitGrouping:
+    def test_one_warning_record_for_the_whole_group(self, caplog):
+        configs = {
+            "flatten": _tool(["category", "key"]),
+            "assemble": _tool(["id", "items"]),
+        }
+        dagfit = _capture_dagfit(configs, caplog)
+        assert len(dagfit) == 1, (
+            f"expected ONE aggregated dag-fit warning, got {len(dagfit)} records"
+        )
+
+    def test_fields_grouped_per_action_on_one_line(self, caplog):
+        configs = {
+            "flatten": _tool(["category", "key", "steps"]),
+            "assemble": _tool(["id", "items"]),
+        }
+        msg = _capture_dagfit(configs, caplog)[0].getMessage()
+        assert "flatten: category, key, steps" in msg, msg
+        assert "assemble: id, items" in msg, msg
+
+    def test_remedy_stated_once_not_per_field(self, caplog):
+        configs = {
+            "flatten": _tool(["category", "key", "steps"]),
+            "assemble": _tool(["id", "items"]),
+        }
+        combined = "\n".join(r.getMessage() for r in _capture_dagfit(configs, caplog))
+        assert combined.count("optional in the consumer schema") == 1, combined
+
+    def test_header_carries_counts(self, caplog):
+        configs = {
+            "flatten": _tool(["category", "key", "steps"]),
+            "assemble": _tool(["id", "items"]),
+        }
+        msg = _capture_dagfit(configs, caplog)[0].getMessage()
+        first_line = msg.splitlines()[0]
+        assert "5" in first_line and "2" in first_line, (
+            f"header should carry field and action counts: {first_line!r}"
+        )
+
+    def test_quiet_when_no_gaps(self, caplog):
+        configs = {"ok": _tool([])}
+        assert _capture_dagfit(configs, caplog) == []