.forgejo/workflows/ci.yml

@@ -290,18 +290,43 @@ jobs:
         container:
             image: python:3.13-slim
         needs: [lint, typecheck, security, quality, unit_tests]
+        # Bound the job so a hung run fails cleanly with diagnostics instead of
+        # being externally reaped to "no data". The parallel engine runs in
+        # ~4min; this leaves generous headroom for install + chunk tail.
+        timeout-minutes: 30
         steps:
+            # Operator kill switch: repo variable skip_coverage. When set to
+            # "true", every real step below is gated off and the job returns
+            # success without running the (long) coverage work. This guard step
+            # always runs, so the job has a successful step and the status check
+            # sees coverage == success. Lets ops disable coverage fast without a
+            # deploy.
+            - name: Resolve skip_coverage gate
+              id: gate
+              run: |
+                  v="$(printf '%s' "${{ vars.skip_coverage }}" | tr '[:upper:]' '[:lower:]')"
+                  if [ "$v" = "true" ]; then
+                      echo "coverage SKIPPED — repo variable skip_coverage=$v"
+                      echo "run=false" >> "$GITHUB_OUTPUT"
+                  else
+                      echo "run=true" >> "$GITHUB_OUTPUT"
+                  fi
+
             - name: Install system dependencies (nodejs for checkout, git for merge tests)
