CHANGELOG.md

@@ -27,6 +27,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   26-character identifier. The full ULID is now displayed in all output formats (Rich, plain,
   JSON, YAML, table).
 
+- **Suppress passing BDD scenario output in `unit_tests` by default** (#10987): Implemented
+  `PassSuppressFormatter`, a custom Behave formatter (in `scripts/behave_pass_suppress_formatter.py`)
+  that buffers all per-scenario output and only flushes it to stdout when a scenario fails or
+  errors. An all-passing `nox -s unit_tests` run now produces ≤ 10 lines (the summary block
+  only), eliminating ~100,000 lines of noise that previously made CI logs unreadable. The
+  formatter is embedded in the `behave-parallel` in-process runner; coverage mode
+  (`BEHAVE_PARALLEL_COVERAGE=1`) is unaffected.
+
 ### Security
 
 - **aiohttp upgraded to >=3.13.4 to remediate CVE-2026-34513 and CVE-2026-34515** (#1549, #1544):

features/acms/index_data_model_and_traversal.feature

@@ -9,6 +9,7 @@ Feature: ACMS Index Data Model and File Traversal Engine
 
   Scenario: Create an index entry with file metadata
     When I create an index entry with:
+      | key | value |
       | path | /project/src/main.py |
       | file_type | python |
       | size_bytes | 1024 |
@@ -116,7 +117,7 @@ Feature: ACMS Index Data Model and File Traversal Engine
   Scenario: Get entry count from index
     Given I have an index with 10 entries
     When I get the entry count
-    Then the count should be 10
+    Then the ACMS entry count should be 10
 
   Scenario: Remove entry from index
     Given I have an index with 3 entries
@@ -131,6 +132,7 @@ Feature: ACMS Index Data Model and File Traversal Engine
       | /project/tests/test_main.py | python | test | cold |
       | /project/docs/readme.md | markdown | docs | cold |
     When I query the index with filters:
+      | filter | value |
       | path_pattern | src |
       | file_type | python |
       | tier | hot |

features/behave_parallel_log_filtering.feature

@@ -99,3 +99,23 @@ Feature: Behave-parallel conditional log replay
     Given behave_parallel worker results with one crashed chunk and one passing chunk
     When behave_parallel I aggregate the worker results
     Then behave_parallel the aggregated summary should have failures
+
+  # ---- PassSuppressFormatter: per-scenario output suppression ----
+
+  Scenario: PassSuppressFormatter suppresses output for a passing scenario
+    Given behave_parallel a PassSuppressFormatter backed by a captured stream
+    When behave_parallel I simulate a passing scenario through the formatter
+    Then behave_parallel the formatter real stream output should be empty
+
+  Scenario: PassSuppressFormatter emits full output for a failing scenario
+    Given behave_parallel a PassSuppressFormatter backed by a captured stream
+    When behave_parallel I simulate a failing scenario through the formatter
+    Then behave_parallel the formatter real stream output should contain "Failing scenario"
+    And behave_parallel the formatter real stream output should contain "I fail"
+    And behave_parallel the formatter real stream output should contain "AssertionError"
+
+  Scenario: PassSuppressFormatter only shows failed scenarios in a mixed run
+    Given behave_parallel a PassSuppressFormatter backed by a captured stream
+    When behave_parallel I simulate one passing then one failing scenario through the formatter
+    Then behave_parallel the formatter real stream output should not contain "Passing scenario"
+    And behave_parallel the formatter real stream output should contain "Failing scenario"

features/steps/acms_index_data_model_traversal_steps.py

@@ -361,7 +361,7 @@ def step_get_entry_count(context):
     context.entry_count = context.index.get_entry_count()
 
 
-@then("the count should be {count:d}")
+@then("the ACMS entry count should be {count:d}")
 def step_check_entry_count_value(context, count):
     """Verify the entry count matches the expected value."""
     assert context.entry_count == count

features/steps/behave_parallel_log_filtering_steps.py

@@ -1,7 +1,8 @@
 """Step definitions for behave_parallel_log_filtering.feature.
 
 Tests the conditional log replay and worker exception handling in
-``scripts/run_behave_parallel.py``.
+``scripts/run_behave_parallel.py``, as well as the PassSuppressFormatter
+per-scenario output buffering behaviour.
 """
 
 from __future__ import annotations
@@ -12,8 +13,11 @@ import sys
 from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
 from types import ModuleType
+from typing import Any
 
 from behave import given, then, when  # type: ignore[import-untyped]
+from behave.configuration import Configuration  # type: ignore[import-untyped]
+from behave.formatter.base import StreamOpener  # type: ignore[import-untyped]
 from behave.runner import Context  # type: ignore[import-untyped]
 
 
@@ -52,6 +56,7 @@ _empty_summary = _runner_mod._empty_summary
 _worker_run_features = _runner_mod._worker_run_features
 _aggregate_worker_results = _runner_mod._aggregate_worker_results
 _has_failures = _runner_mod._has_failures
+PassSuppressFormatter = _runner_mod.PassSuppressFormatter
 
 
 def _passing_summary(features: int = 1, scenarios: int = 1) -> Summary:
@@ -357,3 +362,132 @@ def step_worker_summary_features_errors_1(context: Context) -> None:
         f"Expected features.errors=1 so _has_failures() detects partial crash, "
         f"got: {errors}"
     )
+
+
+# ---- PassSuppressFormatter helpers ----
+
+
+class _MockStatus:
+    """Minimal status object mirroring behave's Status enum interface."""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+
+class _MockScenario:
+    """Minimal scenario object for PassSuppressFormatter unit tests."""
+
+    def __init__(self, name: str, status_name: str) -> None:
+        self.keyword = "Scenario"
+        self.name = name
+        self.status = _MockStatus(status_name)
+
+
+class _MockStep:
+    """Minimal step object for PassSuppressFormatter unit tests."""
+
+    def __init__(
+        self,
+        keyword: str,
+        name: str,
+        status_name: str,
+        error_message: str | None,
+    ) -> None:
+        self.keyword = keyword
+        self.name = name
+        self.status = _MockStatus(status_name)
+        self.error_message = error_message
+
+
+def _make_pass_suppress_formatter() -> tuple[Any, io.StringIO]:
+    """Create a PassSuppressFormatter writing to a fresh StringIO buffer.
+
+    Returns ``(formatter, buffer)`` so callers can inspect what was written
+    to the formatter's real output stream after simulating scenario events.
+    """
+    buf: io.StringIO = io.StringIO()
+    opener = StreamOpener(stream=buf)
+    config = Configuration(command_args=["-q"])
+    formatter: Any = PassSuppressFormatter(opener, config)
+    return formatter, buf
+
+
+# ---- PassSuppressFormatter: Given ----
+
+
+@given("behave_parallel a PassSuppressFormatter backed by a captured stream")
+def step_create_pass_suppress_formatter(context: Context) -> None:
+    formatter, buf = _make_pass_suppress_formatter()
+    context.bp_formatter = formatter
+    context.bp_formatter_stream = buf
+
+
+# ---- PassSuppressFormatter: When ----
+
+
+@when("behave_parallel I simulate a passing scenario through the formatter")
+def step_simulate_passing_scenario(context: Context) -> None:
+    formatter: Any = context.bp_formatter
+    scenario = _MockScenario("Passing scenario", "passed")
+    step = _MockStep("Given", "I pass", "passed", None)
+    formatter.scenario(scenario)
+    formatter.result(step)
+    formatter.eof()
+
+
+@when("behave_parallel I simulate a failing scenario through the formatter")
+def step_simulate_failing_scenario(context: Context) -> None:
+    formatter: Any = context.bp_formatter
+    scenario = _MockScenario("Failing scenario", "failed")
+    step = _MockStep(
+        "When",
+        "I fail",
+        "failed",
+        "AssertionError: expected True got False",
+    )
+    formatter.scenario(scenario)
+    formatter.result(step)
+    formatter.eof()
+
+
+@when(
+    "behave_parallel I simulate one passing then one failing scenario"
+    " through the formatter"
+)
+def step_simulate_mixed_scenarios(context: Context) -> None:
+    formatter: Any = context.bp_formatter
+    # First: a passing scenario.
+    passing_scenario = _MockScenario("Passing scenario", "passed")
+    step1 = _MockStep("Given", "I pass", "passed", None)
+    formatter.scenario(passing_scenario)
+    formatter.result(step1)
+    # Second: a failing scenario.  Calling formatter.scenario() here finalises
+    # the previous (passing) scenario, which should be discarded.
+    failing_scenario = _MockScenario("Failing scenario", "failed")
+    step2 = _MockStep("When", "I fail", "failed", "AssertionError: it broke")
+    formatter.scenario(failing_scenario)
+    formatter.result(step2)
+    formatter.eof()
+
+
+# ---- PassSuppressFormatter: Then ----
+
+
+@then("behave_parallel the formatter real stream output should be empty")
+def step_formatter_output_empty(context: Context) -> None:
+    output: str = context.bp_formatter_stream.getvalue()
+    assert output == "", f"Expected empty output, got: {output!r}"
+
+
+@then('behave_parallel the formatter real stream output should contain "{text}"')
+def step_formatter_output_contains(context: Context, text: str) -> None:
+    output: str = context.bp_formatter_stream.getvalue()
+    assert text in output, f"Expected {text!r} in formatter output, got: {output!r}"
+
+
+@then('behave_parallel the formatter real stream output should not contain "{text}"')
+def step_formatter_output_not_contains(context: Context, text: str) -> None:
+    output: str = context.bp_formatter_stream.getvalue()
+    assert text not in output, (
+        f"Expected {text!r} NOT in formatter output, got: {output!r}"
+    )

features/steps/cli_init_yes_flag_steps.py

@@ -27,7 +27,8 @@ def _restore_cwd(context):
         os.environ.pop("CLEVERAGENTS_HOME", None)
     else:
         os.environ["CLEVERAGENTS_HOME"] = context._init_original_home
-    shutil.rmtree(context.temp_dir, ignore_errors=True)
+    if getattr(context, "temp_dir", None) is not None:
+        shutil.rmtree(context.temp_dir, ignore_errors=True)
 
 
 def _create_init_mocks(context):

features/steps/pr_compliance_checklist_steps.py

@@ -5,7 +5,7 @@ from typing import Any
 
 from behave import given, then, when
 
-PROJECT_ROOT = Path(__file__).resolve().parents[3]
+PROJECT_ROOT = Path(__file__).resolve().parents[2]
 AGENT_DEF_PATH = PROJECT_ROOT / ".opencode" / "agents" / "implementation-supervisor.md"
 
 

noxfile.py

@@ -111,6 +111,13 @@ def _install_behave_parallel(session: nox.Session) -> None:
     runner_script = Path(__file__).parent / "scripts" / "run_behave_parallel.py"
     (pkg_dir / "cli.py").write_text(runner_script.read_text(encoding="utf-8"))
 
+    formatter_script = (
+        Path(__file__).parent / "scripts" / "behave_pass_suppress_formatter.py"
+    )
+    (pkg_dir / "behave_pass_suppress_formatter.py").write_text(
+        formatter_script.read_text(encoding="utf-8")
+    )
+
     setup_path = source_dir / "setup.py"
     setup_path.write_text(
         "from setuptools import find_packages, setup\n"

robot/helper_behave_parallel_log_filtering.py

@@ -26,6 +26,15 @@ def _load_runner_module() -> ModuleType:
     first or modify the path to be anchored relative to ``__file__``.
     """
     script_path = Path("scripts") / "run_behave_parallel.py"
+    # Ensure the scripts/ directory is on sys.path so that
+    # ``run_behave_parallel.py`` can ``from behave_pass_suppress_formatter
+    # import PassSuppressFormatter``.  This is necessary when the helper is
+    # invoked outside of a nox session (e.g. integration tests), where the
+    # behave_parallel package created by noxfile.py for unit_tests is not
+    # available.
+    scripts_dir = str(script_path.parent.resolve())
+    if scripts_dir not in sys.path:
+        sys.path.insert(0, scripts_dir)
     spec = importlib.util.spec_from_file_location(
         "run_behave_parallel", str(script_path)
     )

scripts/behave_pass_suppress_formatter.py

@@ -0,0 +1,125 @@
+"""PassSuppressFormatter — a custom Behave formatter that suppresses output
+for passing and skipped scenarios, reducing CI log noise from ~100,000 lines
+to ≤ 10 lines for an all-passing suite.
+
+This module is embedded in the same directory as ``run_behave_parallel.py``
+so that ``_install_behave_parallel()`` in ``noxfile.py`` can copy both files
+into the temporary ``behave_parallel`` package.
+"""
+
+from __future__ import annotations
+
+import io
+from typing import Any
+
+from behave.formatter.base import (
+    Formatter as _BehaveFormatter,  # type: ignore[import-untyped]
+)
+
+# Sentinel set of statuses whose output should be suppressed.
+# Every status not in this set (e.g. "failed", "undefined") causes the
+# buffered scenario output to be flushed to the real output stream.
+_SUPPRESS_STATUSES: frozenset[str] = frozenset({"passed", "skipped"})
+
+
+class PassSuppressFormatter(_BehaveFormatter):
+    """Behave formatter that suppresses output for passing scenarios.
+
+    All per-scenario output is buffered in a :class:`io.StringIO`.  When a
+    scenario ends (signalled by the next :meth:`scenario` call or by
+    :meth:`eof`):
+
+    * **Failed / errored scenario** — the buffer is flushed to the real
+      output stream so developers and CI tools see the full step-level
+      details.
+    * **Passed / skipped scenario** — the buffer is silently discarded.
+
+    The result for an all-passing suite is zero formatter output, leaving
+    only the ``_print_overall_summary`` block emitted by the runner as
+    visible output (~5-10 lines regardless of suite size).
+
+    Coverage mode (``BEHAVE_PARALLEL_COVERAGE=1``) bypasses this formatter
+    entirely via ``_make_runner()``, which falls back to behave's default
+    format so that slipcover can instrument a single sequential process
+    without output interference.
+    """
+
+    name: str = "pass_suppress"
+    description: str = "Suppress passing scenario output; show failures fully"
+
+    def __init__(self, stream_opener: Any, config: Any) -> None:
+        super().__init__(stream_opener, config)
+        # Ensure the real output stream is open (opens it if stream_opener
+        # was constructed with only a filename, not a pre-opened stream).
+        self.stream = self.open()
+
+        # Per-scenario buffering state.
+        self._current_scenario: Any = None
+        self._scenario_buf: io.StringIO = io.StringIO()
+
+    # ------------------------------------------------------------------
+    # Internal helpers
+    # ------------------------------------------------------------------
+
+    def _finalize_previous_scenario(self) -> None:
+        """Flush the scenario buffer to the real stream if scenario failed.
+
+        Called at the start of each new scenario and at end-of-feature so
+        the previous scenario's buffered output is either committed or
+        discarded based on the scenario's final status.
+        """
+        if self._current_scenario is None:
+            return
+        status: Any = getattr(self._current_scenario, "status", None)
+        status_name: str = (
+            getattr(status, "name", str(status)) if status is not None else ""
+        )
+        if status_name not in _SUPPRESS_STATUSES:
+            output = self._scenario_buf.getvalue()
+            if output:
+                self.stream.write(output)
+                self.stream.flush()
+        # Reset the buffer regardless — the next scenario starts fresh.
+        self._scenario_buf = io.StringIO()
+
+    # ------------------------------------------------------------------
+    # Formatter interface (behave lifecycle hooks)
+    # ------------------------------------------------------------------
+
+    def uri(self, uri: str) -> None:
+        """Called before each feature file; no output needed."""
+
+    def feature(self, feature: Any) -> None:
+        """Called at feature start; no output needed."""
+
+    def background(self, background: Any) -> None:
+        """Called when a feature background is declared; no output needed."""
+
+    def scenario(self, scenario: Any) -> None:
+        """Start buffering output for *scenario*, finalising the previous one."""
+        self._finalize_previous_scenario()
+        self._current_scenario = scenario
+        # Write the scenario header into the new buffer.
+        self._scenario_buf.write(f"\n  {scenario.keyword}: {scenario.name}\n")
+
+    def step(self, step: Any) -> None:
+        """Step announced before execution; no output needed at this point."""
+
+    def match(self, match: Any) -> None:
+        """Step matched; no output needed."""
+
+    def result(self, step: Any) -> None:
+        """Record a step result in the per-scenario buffer."""
+        status: Any = getattr(step, "status", None)
+        status_name: str = (
+            getattr(status, "name", "unknown") if status is not None else "unknown"
+        )
+        self._scenario_buf.write(f"    {step.keyword} {step.name} ... {status_name}\n")
+        error_msg: str | None = getattr(step, "error_message", None)
+        if error_msg:
+            self._scenario_buf.write(f"{error_msg}\n")
+
+    def eof(self) -> None:
+        """End of feature file: finalise the current (last) scenario."""
+        self._finalize_previous_scenario()
+        self._current_scenario = None

scripts/run_behave_parallel.py

@@ -1,20 +1,16 @@
 """In-process parallel behave runner.
 
-Replaces the old subprocess-per-feature model with direct use of
-behave's ``Runner`` API.  Step definitions and environment hooks are
-loaded once per process; feature files are parsed and executed without
-Python interpreter startup overhead.
+Uses behave's ``Runner`` API directly: step definitions and environment
+hooks are loaded once per process; feature files are parsed and executed
+without Python interpreter startup overhead.
 
 Parallelism modes
 -----------------
 * **Sequential** (``--processes 1`` or ``BEHAVE_PARALLEL_COVERAGE=1``):
   All features run in a single ``Runner.run()`` call.
 * **Parallel** (``--processes N``, N > 1, no coverage):
-  Features are split into *N* equal-size chunks.  A
-  ``multiprocessing.Pool`` with the ``fork`` start method dispatches
-  each chunk to a worker that creates its own ``Runner`` (hooks and
-  step definitions are re-loaded cheaply because all heavy modules are
-  already in memory from the parent).
+  Features split into *N* equal-size chunks dispatched via
+  ``multiprocessing.Pool`` with the ``fork`` start method.
 """
 
 from __future__ import annotations
@@ -31,6 +27,13 @@ from contextlib import redirect_stderr, redirect_stdout, suppress
 from pathlib import Path
 from typing import Any
 
+try:
+    from behave_pass_suppress_formatter import PassSuppressFormatter
+except ImportError:
+    from behave_parallel.behave_pass_suppress_formatter import (  # type: ignore[import-untyped,unused-ignore]
+        PassSuppressFormatter,
+    )
+
 DEFAULT_FEATURE_ROOT = "features/"
 
 
@@ -79,6 +82,7 @@ def _is_btrfs_or_overlayfs() -> bool:
 
 # Type alias for summary dictionaries
 Summary = dict[str, Any]
+WorkerResult = tuple[bool, str, str, Summary]
 
 
 # ---------------------------------------------------------------------------
@@ -174,10 +178,7 @@ def _format_duration(seconds: float) -> str:
     return f"{remainder:.3f}s"
 
 
-def _print_overall_summary(
-    total: Summary,
-    wall_seconds: float | None = None,
-) -> None:
+def _print_overall_summary(total: Summary, wall_seconds: float | None = None) -> None:
     print("\nOverall summary:")
     for bucket in ("features", "scenarios", "steps"):
         b = total[bucket]
@@ -272,35 +273,50 @@ def _make_runner(feature_paths: list[str], behave_args: list[str]) -> Any:
     Mirrors the format-defaulting logic from ``behave.__main__.run_behave``
     so that ``-q`` and bare invocations get a sensible formatter instead
     of crashing on ``config.format is None``.
+
+    When no explicit ``--format``/``-f`` flag is present in *behave_args* and
+    ``BEHAVE_PARALLEL_COVERAGE`` is **not** set, the runner uses
+    :class:`PassSuppressFormatter` so that passing scenarios produce no
+    output.  Coverage mode falls back to ``config.default_format`` (normally
+    ``pretty``) so that slipcover can instrument a single sequential process
+    without output interference.
     """
     from behave.configuration import Configuration
+    from behave.formatter._registry import register_as
     from behave.runner import Runner
     from behave.runner_util import reset_runtime
 
     reset_runtime()
+    # Register our custom formatter so behave can resolve it by name when
+    # make_formatters() is called during Runner initialisation.
+    register_as(PassSuppressFormatter.name, PassSuppressFormatter)
+
     args = list(behave_args) + [str(p) for p in feature_paths]
     config = Configuration(command_args=args)
     if not config.format:
-        config.format = [config.default_format]
+        # No explicit format was requested by the caller.  Use pass_suppress
+        # unless coverage instrumentation is active (BEHAVE_PARALLEL_COVERAGE=1),
+        # where slipcover needs unmodified output from a single sequential process.
+        coverage_mode = bool(os.environ.get("BEHAVE_PARALLEL_COVERAGE"))
+        if coverage_mode:
+            config.format = [config.default_format]
+        else:
+            config.format = [PassSuppressFormatter.name]
     return Runner(config)
 
 
-def _run_features_inprocess(
-    feature_paths: list[str], behave_args: list[str]
-) -> tuple[bool, Summary]:
+def _run_features_inprocess(paths: list[str], args: list[str]) -> tuple[bool, Summary]:
     """Run *all* feature_paths in a single behave Runner invocation.
 
     Returns ``(failed: bool, summary: dict)``.
     """
-    runner = _make_runner(feature_paths, behave_args)
+    runner = _make_runner(paths, args)
     failed = runner.run()
     summary = _extract_summary(runner)
     return failed, summary
 
 
-def _worker_run_features(
-    payload: tuple[list[str], list[str]],
-) -> tuple[bool, str, str, Summary]:
+def _worker_run_features(payload: tuple[list[str], list[str]]) -> WorkerResult:
     """Entry point for multiprocessing workers.
 
     Runs a chunk of feature files in a forked child process.  Heavy
@@ -348,9 +364,7 @@ def _worker_run_features(
 # ---------------------------------------------------------------------------
 
 
-def _aggregate_worker_results(
-    results: list[tuple[bool, str, str, Summary]],
-) -> Summary:
+def _aggregate_worker_results(results: list[WorkerResult]) -> Summary:
     """Aggregate worker results, replaying logs only for failed chunks.
 
     Iterates over the list of ``(worker_failed, stdout, stderr, summary)``

src/cleveragents/acms/index.py

@@ -10,11 +10,12 @@ Based on issue #9579 and ``docs/specification.md`` ~lines 44405-44420.
 from __future__ import annotations
 
 from collections.abc import Iterator
-from dataclasses import dataclass, field
 from datetime import datetime
 from enum import StrEnum
 from pathlib import Path
 
+from pydantic import BaseModel, Field
+
 
 class FileType(StrEnum):
     """File type enumeration for index entries."""
@@ -45,8 +46,7 @@ class TierLevel(StrEnum):
     ARCHIVE = "archive"  # Reference
 
 
-@dataclass
-class IndexEntry:
+class IndexEntry(BaseModel):
     """Represents a single indexed context entry.
 
     Attributes:
@@ -65,9 +65,9 @@ class IndexEntry:
     size_bytes: int
     created_at: datetime
     modified_at: datetime
-    tags: set[str] = field(default_factory=set)
+    tags: set[str] = Field(default_factory=set)
     tier: TierLevel = TierLevel.COLD
-    metadata: dict[str, str] = field(default_factory=dict)
+    metadata: dict[str, str] = Field(default_factory=dict)
 
     def add_tag(self, tag: str) -> None:
         """Add a tag to this entry.
@@ -104,7 +104,6 @@ class IndexEntry:
         self.tier = tier
 
 
-@dataclass
 class ACMSIndex:
     """ACMS Index for storing and querying indexed context entries.
 
@@ -115,7 +114,8 @@ class ACMSIndex:
         entries: Dictionary mapping file paths to IndexEntry objects
     """
 
-    entries: dict[str, IndexEntry] = field(default_factory=dict)
+    def __init__(self) -> None:
+        self.entries: dict[str, IndexEntry] = {}
 
     def add_entry(self, entry: IndexEntry) -> None:
         """Add an index entry to the index.