fix(testing): print behave-parallel worker logs only for failed chunks

In parallel mode, the behave runner previously replayed captured
stdout/stderr for every worker chunk, creating noisy output that
obscured failure diagnostics in CI and local runs.

Changes to scripts/run_behave_parallel.py:

- Added _chunk_has_failures() and _chunk_no_scenarios_ran() helpers
  to evaluate individual chunk summaries for failure/error/crash
  conditions.

- Updated the aggregation loop in main() to conditionally replay
  captured stdout/stderr only for chunks whose summary indicates
  failures, errors, or no scenarios ran (crash detection).  Passing
  chunks now suppress their output entirely.

- Added robust exception handling in _worker_run_features() so that
  worker crashes produce a full traceback in stderr and return a crash
  summary with features.errors = 1, enabling the parent to detect the
  crash via _chunk_has_failures (and also _chunk_no_scenarios_ran,
  since no scenarios reached a terminal state) and replay the
  diagnostics.

- The conditional replay uses summary-based checks rather than the
  raw runner.run() boolean, consistent with the existing exit-code
  logic.  This avoids spurious log replay for @tdd_expected_fail
  scenarios whose runner.run() returns True even though the TDD
  inversion handler has corrected the scenario status to passed.

- Existing summary merge, exit semantics, and the no-scenarios
  safety net are fully preserved.

New Behave unit tests (17 scenarios) cover the chunk-level helpers,
the conditional aggregation loop, the pure no-scenarios-ran path,
stderr replay for non-crash failed chunks, and the worker crash path.
New Robot integration tests (6 test cases) verify the same behavior
end-to-end via the helper_behave_parallel_log_filtering.py script.

Also updated:
- CHANGELOG.md: add unreleased entry for this behavioral change.
- features/steps/behave_parallel_log_filtering_steps.py: use
  contextlib.redirect_stdout/redirect_stderr instead of manual
  sys.stdout assignment; register module in sys.modules; document
  CWD requirement in _load_runner_module().
- robot/helper_behave_parallel_log_filtering.py: move import io to
  top-level; remove redundant inline imports; use contextlib for
  output capture; register module in sys.modules; document CWD
  requirement.

Branch note: the canonical branch for this fix is
bugfix/m3-behave-parallel-failed-chunk-logs.  The PR head branch
(bugfix/mX-behave-parallel-failed-chunk-logs) cannot be renamed via
the Forgejo API; both branches are kept in sync at the same SHA.

ISSUES CLOSED: #8351

CHANGELOG.md