+              if: steps.gate.outputs.run == 'true'
               run: |
                   apt-get update && apt-get install -y -qq nodejs git && rm -rf /var/lib/apt/lists/*
 
             - uses: actions/checkout@v4
+              if: steps.gate.outputs.run == 'true'
 
             - name: Install uv and nox
+              if: steps.gate.outputs.run == 'true'
               run: |
                   pip install -q uv==${{ env.UV_VERSION }} nox
 
             - name: Cache uv packages
+              if: steps.gate.outputs.run == 'true'
               uses: actions/cache@v3
               with:
                   path: ~/.cache/uv
@@ -309,40 +334,28 @@ jobs:
                   restore-keys: |
                       uv-
 
-            - name: Run coverage report via nox (fail-under 97%)
+            - name: Run coverage report via nox (fail-under from pyproject)
               id: coverage
+              if: steps.gate.outputs.run == 'true'
+              # The coverage gate is nox's own --fail-under (sourced from
+              # pyproject [tool.coverage.report].fail_under). Propagate nox's
+              # exit EXPLICITLY via PIPESTATUS so gating does not rely on the
+              # runner's implicit `bash -eo pipefail` default and cannot be
+              # silently un-gated.
+              shell: bash
               run: |
+                  set -uo pipefail
                   mkdir -p build
                   nox -s coverage_report 2>&1 | tee build/nox-coverage-output.log
-                  # Extract the single-line CI summary from nox output
+                  rc=${PIPESTATUS[0]}
+                  # Surface the load-bearing single-line CI summary the pipeline greps.
                   grep -E '^(nox > )?COVERAGE (OK|FAILED):' build/nox-coverage-output.log || true
+                  exit "$rc"
               env:
                   NOX_DEFAULT_VENV_BACKEND: uv
 
-            - name: Surface coverage summary
-              if: always()
-              run: |
-                  if [ -f build/coverage.json ]; then
-                    python3 -c "
-                  import json, sys
-                  with open('build/coverage.json') as f:
-                      data = json.load(f)
-                  summary = data.get('summary') or data.get('totals') or {}
-                  pct = round(summary.get('percent_covered', 0), 1)
-                  threshold = 50  # Temporarily lowered; see issues #4183 and #4184
-                  if pct >= threshold:
-                      print(f'COVERAGE OK: {pct}% (threshold: {threshold}%)')
-                  else:
-                      print(f'COVERAGE FAILED: {pct}% < {threshold}% threshold')
-                      sys.exit(1)
-                  "
-                  else
-                    echo "COVERAGE FAILED: no coverage data generated"
-                    exit 1
-                  fi
-
             - name: Upload coverage log artifact
-              if: always()
+              if: always() && steps.gate.outputs.run == 'true'
               uses: actions/upload-artifact@v3
               with:
                   name: ci-logs-coverage
@@ -350,7 +363,7 @@ jobs:
                   retention-days: 30
 
             - name: Upload coverage artifacts
-              if: always()
+              if: always() && steps.gate.outputs.run == 'true'
               uses: actions/upload-artifact@v3
               with:
                   name: coverage-reports

features/coverage_threshold_enforcement.feature

@@ -1,5 +1,5 @@
 Feature: Coverage threshold enforcement
-  The project enforces a minimum 97% code coverage threshold.
+  The project enforces a minimum 96.5% code coverage threshold.
   This feature validates that the coverage configuration and nox session
   are properly set up to enforce this requirement.
 

features/coverage_validation_error_paths.feature

@@ -0,0 +1,146 @@
+@coverage
+Feature: Structural validation error-path coverage
+  Exercises uncovered defensive branches in cleveragents.core.validation
+  (plan-tree, decision-dict, and structured-output validators) that the
+  existing structural_validation.feature does not reach. Pure-function tests
+  routed through the public validate_structured_component_output dispatcher.
+
+  # -- plan tree validator --
+
+  Scenario: Plan tree node that is not a dict
+    When I structurally validate "plan_tree" with the JSON payload
+      """
+      [123]
+      """
+    Then the structural validation fails
+    And the structural errors include "is not a dict"
+
+  Scenario: Plan tree with a duplicate decision_id
+    When I structurally validate "plan_tree" with the JSON payload
+      """
+      [{"decision_id": "01ARZ3NDEKTSV4XXFFJFRC889A"}, {"decision_id": "01ARZ3NDEKTSV4XXFFJFRC889A"}]
+      """
+    Then the structural validation fails
+    And the structural errors include "duplicate decision_id"
+
+  Scenario: Plan tree node with a blank type
+    When I structurally validate "plan_tree" with the JSON payload
+      """
+      [{"type": "   "}]
+      """
+    Then the structural validation fails
+    And the structural errors include "'type' must be a non-empty string"
+
+  Scenario: Plan tree node with a blank question
+    When I structurally validate "plan_tree" with the JSON payload
+      """
+      [{"question": ""}]
+      """
+    Then the structural validation fails
+    And the structural errors include "'question' must be a non-empty string"
+
+  Scenario: Plan tree children is not a list
+    When I structurally validate "plan_tree" with the JSON payload
+      """
+      [{"children": "nope"}]
+      """
+    Then the structural validation fails
+    And the structural errors include "'children' must be a list"
+
+  Scenario: Plan tree child with an invalid ULID
+    When I structurally validate "plan_tree" with the JSON payload
+      """
+      [{"children": [{"decision_id": "not-a-ulid"}]}]
+      """
+    Then the structural validation fails
+    And the structural errors include "must be a valid ULID"
+
+  # -- decision dict validator --
+
+  Scenario: Decision dict missing all required fields
+    When I structurally validate "decision" with the JSON payload
+      """
+      {}
+      """
+    Then the structural validation fails
+    And the structural errors include "missing required field"
+
+  Scenario: Decision dict with an invalid decision_id ULID
+    When I structurally validate "decision" with the JSON payload
+      """
+      {"decision_id": "bad", "plan_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "type": "t", "sequence": 0, "question": "q", "chosen": "c", "confidence": 0.5, "parent": "(root)", "is_correction": false, "superseded": false}
+      """
+    Then the structural validation fails
+    And the structural errors include "must be a valid ULID"
+
+  Scenario: Decision dict with a wrong-typed confidence
+    When I structurally validate "decision" with the JSON payload
+      """
+      {"decision_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "plan_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "type": "t", "sequence": 0, "question": "q", "chosen": "c", "confidence": "high", "parent": "(root)", "is_correction": false, "superseded": false}
+      """
+    Then the structural validation fails
+    And the structural errors include "must be float or None"
+
+  Scenario: Decision dict with a non-string parent
+    When I structurally validate "decision" with the JSON payload
+      """
+      {"decision_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "plan_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "type": "t", "sequence": 0, "question": "q", "chosen": "c", "confidence": null, "parent": 123, "is_correction": false, "superseded": false}
+      """
+    Then the structural validation fails
+    And the structural errors include "'parent' must be a string"
+
+  Scenario: Decision dict with a non-integer sequence
+    When I structurally validate "decision" with the JSON payload
+      """
+      {"decision_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "plan_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "type": "t", "sequence": "five", "question": "q", "chosen": "c", "confidence": null, "parent": "(root)", "is_correction": false, "superseded": false}
+      """
+    Then the structural validation fails
+    And the structural errors include "'sequence' must be an integer"
+
+  # -- structured output validator --
+
+  Scenario: Structured output with a blank command
+    When I structurally validate "structured_output" with the JSON payload
+      """
+      {"command": "", "session_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "status": "ok"}
+      """
+    Then the structural validation fails
+    And the structural errors include "'command' must be a non-empty string"
+
+  Scenario: Structured output with an invalid session_id
+    When I structurally validate "structured_output" with the JSON payload
+      """
+      {"command": "run", "session_id": "bad", "status": "ok"}
+      """
+    Then the structural validation fails
+    And the structural errors include "'session_id' must be a valid ULID"
+
+  Scenario: Structured output with a non-integer exit_code
+    When I structurally validate "structured_output" with the JSON payload
+      """
+      {"command": "run", "session_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "status": "ok", "exit_code": "x"}
+      """
+    Then the structural validation fails
+    And the structural errors include "'exit_code' must be a non-negative integer"
+
+  Scenario: Structured output with elements that are not a list
+    When I structurally validate "structured_output" with the JSON payload
+      """
+      {"command": "run", "session_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "status": "ok", "elements": "nope"}
+      """
+    Then the structural validation fails
+    And the structural errors include "'elements' must be a list"
+
+  Scenario: Structured output with a non-dict element
+    When I structurally validate "structured_output" with the JSON payload
+      """
+      {"command": "run", "session_id": "01ARZ3NDEKTSV4XXFFJFRC889A", "status": "ok", "elements": [123]}
+      """
+    Then the structural validation fails
+    And the structural errors include "is not a dict"
+
+  # -- dispatcher --
+
+  Scenario: Unknown target type raises a ValidationError
+    When I structurally validate the unknown target type "totally_unknown"
+    Then a structural ValidationError is raised

features/steps/coverage_threshold_config_steps.py

@@ -81,20 +81,37 @@ def step_coverage_branch_enabled(context: Context) -> None:
         raise AssertionError("branch = true not found in coverage config")
 
 
+def _pyproject_fail_under() -> float | None:
+    """The canonical coverage floor: pyproject ``[tool.coverage.report]``
+    ``fail_under`` (the single source the noxfile + worker read). None if
+    absent/unreadable."""
+    import tomllib
+
+    toml_path = Path(__file__).resolve().parent.parent.parent / "pyproject.toml"
+    try:
+        data = tomllib.loads(toml_path.read_text(encoding="utf-8"))
+        return float(data["tool"]["coverage"]["report"]["fail_under"])
+    except (OSError, KeyError, TypeError, ValueError, tomllib.TOMLDecodeError):
+        return None
+
+
 @then("the noxfile should contain a fail-under threshold of at least {threshold}")
 def step_noxfile_fail_under(context: Context, threshold: str) -> None:
-    """Assert noxfile has fail-under >= given threshold.
+    """Assert noxfile enforces a fail-under >= the given threshold.
 
-    Supports both literal ``--fail-under=96.5`` and f-string
-    ``f"--fail-under={COVERAGE_THRESHOLD}"`` patterns. When the
-    f-string form is found, the COVERAGE_THRESHOLD constant value
-    is resolved from the source.
+    Handles three forms, newest first:
+    - the noxfile delegates ``COVERAGE_THRESHOLD`` to pyproject's single
+      source (``COVERAGE_THRESHOLD = _read_coverage_fail_under()`` +
+      ``f"--fail-under={COVERAGE_THRESHOLD:g}"``) — resolve the effective floor
+      from pyproject ``[tool.coverage.report].fail_under`` (WS5);
+    - a literal ``--fail-under=96.5``;
+    - an f-string over a ``COVERAGE_THRESHOLD`` *constant*.
     """
     import ast
 
     threshold_val = float(threshold)
 
-    # First try literal --fail-under=N
+    # Literal --fail-under=N (the delegated f-string form has no digits).
     matches = re.findall(r"--fail-under=([\d.]+)", context.noxfile_text)
     if matches:
         max_threshold = max(float(m) for m in matches)
@@ -104,7 +121,8 @@ def step_noxfile_fail_under(context: Context, threshold: str) -> None:
             )
         return
 
-    # Fall back: check for f-string referencing COVERAGE_THRESHOLD constant
+    # f-string over COVERAGE_THRESHOLD: resolve a literal constant, else fall
+    # through to the pyproject single source the constant now delegates to.
     if (
         "fail-under" in context.noxfile_text
         and "COVERAGE_THRESHOLD" in context.noxfile_text
@@ -118,13 +136,22 @@ def step_noxfile_fail_under(context: Context, threshold: str) -> None:
                         and target.id == "COVERAGE_THRESHOLD"
                         and isinstance(node.value, ast.Constant)
                     ):
-                        raw = node.value.value
-                        value = float(str(raw))
+                        value = float(str(node.value.value))
                         if value < threshold_val:
                             raise AssertionError(
-                                f"COVERAGE_THRESHOLD={value} is below required {threshold_val}"
+                                f"COVERAGE_THRESHOLD={value} is below required "
+                                f"{threshold_val}"
                             )
                         return
+        # WS5: COVERAGE_THRESHOLD = _read_coverage_fail_under() → pyproject.
+        resolved = _pyproject_fail_under()
+        if resolved is not None:
+            if resolved < threshold_val:
+                raise AssertionError(
+                    f"pyproject fail_under={resolved:g} is below required "
+                    f"{threshold_val}"
+                )
+            return
     raise AssertionError("No --fail-under found in noxfile.py")
 
 

features/steps/coverage_threshold_enforcement_steps.py

@@ -41,6 +41,18 @@ def step_ci_workflow_exists(context: Any) -> None:
     context.ci_file_content = ci_file.read_text(encoding="utf-8")
 
 
+def _pyproject_fail_under() -> float | None:
+    """The canonical coverage floor: pyproject ``[tool.coverage.report]``
+    ``fail_under`` (the single source the noxfile delegates to via
+    ``_read_coverage_fail_under()``). None if absent/unreadable."""
+    pyproject = PROJECT_ROOT / "pyproject.toml"
+    try:
+        data = tomllib.loads(pyproject.read_text(encoding="utf-8"))
+        return float(data["tool"]["coverage"]["report"]["fail_under"])
+    except (OSError, KeyError, TypeError, ValueError, tomllib.TOMLDecodeError):
+        return None
+
+
 @when("I parse the COVERAGE_THRESHOLD constant from noxfile.py")
 def step_parse_threshold(context: Any) -> None:
     content = context.noxfile_content
@@ -55,8 +67,16 @@ def step_parse_threshold(context: Any) -> None:
                     and isinstance(node.value, ast.Constant)
                 ):
                     threshold = node.value.value
+    # WS5: the noxfile now delegates the constant to pyproject's single source
+    # (``COVERAGE_THRESHOLD = _read_coverage_fail_under()``), so the AST value is
+    # a Call, not a Constant. Resolve the effective floor from pyproject.
     if threshold is None:
-        raise ValueError("COVERAGE_THRESHOLD constant not found in noxfile.py")
+        threshold = _pyproject_fail_under()
+    if threshold is None:
+        raise ValueError(
+            "COVERAGE_THRESHOLD not found in noxfile.py and pyproject "
+            "[tool.coverage.report].fail_under is unreadable"
+        )
     context.coverage_threshold = threshold
 
 

features/steps/coverage_validation_error_paths_steps.py

@@ -0,0 +1,73 @@
+"""Behave steps exercising uncovered error-path branches in
+``cleveragents.core.validation``.
+
+Targets the defensive branches in the plan-tree, decision-dict, and
+structured-output validators that ``structural_validation.feature`` does not
+reach (non-dict nodes, duplicate decision_id, non-list children, invalid child
+ULID, missing decision fields, wrong-typed confidence/parent/sequence,
+malformed structured-output fields, unknown dispatcher target). These are pure
+functions -- no database, CLI runner, or mocks required.
+
+Step phrasing is deliberately distinct from ``structural_validation_steps.py``
+to avoid step-definition collisions.
+"""
+
+from __future__ import annotations
+
+import json
+
+from behave import then, when
+from behave.runner import Context
+
+from cleveragents.core.validation import (
+    ValidationError,
+    validate_structured_component_output,
+)
+
+
+@when('I structurally validate "{target_type}" with the JSON payload')
+def step_structural_validate_payload(context: Context, target_type: str) -> None:
+    data = json.loads(context.text)
+    context.cov_validation_error = None
+    try:
+        context.cov_validation_result = validate_structured_component_output(
+            target_type, data
+        )
+    except ValidationError as exc:
+        context.cov_validation_result = None
+        context.cov_validation_error = exc
+
+
+@when('I structurally validate the unknown target type "{target_type}"')
+def step_structural_validate_unknown(context: Context, target_type: str) -> None:
+    context.cov_validation_error = None
+    try:
+        context.cov_validation_result = validate_structured_component_output(
+            target_type, []
+        )
+    except ValidationError as exc:
+        context.cov_validation_result = None
+        context.cov_validation_error = exc
+
+
+@then("the structural validation fails")
+def step_structural_validation_fails(context: Context) -> None:
+    assert context.cov_validation_result is not None, "expected a result dict"
+    assert context.cov_validation_result["valid"] is False, (
+        f"expected invalid, got {context.cov_validation_result}"
+    )
+
+
+@then('the structural errors include "{substring}"')
+def step_structural_errors_include(context: Context, substring: str) -> None:
+    errors = context.cov_validation_result["errors"]
+    assert any(substring in err for err in errors), (
+        f"no error contained {substring!r}; errors={errors}"
+    )
+
+
+@then("a structural ValidationError is raised")
+def step_structural_validation_error_raised(context: Context) -> None:
+    assert isinstance(context.cov_validation_error, ValidationError), (
+        f"expected ValidationError, got {context.cov_validation_error!r}"
+    )

noxfile.py

@@ -1,7 +1,12 @@
 import json
 import os
+import re
 import shutil
+import subprocess
 import sys
+import time
+import tomllib
+from concurrent.futures import ThreadPoolExecutor
 from pathlib import Path
 
 import nox
@@ -47,8 +52,12 @@ def _pabot_parallel_args(posargs: list[str]) -> list[str]:
         return []
     # Integration tests are significantly heavier than unit tests and can
     # become unstable on shared runners when pabot fans out aggressively.
-    # Keep default parallelism conservative (<=2) unless explicitly overridden.
-    pabot_default = min(2, _default_processes())
+    # Cap parallelism at 6: at the prior <=2 the ~47min suite serialized to
+    # ~24min wall on a 32-core runner (2x speedup == 2 lanes); 6 lanes cut
+    # that to ~6-8min while staying well short of the per-core fan-out that
+    # triggers LLM 429 rate-limits / OOM flakes. Override via
+    # TEST_PROCESSES / --processes.
+    pabot_default = min(6, _default_processes())
     return ["--processes", str(pabot_default)]
 
 
@@ -251,9 +260,10 @@ def build(session: nox.Session):
 def integration_tests(session: nox.Session):
     """Run Robot Framework integration tests (parallel via pabot).
 
-    Defaults to conservative parallelism (<=2 processes) to avoid
-    resource pressure in CI. Override via PABOT_PROCESSES or by passing
-    --processes/--processes=N in session arguments.
+    Defaults to parallelism capped at 6 processes (min(6, cpus)) to avoid
+    resource pressure / LLM 429 flakes in CI while still using the runner.
+    Override via TEST_PROCESSES or by passing --processes/--processes=N in
+    session arguments.
     """
     session.install("-e", ".[tests]")
     session.env["CLEVERAGENTS_AUTO_APPLY_MIGRATIONS"] = "true"
@@ -341,10 +351,10 @@ def slow_integration_tests(session: nox.Session):
     """Run Robot Framework slow integration tests (parallel via pabot).
 
     Runs all tests tagged ``slow`` that are excluded from the standard
-    ``integration_tests`` session.  Defaults to conservative parallelism
-    (<=2 processes) to avoid resource pressure in CI.  Override via
-    TEST_PROCESSES or by passing --processes/--processes=N in session
-    arguments.
+    ``integration_tests`` session.  Defaults to parallelism capped at 6
+    processes (min(6, cpus)) to avoid resource pressure / LLM 429 flakes in
+    CI.  Override via TEST_PROCESSES or by passing --processes/--processes=N
+    in session arguments.
     """
     session.install("-e", ".[tests]")
     session.env["CLEVERAGENTS_AUTO_APPLY_MIGRATIONS"] = "true"
@@ -525,23 +535,352 @@ def e2e_tests(session: nox.Session):
     )
 
 
-COVERAGE_THRESHOLD = 96.5  # Temporarily lowered due to many @tdd_expected_fail tests
-# see issues #4183 and #4184
+def _read_coverage_fail_under() -> float:
+    """Single source of truth for the coverage floor (plan decision 2a):
+    ``pyproject.toml [tool.coverage.report].fail_under``. slipcover does not
+    auto-read pyproject, so this session passes ``--fail-under`` explicitly
+    and reports this value in the summary line. Falls back to 96.5 (the
+    canonical floor, temporarily lowered for the @tdd_expected_fail backlog --
+    issues #4183/#4184) if the key is missing/unreadable.
+
+    Ratchet (decision 2b): the objective is 97% -- raise ``fail_under`` only
+    after observed master coverage has held >= target+buffer for N green
+    commits. The 100%-patch diff-gate keeps total coverage monotonic, so the
+    floor trails real coverage upward; it is not a precondition for shipping.
+    """
+    try:
+        data = tomllib.loads((Path(__file__).parent / "pyproject.toml").read_text())
+        return float(data["tool"]["coverage"]["report"]["fail_under"])
+    except (OSError, KeyError, TypeError, ValueError, tomllib.TOMLDecodeError):
+        return 96.5
+
+
+COVERAGE_THRESHOLD = _read_coverage_fail_under()
+
+# Engine tunables (WS1). ``K`` bounds concurrent slipcover processes; ``N`` is
+# the number of bin-packed chunks, decoupled from K (default ~3xK). ``N`` sets
+# per-chunk peak RSS; ``K`` sets the concurrency multiplier (ceiling ~ K x peak).
+COVERAGE_PROCESSES_DEFAULT = 4
+
+
+_COUNT_SCENARIOS_SCRIPT = """
+import json, sys
+from behave.parser import parse_file
+out = {}
+for fp in sys.argv[2:]:
+    try:
+        f = parse_file(fp)
+        out[fp] = 1 if f is None else max(1, sum(1 for _ in f.walk_scenarios()))
+    except Exception:
+        out[fp] = 1
+with open(sys.argv[1], "w") as fh:
+    json.dump(out, fh)
+"""
+
+
+def _count_scenarios_per_feature(
+    session: nox.Session, feature_paths: list[str]
+) -> dict[str, int]:
+    """Map each ``.feature`` path to its scenario count (outline-expanded).
+
+    Parses each feature with behave's gherkin parser and walks its scenarios
+    (so a ``Scenario Outline`` contributes one unit per ``Examples`` row, the
+    same unit the runner executes). Used as the bin-packing weight so chunks
+    balance by real work, not file count (naive contiguous split measured a
+    1.94 max/mean imbalance -- see plan V10).
+
+    behave lives in the *session* venv, not the nox orchestrator process, so
+    the parse runs via the session python in a subprocess writing JSON to a
+    file (keeps stdout clean). On any failure every file is weighted 1 so none
+    is silently dropped (bin-packing still runs, just by file count).
+    """
+    counts_path = str(Path("build/scenario_counts.json").resolve())
+    proc = subprocess.run(
+        [
+            session.bin + "/python",
+            "-c",
+            _COUNT_SCENARIOS_SCRIPT,
+            counts_path,
+            *feature_paths,
+        ],
+        env=_child_env(session),
+        capture_output=True,
+        text=True,
+    )
+    if proc.returncode != 0:
+        session.log(
+            "scenario count failed; falling back to file-count bins: "
+            f"{proc.stderr.strip()[-300:]}"
+        )
+        return {fp: 1 for fp in feature_paths}
+    with open(counts_path) as fh:
+        return json.load(fh)
+
+
+def _bin_pack(
+    feature_paths: list[str], counts: dict[str, int], n_bins: int
+) -> list[list[str]]:
+    """Greedy largest-first bin-packing into ``n_bins`` least-loaded bins.
+
+    Sorts features by scenario count descending, then drops each into the
+    currently-lightest bin. Returns only non-empty bins (so the chunk count
+    never exceeds the feature count). This keeps per-chunk scenario load
+    balanced, bounding per-process peak RSS and tail wall-clock.
+    """
+    n_bins = max(1, min(n_bins, len(feature_paths)))
+    ordered = sorted(feature_paths, key=lambda p: counts.get(p, 1), reverse=True)
+    bins: list[list[str]] = [[] for _ in range(n_bins)]
+    loads = [0] * n_bins
+    for fp in ordered:
+        i = loads.index(min(loads))
+        bins[i].append(fp)
+        loads[i] += counts.get(fp, 1)
+    return [b for b in bins if b]
+
+
+def _child_env(session: nox.Session) -> dict[str, str]:
+    """Environment for a slipcover child process.
+
+    Inherits the session env (PYTHONPATH, NO_COLOR, BEHAVE_PARALLEL_COVERAGE,
+    CLEVERAGENTS_TEMPLATE_DB) but DEFENSIVELY strips any inherited
+    ``CLEVERAGENTS_DATABASE_URL``: ``before_all`` mktemps a per-process DB
+    only-if-unset, so a fixed value would make all K children share one DB and
+    collide (plan V15). Only the template DB is passed through.
+    """
+    merged = {**os.environ, **session.env}
+    merged.pop("CLEVERAGENTS_DATABASE_URL", None)
+    # session.env may carry None-valued keys (nox unset markers); subprocess
+    # rejects non-str env values, so drop them.
+    env = {k: str(v) for k, v in merged.items() if v is not None}
+    return env
+
+
+_MAXRSS_RE = re.compile(r"Maximum resident set size \(kbytes\):\s*(\d+)")
+
+
+def _run_coverage_chunk(
+    venv_python: str,
+    source_paths: str,
+    omit_patterns: str,
+    behave_cmd: str,
+    chunk: list[str],
+    index: int,
+    env: dict[str, str],
+) -> dict:
+    """Run one slipcover subprocess over ``chunk`` and report its result.
+
+    Wraps the child in ``/usr/bin/time -v`` (when available) to capture peak
+    RSS. Child stdout+stderr are tee'd to ``build/coverage.<i>.log`` so a
+    failing chunk's diagnostics survive without buffering the whole suite's
+    output in memory. Returns a result dict for the caller to validate.
+    """
+    out_json = f"build/coverage.{index}.json"
+    log_path = f"build/coverage.{index}.log"
+    Path(out_json).unlink(missing_ok=True)
+
+    slip_cmd = [
+        venv_python,
+        "-m",
+        "slipcover",
+        "--json",
+        "--out",
+        out_json,
+        "--source",
+        source_paths,
+        "--omit",
+        omit_patterns,
+        "--",
+        behave_cmd,
+        "-q",
+        "--no-capture",
+        *chunk,
+    ]
+    time_bin = "/usr/bin/time"
+    cmd = [time_bin, "-v", *slip_cmd] if os.path.exists(time_bin) else slip_cmd
+
+    start = time.monotonic()
+    with open(log_path, "wb") as log_file:
+        proc = subprocess.Popen(cmd, stdout=log_file, stderr=subprocess.STDOUT, env=env)
+        returncode = proc.wait()
+    wall = time.monotonic() - start
+
+    try:
+        log_text = Path(log_path).read_text(errors="replace")
+    except OSError:
+        log_text = ""
+    rss_match = _MAXRSS_RE.search(log_text)
+    peak_rss_kb = int(rss_match.group(1)) if rss_match else None
+
+    return {
+        "index": index,
+        "returncode": returncode,
+        "out_json": out_json,
+        "log_path": log_path,
+        "wall": wall,
+        "peak_rss_kb": peak_rss_kb,
+        "n_features": len(chunk),
+    }
+
+
+def _validate_coverage_chunk(result: dict) -> str | None:
+    """Three-part liveness check for a chunk. Returns a diagnostic or None.
+
+    A chunk is alive iff: exit code in {0, 1} (0 = pass, 1 = test failures,
+    both still emit coverage) AND its JSON parses AND it recorded a non-empty
+    executed-line set. Anything else (OOM/137, crash, truncated JSON, empty
+    data) is a DEAD chunk -- the caller must fail the whole job rather than
+    merge survivors, which would report false-low coverage.
+    """
+    idx = result["index"]
+    rc = result["returncode"]
+    if rc not in (0, 1):
+        return f"chunk {idx} exited {rc} (e.g. 137 = OOM/kill)"
+
+    out_json = result["out_json"]
+    if not os.path.exists(out_json):
+        return f"chunk {idx} produced no JSON ({out_json} missing)"
+    try:
+        with open(out_json) as f:
+            payload = json.load(f)
+    except (OSError, json.JSONDecodeError) as exc:
+        return f"chunk {idx} JSON did not parse: {exc}"
+
+    files = payload.get("files") or {}
+    executed_total = sum(
+        len(meta.get("executed_lines") or []) for meta in files.values()
+    )
+    if executed_total == 0:
+        return f"chunk {idx} recorded an empty executed-line set"
+    return None
+
+
+def _log_tail(log_path: str, max_lines: int = 40) -> str:
+    try:
+        lines = Path(log_path).read_text(errors="replace").splitlines()
+    except OSError:
+        return "(log unavailable)"
+    return "\n".join(lines[-max_lines:])
+
+
+def _run_parallel_coverage(
+    session: nox.Session,
+    behave_cmd: str,
+    source_paths: str,
+    omit_patterns: str,
+) -> None:
+    """Full-suite coverage via K concurrent slipcover processes over N chunks.
+
+    Enumerates ``features/``, bin-packs by scenario count into N chunks, runs
+    them K-at-a-time, fails loud on any dead chunk, and merges the survivors'
+    JSON into ``build/coverage.json``. The downstream xml/report/summary steps
+    consume the merged JSON unchanged.
+    """
+    # Enumerate features directly rather than importing run_behave_parallel:
+    # that module's top level imports behave/behave_parallel, which live only in
+    # the *session* venv, not the nox orchestrator process -- importing it here
+    # would raise ModuleNotFoundError. This glob mirrors the runner's
+    # ``_iter_features`` directory branch and keeps the engine self-contained.
+    feature_paths = sorted(str(fp) for fp in Path("features").rglob("*.feature"))
+    if not feature_paths:
+        session.error("COVERAGE FAILED: no feature files found under features/")
+
+    k = max(1, int(os.environ.get("COVERAGE_PROCESSES", COVERAGE_PROCESSES_DEFAULT)))
+    n = int(os.environ.get("COVERAGE_CHUNKS", 3 * k))
+    n = max(k, min(n, len(feature_paths)))
+
+    counts = _count_scenarios_per_feature(session, feature_paths)
+    total_scenarios = sum(counts.values())
+    chunks = _bin_pack(feature_paths, counts, n)
+    chunk_loads = [sum(counts.get(fp, 1) for fp in c) for c in chunks]
+    session.log(
+        f"coverage engine: K={k} processes, N={len(chunks)} chunks, "
+        f"{len(feature_paths)} features / {total_scenarios} scenarios; "
+        f"chunk loads min/max={min(chunk_loads)}/{max(chunk_loads)}"
+    )
+
+    venv_python = session.bin + "/python"
+    env = _child_env(session)
+
+    suite_start = time.monotonic()
+    with ThreadPoolExecutor(max_workers=k) as pool:
+        results = list(
+            pool.map(
+                lambda item: _run_coverage_chunk(
+                    venv_python,
+                    source_paths,
+                    omit_patterns,
+                    behave_cmd,
+                    item[1],
+                    item[0],
+                    env,
+                ),
+                enumerate(chunks),
+            )
+        )
+    suite_wall = time.monotonic() - suite_start
+
+    # Per-chunk telemetry (peak RSS + wall) for the equivalence report.
+    for r in sorted(results, key=lambda r: r["index"]):
+        rss = f"{r['peak_rss_kb'] / 1024:.0f}MB" if r["peak_rss_kb"] else "n/a"
+        session.log(
+            f"  chunk {r['index']}: rc={r['returncode']} "
+            f"features={r['n_features']} wall={r['wall']:.0f}s peakRSS={rss}"
+        )
+    peaks = [r["peak_rss_kb"] for r in results if r["peak_rss_kb"]]
+    if peaks:
+        session.log(
+            f"coverage engine wall={suite_wall:.0f}s "
+            f"max per-chunk peakRSS={max(peaks) / 1024:.0f}MB "
+            f"(ceiling ~ K x peak = {k * max(peaks) / 1024:.0f}MB)"
+        )
+
+    # Fail loud on ANY dead chunk -- never merge survivors (a partial merge
+    # reports artificially-low coverage = the exact false-failure we kill).
+    diagnostics = [d for d in (_validate_coverage_chunk(r) for r in results) if d]
+    if diagnostics:
+        detail = "\n".join(diagnostics)
+        worst = next(r for r in results if _validate_coverage_chunk(r) is not None)
+        session.error(
+            "COVERAGE FAILED: dead chunk(s) -- refusing to merge survivors "
+            f"(would report false-low coverage).\n{detail}\n"
+            f"--- tail of {worst['log_path']} ---\n{_log_tail(worst['log_path'])}"
+        )
+
+    # Merge the per-chunk JSON into the canonical coverage.json (verified
+    # correct union -- plan V1). The downstream xml/report steps run on this.
+    merge_inputs = [r["out_json"] for r in sorted(results, key=lambda r: r["index"])]
+    session.run(
+        "python",
+        "-m",
+        "slipcover",
+        "--merge",
+        *merge_inputs,
+        "--json",
+        "--out",
+        "build/coverage.json",
+    )
 
 
 @nox.session(python=DEFAULT_PYTHON, reuse_venv=True, venv_backend="uv")
 def coverage_report(session: nox.Session):
     """Generate coverage report from Behave tests.
 
-    Runs all behave features in a single process under slipcover.
-    The in-process behave-parallel runner avoids subprocess overhead,
-    so a single slipcover invocation collects coverage for the entire
-    suite -- no per-worker files or merge step required.
+    Full-suite runs use a K-way parallel engine: ``features/`` is bin-packed
+    by scenario count into N chunks, each measured by its own short-lived
+    ``slipcover --json`` process (K concurrent), then the per-chunk JSON is
+    merged. This bounds per-process peak RSS (killing the single-process
+    OOM/reaper collapse) and cuts wall-clock, while remaining bit-for-bit
+    equivalent to the single-process measurement. K = env ``COVERAGE_PROCESSES``
+    (default 4); N = env ``COVERAGE_CHUNKS`` (default ~3xK).
 
-    Coverage threshold is enforced at >=96.5%.
+    A *targeted* run (posargs naming ``.feature``) keeps the single-process
+    path, preserving the scenario-specifier recipe.
 
-    On success, emits: COVERAGE OK: <pct>% (threshold: 97%)
-    On failure, emits: COVERAGE FAILED: <pct>% < 97% threshold
+    Coverage threshold is enforced at >= the pyproject
+    ``[tool.coverage.report].fail_under`` floor (plan decision 2a).
+
+    On success, emits: COVERAGE OK: <pct>% (threshold: <floor>%)
+    On failure, emits: COVERAGE FAILED: <pct>% < <floor>% threshold
     Both are single-line, CI-parseable summary strings.
     """
     session.install("-e", ".[tests]")
@@ -571,53 +910,50 @@ def coverage_report(session: nox.Session):
             "*/.nox/*",
             "src/cleveragents/discovery/*",
             "src/cleveragents/tui/materializer.py",
+            # TYPE_CHECKING-only re-export shim: 100% of its "missing" lines are
+            # inside an `if TYPE_CHECKING:` block (never executes at runtime),
+            # and slipcover has no per-line pragma to exclude them. Omit the
+            # whole module so those uncoverable lines stop dragging the total.
+            "src/cleveragents/application/services/__init__.py",
         ]
     )
 
     # Force sequential mode so slipcover wraps the entire process.
     session.env["BEHAVE_PARALLEL_COVERAGE"] = "1"
 
-    # Clean up any existing slipcover data
+    # Clean up any existing slipcover / per-chunk data
     for path in Path("build").glob(".slipcover.*.json"):
         path.unlink()
+    for path in Path("build").glob("coverage.*.json"):
+        path.unlink()
 
-    # Build behave-parallel args (sequential for coverage).
     behave_cmd = session.bin + "/behave-parallel"
     has_feature_files = any(arg.endswith(".feature") for arg in session.posargs)
     if has_feature_files:
-        behave_args = [
+        # Targeted run: a small, scoped selection. Keep the single-process
+        # path -- chunking adds startup overhead for no balance benefit. Allow
+        # exit code 1 (test failures) -- coverage data is still produced.
+        session.run(
+            "python",
+            "-m",
+            "slipcover",
+            "--json",
+            "--out",
+            "build/coverage.json",
+            "--source",
+            source_paths,
+            "--omit",
+            omit_patterns,
+            "--",
             behave_cmd,
             "-q",
             "--no-capture",
             *session.posargs,
-        ]
+            success_codes=[0, 1],
+        )
     else:
-        behave_args = [
-            behave_cmd,
-            "-q",
-            "--no-capture",
-            "features/",
-            *session.posargs,
-        ]
-
-    # Wrap the entire behave-parallel run under slipcover.
-    # A single process produces a single JSON output file directly.
-    # Allow exit code 1 (test failures) — coverage data is still produced.
-    session.run(
-        "python",
-        "-m",
-        "slipcover",
-        "--json",
-        "--out",
-        "build/coverage.json",
-        "--source",
-        source_paths,
-        "--omit",
-        omit_patterns,
-        "--",
-        *behave_args,
-        success_codes=[0, 1],
-    )
+        # Full suite: K-way parallel fan-out -> merged build/coverage.json.
+        _run_parallel_coverage(session, behave_cmd, source_paths, omit_patterns)
 
     # Generate XML report for CI
     session.run(
@@ -700,10 +1036,10 @@ def coverage_report(session: nox.Session):
     rounded_pct = round(total_pct, 1)
 
     if rounded_pct >= COVERAGE_THRESHOLD:
-        session.log(f"COVERAGE OK: {rounded_pct}% (threshold: {COVERAGE_THRESHOLD}%)")
+        session.log(f"COVERAGE OK: {rounded_pct}% (threshold: {COVERAGE_THRESHOLD:g}%)")
     else:
         session.error(
-            f"COVERAGE FAILED: {rounded_pct}% < {COVERAGE_THRESHOLD}% threshold"
+            f"COVERAGE FAILED: {rounded_pct}% < {COVERAGE_THRESHOLD:g}% threshold"
         )
 
 

pyproject.toml

@@ -202,9 +202,22 @@ omit = [
     "*/.nox/*",
     "src/cleveragents/discovery/*",
     "src/cleveragents/tui/materializer.py",
+    # TYPE_CHECKING-only re-export shim (every miss is in an `if TYPE_CHECKING:`
+    # block, never executes at runtime; slipcover has no per-line pragma).
+    "src/cleveragents/application/services/__init__.py",
 ]
 data_file = "build/.coverage"
 
+[tool.coverage.report]
+# Single source of truth for the coverage floor (plan decision 2a). The
+# nox engine and the worker diff-coverage gate read this via tomllib;
+# slipcover does NOT auto-read pyproject, so callers pass --fail-under.
+# Ratchet rule (decision 2b): objective is 97% — bump fail_under only AFTER
+# observed master coverage has held >= target+buffer for N green commits.
+# The 100%-patch diff-gate keeps total coverage monotonic, so this floor
+# trails real coverage upward; it is not a precondition for shipping.
+fail_under = 96.5
+
 [tool.coverage.html]
 directory = "build/htmlcov"
 

robot/coverage_threshold.robot

@@ -8,11 +8,17 @@ Suite Teardown    Cleanup Test Environment
 
 *** Test Cases ***
 Noxfile Contains Coverage Threshold Constant
-    [Documentation]    Verify COVERAGE_THRESHOLD = 96.5 is defined in noxfile.py
+    [Documentation]    Verify COVERAGE_THRESHOLD resolves the floor (96.5) from
+    ...                pyproject's single source [tool.coverage.report].fail_under
     [Tags]    coverage    config    tdd_issue    tdd_issue_4305
 
+    # WS5: the noxfile delegates the constant to pyproject's single source
+    # (COVERAGE_THRESHOLD = _read_coverage_fail_under()), so assert the
+    # delegation plus the pyproject floor rather than a hardcoded literal.
     ${content}=    Get File    ${WORKSPACE}/noxfile.py
-    Should Contain    ${content}    COVERAGE_THRESHOLD = 96.5
+    Should Contain    ${content}    COVERAGE_THRESHOLD = _read_coverage_fail_under()
+    ${pyproject}=    Get File    ${WORKSPACE}/pyproject.toml
+    Should Contain    ${pyproject}    fail_under = 96.5
 
 Pyproject Contains Coverage Run Section
     [Documentation]    Verify [tool.coverage.run] section exists in pyproject.toml