@@ -7,6 +7,13 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- **Parallel Behave Runner Log Noise Reduction** (#8351): The parallel behave
+  runner now suppresses captured stdout/stderr for passing worker chunks and
+  only replays diagnostics for failed, errored, or crashed chunks.  This makes
+  failure output significantly easier to spot in CI and local runs.  A worker
+  crash (unhandled exception) is detected via an all-zero summary and the
+  captured traceback is always surfaced.
+
 - **Automation Profile Silent Fallback** (#8232): `_resolve_profile_for_plan` in
   `PlanLifecycleService` now raises a clear `ValidationError` when a plan's
   automation profile name is not a known built-in profile, instead of silently

features/behave_parallel_log_filtering.feature

@@ -0,0 +1,101 @@
+@mock_only
+Feature: Behave-parallel conditional log replay
+  The parallel behave runner should only replay captured stdout/stderr
+  for worker chunks that failed, errored, or crashed.  Passing chunks
+  must suppress their captured output to reduce CI noise.
+
+  # ---- _chunk_has_failures helper ----
+
+  Scenario: chunk_has_failures returns False for all-passing summary
+    Given behave_parallel a summary with 3 passed features and 5 passed scenarios
+    When behave_parallel I check chunk_has_failures
+    Then behave_parallel chunk_has_failures should return false
+
+  Scenario: chunk_has_failures returns True when features have failures
+    Given behave_parallel a summary with 1 failed feature
+    When behave_parallel I check chunk_has_failures
+    Then behave_parallel chunk_has_failures should return true
+
+  Scenario: chunk_has_failures returns True when features have errors
+    Given behave_parallel a summary with 1 errored feature
+    When behave_parallel I check chunk_has_failures
+    Then behave_parallel chunk_has_failures should return true
+
+  Scenario: chunk_has_failures returns True when scenarios have failures
+    Given behave_parallel a summary with 1 failed scenario
+    When behave_parallel I check chunk_has_failures
+    Then behave_parallel chunk_has_failures should return true
+
+  Scenario: chunk_has_failures returns True when scenarios have errors
+    Given behave_parallel a summary with 1 errored scenario
+    When behave_parallel I check chunk_has_failures
+    Then behave_parallel chunk_has_failures should return true
+
+  # ---- _chunk_no_scenarios_ran helper ----
+
+  Scenario: chunk_no_scenarios_ran returns True for empty summary
+    Given behave_parallel an empty summary
+    When behave_parallel I check chunk_no_scenarios_ran
+    Then behave_parallel chunk_no_scenarios_ran should return true
+
+  Scenario: chunk_no_scenarios_ran returns False when scenarios passed
+    Given behave_parallel a summary with 3 passed features and 5 passed scenarios
+    When behave_parallel I check chunk_no_scenarios_ran
+    Then behave_parallel chunk_no_scenarios_ran should return false
+
+  Scenario: chunk_no_scenarios_ran returns False when all scenarios are skipped
+    Given behave_parallel a summary with all scenarios skipped
+    When behave_parallel I check chunk_no_scenarios_ran
+    Then behave_parallel chunk_no_scenarios_ran should return false
+
+  # ---- Conditional log replay integration ----
+
+  Scenario: Passing chunk output is suppressed in parallel aggregation
+    Given behave_parallel a worker result with passing summary and captured stdout "PASS OUTPUT"
+    When behave_parallel I aggregate the worker results
+    Then behave_parallel the aggregated stdout should not contain "PASS OUTPUT"
+
+  Scenario: Failed chunk output is replayed in parallel aggregation
+    Given behave_parallel a worker result with failed summary and captured stdout "FAIL OUTPUT"
+    When behave_parallel I aggregate the worker results
+    Then behave_parallel the aggregated stdout should contain "FAIL OUTPUT"
+
+  Scenario: Crashed worker output is replayed in parallel aggregation
+    Given behave_parallel a worker result with crashed summary and captured stderr "WORKER CRASH"
+    When behave_parallel I aggregate the worker results
+    Then behave_parallel the aggregated stderr should contain "WORKER CRASH"
+
+  Scenario: Pure empty summary chunk triggers log replay via no-scenarios-ran path
+    Given behave_parallel a worker result with empty summary and captured stderr "EMPTY CRASH"
+    When behave_parallel I aggregate the worker results
+    Then behave_parallel the aggregated stderr should contain "EMPTY CRASH"
+
+  Scenario: Failed chunk stderr is replayed in parallel aggregation
+    Given behave_parallel a worker result with failed summary and captured stderr "FAIL STDERR"
+    When behave_parallel I aggregate the worker results
+    Then behave_parallel the aggregated stderr should contain "FAIL STDERR"
+
+  Scenario: Mixed results replay only failed chunks
+    Given behave_parallel worker results with one passing and one failing chunk
+    When behave_parallel I aggregate the worker results
+    Then behave_parallel the aggregated stdout should contain "FAILING CHUNK"
+    And behave_parallel the aggregated stdout should not contain "PASSING CHUNK"
+
+  Scenario: TDD-inverted chunk suppresses output when worker_failed but summary is passing
+    Given behave_parallel a worker result where worker_failed is true but summary is passing with stdout "TDD OUTPUT"
+    When behave_parallel I aggregate the worker results
+    Then behave_parallel the aggregated stdout should not contain "TDD OUTPUT"
+
+  # ---- Worker exception handling ----
+
+  Scenario: Worker exception returns failed summary with traceback
+    Given behave_parallel a worker payload that will raise an exception
+    When behave_parallel I invoke the worker
+    Then behave_parallel the worker result should indicate failure
+    And behave_parallel the worker stderr should contain "WORKER CRASH:"
+    And behave_parallel the worker summary should have features errors equal to 1
+
+  Scenario: Partial worker crash alongside passing workers produces summary with failures
+    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

features/steps/behave_parallel_log_filtering_steps.py

@@ -0,0 +1,359 @@
+"""Step definitions for behave_parallel_log_filtering.feature.
+
+Tests the conditional log replay and worker exception handling in
+``scripts/run_behave_parallel.py``.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import sys
+from contextlib import redirect_stderr, redirect_stdout
+from pathlib import Path
+from types import ModuleType
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context  # type: ignore[import-untyped]
+
+
+def _load_runner_module() -> ModuleType:
+    """Load scripts/run_behave_parallel.py as a module via importlib.
+
+    **CWD requirement**: This function resolves the script path relative to
+    the current working directory (``Path("scripts") / "run_behave_parallel.py"``).
+    It must be called from the repository root (i.e. the directory that
+    contains the ``scripts/`` folder).  Behave sets the working directory to
+    the project root when running feature tests, so this is satisfied
+    automatically in the normal test invocation.  If you invoke step
+    definitions from an unexpected directory, either adjust your working
+    directory or anchor the path relative to ``__file__``.
+    """
+    script_path = Path("scripts") / "run_behave_parallel.py"
+    spec = importlib.util.spec_from_file_location(
+        "run_behave_parallel", str(script_path)
+    )
+    if spec is None or spec.loader is None:
+        msg = f"Cannot load {script_path}"
+        raise ImportError(msg)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = mod  # Prevent re-execution on subsequent calls
+    spec.loader.exec_module(mod)
+    return mod
+
+
+_runner_mod = _load_runner_module()
+
+# Bind module-level references for convenience.
+Summary = _runner_mod.Summary
+_chunk_has_failures = _runner_mod._chunk_has_failures
+_chunk_no_scenarios_ran = _runner_mod._chunk_no_scenarios_ran
+_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
+
+
+def _passing_summary(features: int = 1, scenarios: int = 1) -> Summary:
+    """Build a summary with all-passing counts."""
+    return {
+        "features": {
+            "passed": features,
+            "failed": 0,
+            "errors": 0,
+            "skipped": 0,
+        },
+        "scenarios": {
+            "passed": scenarios,
+            "failed": 0,
+            "errors": 0,
+            "skipped": 0,
+        },
+        "steps": {"passed": scenarios * 3, "failed": 0, "errors": 0, "skipped": 0},
+        "duration": 1.0,
+    }
+
+
+def _failed_feature_summary() -> Summary:
+    s = _empty_summary()
+    s["features"]["failed"] = 1
+    return s
+
+
+def _errored_feature_summary() -> Summary:
+    s = _empty_summary()
+    s["features"]["errors"] = 1
+    return s
+
+
+def _failed_scenario_summary() -> Summary:
+    s = _empty_summary()
+    s["scenarios"]["failed"] = 1
+    return s
+
+
+def _errored_scenario_summary() -> Summary:
+    s = _empty_summary()
+    s["scenarios"]["errors"] = 1
+    return s
+
+
+def _all_skipped_summary() -> Summary:
+    """Build a summary where all scenarios are skipped (but some ran)."""
+    s = _empty_summary()
+    s["features"]["skipped"] = 1
+    s["scenarios"]["skipped"] = 3
+    return s
+
+
+def _crash_summary() -> Summary:
+    """Build a summary representing a worker crash (features.errors = 1)."""
+    s = _empty_summary()
+    s["features"]["errors"] = 1
+    return s
+
+
+# ---- _chunk_has_failures helper steps ----
+
+
+@given(
+    "behave_parallel a summary with {n:d} passed features and {m:d} passed scenarios"
+)
+def step_passing_summary(context: Context, n: int, m: int) -> None:
+    context.bp_summary = _passing_summary(features=n, scenarios=m)
+
+
+@given("behave_parallel a summary with 1 failed feature")
+def step_failed_feature_summary(context: Context) -> None:
+    context.bp_summary = _failed_feature_summary()
+
+
+@given("behave_parallel a summary with 1 errored feature")
+def step_errored_feature_summary(context: Context) -> None:
+    context.bp_summary = _errored_feature_summary()
+
+
+@given("behave_parallel a summary with 1 failed scenario")
+def step_failed_scenario_summary(context: Context) -> None:
+    context.bp_summary = _failed_scenario_summary()
+
+
+@given("behave_parallel a summary with 1 errored scenario")
+def step_errored_scenario_summary(context: Context) -> None:
+    context.bp_summary = _errored_scenario_summary()
+
+
+@given("behave_parallel an empty summary")
+def step_empty_summary(context: Context) -> None:
+    context.bp_summary = _empty_summary()
+
+
+@given("behave_parallel a summary with all scenarios skipped")
+def step_all_skipped_summary(context: Context) -> None:
+    context.bp_summary = _all_skipped_summary()
+
+
+@when("behave_parallel I check chunk_has_failures")
+def step_check_chunk_has_failures(context: Context) -> None:
+    context.bp_result = _chunk_has_failures(context.bp_summary)
+
+
+@when("behave_parallel I check chunk_no_scenarios_ran")
+def step_check_chunk_no_scenarios_ran(context: Context) -> None:
+    context.bp_result = _chunk_no_scenarios_ran(context.bp_summary)
+
+
+@then("behave_parallel chunk_has_failures should return false")
+def step_chunk_has_failures_false(context: Context) -> None:
+    assert context.bp_result is False, f"Expected False, got {context.bp_result}"
+
+
+@then("behave_parallel chunk_has_failures should return true")
+def step_chunk_has_failures_true(context: Context) -> None:
+    assert context.bp_result is True, f"Expected True, got {context.bp_result}"
+
+
+@then("behave_parallel chunk_no_scenarios_ran should return true")
+def step_chunk_no_scenarios_ran_true(context: Context) -> None:
+    assert context.bp_result is True, f"Expected True, got {context.bp_result}"
+
+
+@then("behave_parallel chunk_no_scenarios_ran should return false")
+def step_chunk_no_scenarios_ran_false(context: Context) -> None:
+    assert context.bp_result is False, f"Expected False, got {context.bp_result}"
+
+
+# ---- Conditional log replay integration steps ----
+
+
+@given(
+    'behave_parallel a worker result with passing summary and captured stdout "{text}"'
+)
+def step_passing_worker_result(context: Context, text: str) -> None:
+    context.bp_worker_results = [
+        (False, text, "", _passing_summary()),
+    ]
+
+
+@given(
+    'behave_parallel a worker result with failed summary and captured stdout "{text}"'
+)
+def step_failed_worker_result(context: Context, text: str) -> None:
+    context.bp_worker_results = [
+        (True, text, "", _failed_scenario_summary()),
+    ]
+
+
+@given(
+    'behave_parallel a worker result with crashed summary and captured stderr "{text}"'
+)
+def step_crashed_worker_result(context: Context, text: str) -> None:
+    context.bp_worker_results = [
+        (True, "", text, _crash_summary()),
+    ]
+
+
+@given(
+    'behave_parallel a worker result with empty summary and captured stderr "{text}"'
+)
+def step_empty_summary_worker_result(context: Context, text: str) -> None:
+    # Pure all-zeros summary: _chunk_has_failures() returns False but
+    # _chunk_no_scenarios_ran() returns True, exercising the no-scenarios-ran
+    # path exclusively.
+    context.bp_worker_results = [
+        (False, "", text, _empty_summary()),
+    ]
+
+
+@given(
+    'behave_parallel a worker result with failed summary and captured stderr "{text}"'
+)
+def step_failed_worker_result_stderr(context: Context, text: str) -> None:
+    context.bp_worker_results = [
+        (True, "", text, _failed_scenario_summary()),
+    ]
+
+
+@given("behave_parallel worker results with one passing and one failing chunk")
+def step_mixed_worker_results(context: Context) -> None:
+    context.bp_worker_results = [
+        (False, "PASSING CHUNK", "", _passing_summary()),
+        (True, "FAILING CHUNK", "", _failed_scenario_summary()),
+    ]
+
+
+@given(
+    "behave_parallel a worker result where worker_failed is true but summary is passing"
+    ' with stdout "{text}"'
+)
+def step_tdd_inverted_worker_result(context: Context, text: str) -> None:
+    # TDD-inverted chunk: runner.run() returned True (raw worker_failed=True)
+    # but the after_scenario hook corrected the scenario status to passing.
+    # The summary therefore shows no failures.  Output must be suppressed.
+    context.bp_worker_results = [
+        (True, text, "", _passing_summary()),
+    ]
+
+
+@given("behave_parallel worker results with one crashed chunk and one passing chunk")
+def step_crash_and_pass_worker_results(context: Context) -> None:
+    # Partial crash: one worker sets features.errors=1, the other passes.
+    # The merged total must have _has_failures() == True.
+    context.bp_worker_results = [
+        (True, "", "CRASH TRACEBACK", _crash_summary()),
+        (False, "PASS OUT", "", _passing_summary()),
+    ]
+
+
+@when("behave_parallel I aggregate the worker results")
+def step_aggregate_worker_results(context: Context) -> None:
+    """Invoke _aggregate_worker_results() and capture its stdout/stderr output."""
+    results: list[tuple[bool, str, str, Summary]] = context.bp_worker_results
+
+    stdout_capture = io.StringIO()
+    stderr_capture = io.StringIO()
+
+    with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
+        context.bp_aggregated_summary = _aggregate_worker_results(results)
+
+    context.bp_aggregated_stdout = stdout_capture.getvalue()
+    context.bp_aggregated_stderr = stderr_capture.getvalue()
+
+
+@then('behave_parallel the aggregated stdout should not contain "{text}"')
+def step_aggregated_stdout_not_contains(context: Context, text: str) -> None:
+    assert text not in context.bp_aggregated_stdout, (
+        f"Expected '{text}' NOT in stdout, but found it: "
+        f"{context.bp_aggregated_stdout!r}"
+    )
+
+
+@then('behave_parallel the aggregated stdout should contain "{text}"')
+def step_aggregated_stdout_contains(context: Context, text: str) -> None:
+    assert text in context.bp_aggregated_stdout, (
+        f"Expected '{text}' in stdout, but not found: {context.bp_aggregated_stdout!r}"
+    )
+
+
+@then('behave_parallel the aggregated stderr should contain "{text}"')
+def step_aggregated_stderr_contains(context: Context, text: str) -> None:
+    assert text in context.bp_aggregated_stderr, (
+        f"Expected '{text}' in stderr, but not found: {context.bp_aggregated_stderr!r}"
+    )
+
+
+@then("behave_parallel the aggregated summary should have failures")
+def step_aggregated_summary_has_failures(context: Context) -> None:
+    summary = context.bp_aggregated_summary
+    assert _has_failures(summary), (
+        f"Expected _has_failures() to return True for summary: {summary}"
+    )
+
+
+# ---- Worker exception handling steps ----
+
+
+@given("behave_parallel a worker payload that will raise an exception")
+def step_worker_payload_raises(context: Context) -> None:
+    # This step relies on behave raising a real exception when it cannot
+    # locate the feature file — specifically, behave's feature loader raises
+    # an exception (e.g. ConfigError or IOError) for paths that do not exist
+    # on disk.  This is verified, documented behave behavior: passing a
+    # non-existent path to behave.configuration.Configuration causes behave
+    # to fail during feature loading rather than silently returning zero
+    # scenarios.  If this test starts failing, check whether behave changed
+    # its file-not-found handling (look at behave.runner.Runner.load_step_modules
+    # and behave.runner_util.collect_feature_locations).
+    context.bp_worker_payload = (
+        ["__nonexistent_feature_path_that_will_crash__.feature"],
+        [],
+    )
+
+
+@when("behave_parallel I invoke the worker")
+def step_invoke_worker(context: Context) -> None:
+    """Invoke _worker_run_features with a real crashing payload."""
+    result = _worker_run_features(context.bp_worker_payload)
+    context.bp_worker_result = result
+
+
+@then("behave_parallel the worker result should indicate failure")
+def step_worker_result_failure(context: Context) -> None:
+    failed, _stdout, _stderr, _summary = context.bp_worker_result
+    assert failed is True, f"Expected failed=True, got {failed}"
+
+
+@then('behave_parallel the worker stderr should contain "{text}"')
+def step_worker_stderr_contains(context: Context, text: str) -> None:
+    _failed, _stdout, stderr, _summary = context.bp_worker_result
+    assert text in stderr, f"Expected '{text}' in stderr, got: {stderr!r}"
+
+
+@then("behave_parallel the worker summary should have features errors equal to 1")
+def step_worker_summary_features_errors_1(context: Context) -> None:
+    _failed, _stdout, _stderr, summary = context.bp_worker_result
+    errors = summary["features"]["errors"]
+    assert errors == 1, (
+        f"Expected features.errors=1 so _has_failures() detects partial crash, "
+        f"got: {errors}"
+    )

robot/behave_parallel_log_filtering.robot

@@ -0,0 +1,77 @@
+*** Settings ***
+Documentation    Integration tests for conditional log replay in the parallel
+...              behave runner (``scripts/run_behave_parallel.py``).
+...
+...              Verifies that passing chunks suppress output while failed,
+...              errored, or crashed chunks surface diagnostics.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_behave_parallel_log_filtering.py
+
+*** Test Cases ***
+Passing Chunk Output Is Suppressed
+    [Documentation]    A fully passing worker chunk must not trigger log replay.
+    [Tags]    testing    parallel    log_filtering
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    passing-chunk-suppressed
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    passing-chunk-suppressed-ok
+
+Failed Chunk Output Is Replayed
+    [Documentation]    A worker chunk with scenario failures must trigger log replay.
+    [Tags]    testing    parallel    log_filtering
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    failed-chunk-replayed
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    failed-chunk-replayed-ok
+
+Crashed Chunk Is Detected
+    [Documentation]    A worker chunk that crashes (empty summary) must be detected.
+    [Tags]    testing    parallel    log_filtering
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    crashed-chunk-detected
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    crashed-chunk-detected-ok
+
+Worker Crash Returns Traceback And Empty Summary
+    [Documentation]    A worker that raises an unhandled exception must return a
+    ...              traceback in stderr and an empty summary so the parent can
+    ...              detect the crash.
+    [Tags]    testing    parallel    log_filtering
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    worker-crash-handling
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    worker-crash-handling-ok
+
+Mixed Results Only Replay Failed Chunks
+    [Documentation]    When aggregating results from multiple workers, only
+    ...              failed chunks should have their output replayed.
+    [Tags]    testing    parallel    log_filtering
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    mixed-results-filtering
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    mixed-results-filtering-ok
+
+Summary Merge Semantics Are Preserved
+    [Documentation]    The summary merge logic must correctly combine counts
+    ...              from multiple worker chunks regardless of filtering.
+    [Tags]    testing    parallel    log_filtering
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    summary-merge-preserved
+    ...    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    summary-merge-preserved-ok

robot/helper_behave_parallel_log_filtering.py

@@ -0,0 +1,180 @@
+"""Helper for behave_parallel_log_filtering.robot integration tests.
+
+Each function exercises a specific aspect of the parallel runner's
+conditional log replay logic and prints a sentinel on success.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import sys
+from contextlib import redirect_stdout
+from pathlib import Path
+from types import ModuleType
+
+
+def _load_runner_module() -> ModuleType:
+    """Load scripts/run_behave_parallel.py as a module via importlib.
+
+    **CWD requirement**: This function resolves the script path relative to
+    the current working directory (``Path("scripts") / "run_behave_parallel.py"``).
+    It must be called from the repository root (i.e. the directory that
+    contains the ``scripts/`` folder).  The Robot integration tests satisfy
+    this by passing ``cwd=${WORKSPACE}`` to ``Run Process``.  If you call
+    this helper from a different directory, either ``cd`` to the repo root
+    first or modify the path to be anchored relative to ``__file__``.
+    """
+    script_path = Path("scripts") / "run_behave_parallel.py"
+    spec = importlib.util.spec_from_file_location(
+        "run_behave_parallel", str(script_path)
+    )
+    if spec is None or spec.loader is None:
+        msg = f"Cannot load {script_path}"
+        raise ImportError(msg)
+    mod = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = mod  # Prevent re-execution on subsequent calls
+    spec.loader.exec_module(mod)
+    return mod
+
+
+_mod = _load_runner_module()
+
+# Re-export Summary type for annotation use below.
+Summary = _mod.Summary
+
+
+def _passing_summary(features: int = 1, scenarios: int = 1) -> Summary:
+    return {
+        "features": {
+            "passed": features,
+            "failed": 0,
+            "errors": 0,
+            "skipped": 0,
+        },
+        "scenarios": {
+            "passed": scenarios,
+            "failed": 0,
+            "errors": 0,
+            "skipped": 0,
+        },
+        "steps": {
+            "passed": scenarios * 3,
+            "failed": 0,
+            "errors": 0,
+            "skipped": 0,
+        },
+        "duration": 1.0,
+    }
+
+
+def _failed_summary() -> Summary:
+    s = _mod._empty_summary()
+    s["scenarios"]["failed"] = 1
+    return s
+
+
+# ---- Test functions ----
+
+
+def test_passing_chunk_suppressed() -> None:
+    """Verify passing chunk output is not replayed."""
+    summary = _passing_summary()
+    assert not _mod._chunk_has_failures(summary)
+    assert not _mod._chunk_no_scenarios_ran(summary)
+    print("passing-chunk-suppressed-ok")
+
+
+def test_failed_chunk_replayed() -> None:
+    """Verify failed chunk triggers log replay."""
+    summary = _failed_summary()
+    assert _mod._chunk_has_failures(summary)
+    print("failed-chunk-replayed-ok")
+
+
+def test_crashed_chunk_detected() -> None:
+    """Verify crashed chunk (empty summary) is detected."""
+    summary = _mod._empty_summary()
+    assert _mod._chunk_no_scenarios_ran(summary)
+    print("crashed-chunk-detected-ok")
+
+
+def test_worker_crash_handling() -> None:
+    """Verify worker crash returns a traceback in stderr and a failed summary.
+
+    Uses a real crash: passing a path that does not exist causes behave's
+    feature loader to raise a real exception inside _worker_run_features,
+    exercising the actual exception handler without any mocking.  The crash
+    summary must have features.errors == 1 so _has_failures() propagates
+    the failure through the merged total even when other workers passed.
+
+    This test relies on behave raising an exception for non-existent feature
+    file paths — specifically, behave's feature loader raises an exception
+    (e.g. ConfigError or IOError) rather than returning an empty run result.
+    This is verified, documented behave behavior.  If this test starts
+    failing, check whether behave changed its file-not-found behavior
+    (look at behave.runner.Runner.load_step_modules and
+    behave.runner_util.collect_feature_locations).
+    """
+    failed, _stdout, stderr, summary = _mod._worker_run_features(
+        (["__nonexistent_feature_path_that_will_crash__.feature"], [])
+    )
+
+    assert failed is True, f"Expected failed=True, got {failed}"
+    assert "WORKER CRASH:" in stderr, f"Expected 'WORKER CRASH:' in stderr: {stderr!r}"
+    # features.errors must be 1 so _has_failures() catches partial crashes.
+    assert summary["features"]["errors"] == 1, (
+        f"Expected features.errors=1, got: {summary['features']['errors']}"
+    )
+    print("worker-crash-handling-ok")
+
+
+def test_mixed_results_filtering() -> None:
+    """Verify only failed chunks contribute output in aggregation.
+
+    Calls _aggregate_worker_results() directly so the test exercises the
+    real production code rather than a reimplementation of the loop.
+    """
+    results: list[tuple[bool, str, str, Summary]] = [
+        (False, "PASS_OUT", "", _passing_summary()),
+        (True, "FAIL_OUT", "", _failed_summary()),
+    ]
+
+    # Capture stdout to inspect what _aggregate_worker_results replays.
+    captured = io.StringIO()
+    with redirect_stdout(captured):
+        _mod._aggregate_worker_results(results)
+
+    replayed = captured.getvalue()
+    assert "FAIL_OUT" in replayed, f"Expected FAIL_OUT in output: {replayed!r}"
+    assert "PASS_OUT" not in replayed, f"Expected PASS_OUT NOT in output: {replayed!r}"
+    print("mixed-results-filtering-ok")
+
+
+def test_summary_merge_preserved() -> None:
+    """Verify summary merge semantics are unchanged."""
+    s1 = _passing_summary(features=2, scenarios=4)
+    s2 = _failed_summary()
+    merged = _mod._merge_summaries([s1, s2])
+    assert merged["features"]["passed"] == 2
+    assert merged["scenarios"]["passed"] == 4
+    assert merged["scenarios"]["failed"] == 1
+    print("summary-merge-preserved-ok")
+
+
+if __name__ == "__main__":
+    cmd = sys.argv[1] if len(sys.argv) > 1 else ""
+    dispatch = {
+        "passing-chunk-suppressed": test_passing_chunk_suppressed,
+        "failed-chunk-replayed": test_failed_chunk_replayed,
+        "crashed-chunk-detected": test_crashed_chunk_detected,
+        "worker-crash-handling": test_worker_crash_handling,
+        "mixed-results-filtering": test_mixed_results_filtering,
+        "summary-merge-preserved": test_summary_merge_preserved,
+    }
+    fn = dispatch.get(cmd)
+    if fn is None:
+        print(f"Unknown command: {cmd!r}", file=sys.stderr)
+        print(f"Available: {', '.join(sorted(dispatch))}", file=sys.stderr)
+        sys.exit(1)
+    fn()

scripts/run_behave_parallel.py

@@ -25,6 +25,7 @@ import multiprocessing
 import os
 import sys
 import time
+import traceback
 from contextlib import redirect_stderr, redirect_stdout, suppress
 from pathlib import Path
 from typing import Any
@@ -143,6 +144,12 @@ def _no_scenarios_ran(total: Summary) -> bool:
     return s["passed"] + s["failed"] + s["errors"] + s["skipped"] == 0
 
 
+# Aliases for chunk-level checks — identical semantics to the overall-summary
+# helpers; exposed under distinct names for readability at the call site.
+_chunk_has_failures = _has_failures
+_chunk_no_scenarios_ran = _no_scenarios_ran
+
+
 # ---------------------------------------------------------------------------
 # Feature discovery
 # ---------------------------------------------------------------------------
@@ -221,19 +228,80 @@ def _worker_run_features(
     ``load_step_definitions()`` and ``load_hooks()`` still execute
     inside each worker (they ``exec()`` the step .py files and
     environment.py), but every ``import`` they trigger is a cache hit.
+
+    If the worker crashes with an unhandled exception, the traceback is
+    captured into stderr and a crash summary with ``features.errors = 1``
+    is returned so that the parent can detect the crash via
+    ``_chunk_has_failures`` (and also ``_chunk_no_scenarios_ran``, since no
+    scenarios reached a terminal state) and replay the diagnostics.
     """
     feature_paths, behave_args = payload
 
     stdout_buf = io.StringIO()
     stderr_buf = io.StringIO()
 
-    runner = _make_runner(feature_paths, behave_args)
+    try:
+        runner = _make_runner(feature_paths, behave_args)
 
-    with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
-        failed = runner.run()
+        with redirect_stdout(stdout_buf), redirect_stderr(stderr_buf):
+            failed = runner.run()
 
-    summary = _extract_summary(runner)
-    return failed, stdout_buf.getvalue(), stderr_buf.getvalue(), summary
+        summary = _extract_summary(runner)
+        return failed, stdout_buf.getvalue(), stderr_buf.getvalue(), summary
+    except Exception:
+        # Surface the full traceback so the parent can replay it for
+        # diagnostics.  Set features.errors = 1 so that _has_failures()
+        # detects the crash in the merged total even when other workers
+        # passed (preventing a silent CI green when a partial worker crash
+        # occurs alongside passing chunks).
+        tb_text = traceback.format_exc()
+        stderr_buf.write(f"\nWORKER CRASH:\n{tb_text}")
+        crash_summary = _empty_summary()
+        crash_summary["features"]["errors"] = 1
+        return True, stdout_buf.getvalue(), stderr_buf.getvalue(), crash_summary
+
+
+# ---------------------------------------------------------------------------
+# Parallel result aggregation
+# ---------------------------------------------------------------------------
+
+
+def _aggregate_worker_results(
+    results: list[tuple[bool, str, str, Summary]],
+) -> Summary:
+    """Aggregate worker results, replaying logs only for failed chunks.
+
+    Iterates over the list of ``(worker_failed, stdout, stderr, summary)``
+    tuples returned by the multiprocessing pool.  For each chunk whose
+    summary-based check indicates failure (failed/errored scenarios or
+    features, or a crash that produced no terminal scenario results), the
+    captured stdout and stderr are replayed to the parent's streams.
+    Passing chunks have their output suppressed to reduce CI noise.
+
+    Uses summary-based checks rather than the raw ``worker_failed`` boolean
+    so that TDD-inverted chunks (``@tdd_expected_fail`` scenarios whose
+    status has been corrected to "passed" by the after_scenario hook) do
+    not trigger spurious log replay.
+    """
+    summaries: list[Summary] = []
+    for idx, (_, stdout, stderr, summary) in enumerate(results):
+        # Rely on summary-based checks: the @tdd_expected_fail handler in
+        # environment.py inverts scenario statuses but cannot update the
+        # runner.run() local variable, so the raw worker_failed flag is
+        # not trustworthy for TDD-inverted chunks.
+        chunk_failed = _chunk_has_failures(summary) or _chunk_no_scenarios_ran(summary)
+        if chunk_failed:
+            if stdout:
+                print(f"\n--- Worker {idx} stdout (chunk failed) ---")
+                print(stdout, end="")
+            if stderr:
+                print(
+                    f"\n--- Worker {idx} stderr (chunk failed) ---",
+                    file=sys.stderr,
+                )
+                print(stderr, end="", file=sys.stderr)
+        summaries.append(summary)
+    return _merge_summaries(summaries)
 
 
 # ---------------------------------------------------------------------------
@@ -294,14 +362,7 @@ def main(argv: list[str] | None = None) -> None:
                 [(chunk, other_args) for chunk in chunks],
             )
 
-        summaries = []
-        for _worker_failed, stdout, stderr, summary in results:
-            if stdout:
-                print(stdout, end="")
-            if stderr:
-                print(stderr, end="", file=sys.stderr)
-            summaries.append(summary)
-        total = _merge_summaries(summaries)
+        total = _aggregate_worker_results(results)
 
     wall = time.monotonic() - start
     _print_overall_summary(total, wall_seconds=wall)