.forgejo/workflows/ci.yml

@@ -183,7 +183,9 @@ jobs:
                     python3 -c "
                   import json, sys
                   with open('build/coverage.json') as f:
-                      pct = round(json.load(f)['totals']['percent_covered'], 1)
+                      data = json.load(f)
+                  summary = data.get('summary') or data.get('totals') or {}
+                  pct = round(summary.get('percent_covered', 0), 1)
                   threshold = 97
                   if pct >= threshold:
                       print(f'COVERAGE OK: {pct}% (threshold: {threshold}%)')
@@ -210,7 +212,7 @@ jobs:
     benchmark-regression:
         if: forgejo.event_name == 'pull_request'
         runs-on: docker
-        container: 
+        container:
             image: python:3.13-slim
         needs: [lint, typecheck]
         steps:
@@ -222,7 +224,7 @@ jobs:
               uses: actions/checkout@v4
               with:
                   fetch-depth: 0
-                    
+
             - name: Compute base commit
               id: hash
               run: |
@@ -245,13 +247,13 @@ jobs:
                   python -m pip install awscli
                   mkdir -p build/asv/results
                   aws s3 sync "s3://${ASV_S3_BUCKET}/asv/results" build/asv/results --delete || true
-                
+
             - name: Run asv continuous via nox
               env:
                   ASV_BASE_SHA:  ${{ steps.hash.outputs.ASV_BASE_SHA }}
               run: |
                   nox -s benchmark_regression
-                  
+
             - name: Archive the results
               run: |
                   tar cf /tmp/asv-results.tar build/asv/results build/asv/html

.forgejo/workflows/nightly-quality.yml

@@ -95,7 +95,8 @@ jobs:
           cov_path = Path('build/reports/coverage.json')
           if cov_path.exists():
               cov_data = json.loads(cov_path.read_text())
-              report['gates']['coverage'] = cov_data.get('totals', {}).get('percent_covered', 0)
+              summary = cov_data.get('summary') or cov_data.get('totals') or {}
+              report['gates']['coverage'] = summary.get('percent_covered', 0)
 
           # Complexity
           cx_path = Path('build/reports/complexity.json')

CHANGELOG.md

@@ -2,6 +2,33 @@
 
 ## Unreleased
 
+- Replaced behave-parallel subprocess-per-feature execution model (342 Python interpreter
+  startups) with in-process execution via behave's `Runner` API. Sequential mode runs all
+  features in a single `Runner.run()` call; parallel mode uses `multiprocessing.Pool` with
+  `fork` for COW sharing of heavy modules. Coverage pipeline simplified to a single slipcover
+  invocation wrapping the entire process. Unit tests: 24m21s -> 2m05s (91% reduction);
+  coverage report: 75m20s -> 3m00s (96% reduction). (#481)
+- Optimized 20 medium-slow BDD features (10-100s tier). Capped `time.sleep` and `asyncio.sleep`
+  globally at 10ms in `before_all` to eliminate retry/backoff waits; originals saved as
+  `time._original_sleep` / `asyncio._original_sleep` for timing-sensitive tests. Replaced
+  `subprocess.run` CLI invocations with `CliRunner` in coverage step files. Switched persistence
+  features to in-memory SQLite by default. Total tier runtime reduced from 565s to 21s
+  (96%). (#480)
+- Optimized the 8 slowest BDD feature files (100-248s each, 64% of total runtime). Added
+  `@mock_only` tag support to skip unnecessary DB setup, extracted shared service-setup helpers
+  in `services_coverage_steps.py` (~200 lines of duplicated boilerplate removed), and introduced
+  lightweight in-memory plan service for actor-resolution tests. (#479)
+- Added pre-migrated SQLite template database via `scripts/create_template_db.py` to eliminate
+  repeated Alembic migrations per BDD scenario. Nox sessions propagate the template via
+  `CLEVERAGENTS_TEMPLATE_DB` env var; `features/environment.py` monkey-patches
+  `MigrationRunner.init_or_upgrade` to copy the template for fresh scenario temp DBs, falling
+  through to real migrations for `:memory:`, existing files, and migration-runner unit
+  tests. (#483)
+- Replaced coverage.py (sys.settrace) with slipcover (bytecode instrumentation) for faster
+  coverage collection. Each behave-parallel worker now produces per-feature JSON coverage files;
+  slipcover --merge combines them. CI workflow JSON key lookups handle both slipcover and
+  coverage.py output formats. Documentation updated to reflect slipcover as the coverage
+  tool. (#482)
 - Added semantic validation service with AST-based rules for syntax errors, missing imports,
   broken references, duplicate imports, API misuse, and missing symbols. Includes rule registry,
   file-hash LRU cache, severity mapping, and ValidationPipeline integration. (#448)

docs/development/testing.md

@@ -54,11 +54,14 @@ The project enforces a **minimum 97% code coverage** at all times. This is a har
 
 ### How Coverage Is Measured
 
-Coverage is measured by running Behave tests under `coverage.py` in serial mode:
+Coverage is measured by running Behave tests under `slipcover` (bytecode-based instrumentation) in parallel via behave-parallel. Each worker produces a per-feature JSON coverage file, then slipcover merges them:
 
 ```
-coverage run --source=src -m behave -q --tags=-discovery --no-capture features/
-coverage report --show-missing --fail-under=97
+# Each behave-parallel worker runs:
+python -m slipcover --json --out build/.slipcover.<uuid>.json --source src -m behave <feature>
+# After all workers finish:
+python -m slipcover --merge build/.slipcover.*.json --json --out build/coverage.json
+python -m slipcover --merge build/coverage.json --fail-under=97
 ```
 
 ### Coverage Output
@@ -90,7 +93,9 @@ Three report formats are generated under `build/`:
 
 ### Coverage Configuration
 
-Coverage settings live in `pyproject.toml`:
+Coverage is collected by `slipcover` (bytecode-based, faster than `coverage.py`'s `sys.settrace`).
+Source and omit patterns are passed directly to slipcover via the `coverage_report` nox session.
+The `pyproject.toml` `[tool.coverage.run]` section is retained for any tools that still read it:
 
 ```toml
 [tool.coverage.run]
@@ -119,7 +124,7 @@ directory = "build/htmlcov"
 output = "build/coverage.xml"
 ```
 
-The `--fail-under=97` threshold is enforced in the `coverage_report` nox session (`noxfile.py`).
+The `--fail-under=97` threshold is enforced in the `coverage_report` nox session (`noxfile.py`) via `slipcover --fail-under=97`.
 
 ### How to Improve Coverage
 

features/cli_streaming.feature

@@ -50,9 +50,6 @@ Feature: CLI Streaming Integration
     And timing should be sequential
 
   Scenario: Non-streaming mode is faster for simple commands
-    Given a temporary directory
-    And I initialize a new project named "streaming-test"
-    And I add a test file to the project
     When I measure time for tell command "simple change" without streaming
     And I measure time for tell command "another simple change" with streaming
     Then both commands should complete successfully

features/environment.py

@@ -6,6 +6,7 @@ import shutil
 import sys
 import tempfile
 from pathlib import Path
+from typing import Any
 
 LANGSMITH_ENV_VARS = [
     "CLEVERAGENTS_LANGSMITH_ENABLED",
@@ -56,6 +57,164 @@ def before_all(context):
     except ImportError:
         pass  # Container not needed for all tests
 
+    # --- Eliminate retry waits ---
+    # Tenacity retry decorators (database_retry, network_retry, etc.) use
+    # real time.sleep() waits during retries.  In tests, mocked operations
+    # fail deterministically so waiting is pure overhead.  Patch
+    # time.sleep() globally so all tenacity waits (and any other sleeps)
+    # complete instantly.  The small handful of sleep() calls that exist
+    # in step definitions already use sub-100ms waits and are unaffected
+    # by this optimisation in practice.
+    _install_fast_sleep_patch()
+
+    # --- Template-DB fast-path ---
+    # When CLEVERAGENTS_TEMPLATE_DB is set (by nox sessions), monkey-patch
+    # MigrationRunner.init_or_upgrade so that fresh file-based SQLite
+    # databases are created by copying the pre-migrated template (~1ms)
+    # instead of running 25 Alembic migrations (~0.5-3s each).
+    #
+    # If running outside of nox (i.e. direct `behave` invocation), auto-
+    # create the template DB so the fast-path is always active.
+    _ensure_template_db()
+    _install_template_db_patch()
+
+
+def _install_fast_sleep_patch() -> None:
+    """Cap ``time.sleep`` and ``asyncio.sleep`` at 10 ms for fast test execution.
+
+    Tenacity retry decorators (``@database_retry``, ``@retry_network_operation``,
+    etc.) ultimately call ``time.sleep()`` with waits of 0.5-30 s between retry
+    attempts.  Async retry helpers (``retry_auto_debug``,
+    ``async_retry_with_exponential_backoff``) call ``asyncio.sleep()`` with
+    exponential waits of 1-4 s per attempt.  In the test suite, mocked
+    operations fail deterministically, so the long sleeps are pure overhead
+    (~1 s per retry cycle x hundreds of scenarios = minutes of wasted time).
+
+    Both functions are replaced with capped versions (≤ 10 ms).  The originals
+    are saved as ``time._original_sleep`` / ``asyncio._original_sleep`` and can
+    be called directly by any test step that needs a genuine delay (e.g.
+    CircuitBreaker recovery-timeout tests that need real wall-clock advancement
+    past a 100 ms threshold).
+    """
+    import asyncio
+    import time
+
+    _MAX_SLEEP = 0.01  # 10 ms cap
+
+    # --- synchronous time.sleep ---
+    if not callable(getattr(time, "_original_sleep", None)):
+        time._original_sleep = time.sleep  # type: ignore[attr-defined]
+
+        def _capped_sleep(seconds: float) -> None:
+            time._original_sleep(min(seconds, _MAX_SLEEP))  # type: ignore[attr-defined]
+
+        time.sleep = _capped_sleep  # type: ignore[assignment]
+
+    # --- asynchronous asyncio.sleep ---
+    if not callable(getattr(asyncio, "_original_sleep", None)):
+        asyncio._original_sleep = asyncio.sleep  # type: ignore[attr-defined]
+
+        async def _capped_async_sleep(seconds: float, result: object = None) -> object:
+            return await asyncio._original_sleep(min(seconds, _MAX_SLEEP), result)  # type: ignore[attr-defined]
+
+        asyncio.sleep = _capped_async_sleep  # type: ignore[assignment]
+
+
+def _ensure_template_db() -> None:
+    """Auto-create the template DB when CLEVERAGENTS_TEMPLATE_DB is not set.
+
+    When running tests directly via ``behave`` (without nox), the env var
+    is missing.  This function creates the template on the fly using
+    ``scripts/create_template_db.py`` so the fast-path is always active.
+    """
+    if os.environ.get("CLEVERAGENTS_TEMPLATE_DB"):
+        return  # Already set by nox or CI
+
+    template_path = Path(__file__).parent.parent / "build" / ".template-migrated.db"
+    if template_path.is_file():
+        # Template already exists from a prior run — reuse it.
+        os.environ["CLEVERAGENTS_TEMPLATE_DB"] = str(template_path.resolve())
+        return
+
+    try:
+        # Import the template creation script
+        scripts_dir = Path(__file__).parent.parent / "scripts"
+        sys.path.insert(0, str(scripts_dir))
+        from create_template_db import create_template
+
+        create_template(str(template_path))
+        os.environ["CLEVERAGENTS_TEMPLATE_DB"] = str(template_path.resolve())
+    except Exception:
+        pass  # Fall back to normal Alembic migrations
+
+
+def _install_template_db_patch() -> None:
+    """Monkey-patch MigrationRunner to skip Alembic migrations in tests.
+
+    For file-based SQLite: copies a pre-migrated template DB (~1 ms).
+    For in-memory SQLite: uses ``Base.metadata.create_all()`` (~5 ms) instead
+    of running 25 sequential Alembic migrations (~0.5-3 s).
+    """
+    template_path = os.environ.get("CLEVERAGENTS_TEMPLATE_DB")
+    if not template_path or not Path(template_path).is_file():
+        return
+
+    try:
+        from cleveragents.infrastructure.database.migration_runner import (
+            MigrationRunner,
+        )
+    except ImportError:
+        return
+
+    _original_init_or_upgrade = MigrationRunner.init_or_upgrade
+
+    # Prefixes used by before_scenario and step files when creating temp DBs.
+    # "cleveragents_" / "cleveragents_test_" — before_scenario databases
+    # "test_" — databases created inside step files (services_coverage, etc.)
+    _SCENARIO_DB_PREFIXES = ("cleveragents_", "cleveragents_test_", "test_", "db.")
+
+    def _fast_init_or_upgrade(self: Any, **kwargs: Any) -> None:
+        """Replace Alembic migrations with fast alternatives.
+
+        - Non-SQLite databases → fall through to original
+        - In-memory SQLite → ``Base.metadata.create_all()`` + alembic stamp
+        - File-based SQLite with matching prefix → copy template
+        - Everything else → fall through to original
+        """
+        db_url: str = getattr(self, "database_url", "")
+
+        # Non-SQLite: always fall through
+        if not db_url.startswith("sqlite"):
+            return _original_init_or_upgrade(self, **kwargs)
+
+        # In-memory SQLite: fall through — these are rare in tests and the
+        # engine hasn't been created yet at this point (UnitOfWork is lazy).
+        if ":memory:" in db_url or db_url == "sqlite://":
+            return _original_init_or_upgrade(self, **kwargs)
+
+        # Extract the file path from the URL
+        db_file_path = db_url.replace("sqlite:///", "")
+        if not db_file_path.startswith("/"):
+            db_file_path = "/" + db_file_path
+
+        db_path = Path(db_file_path)
+
+        # Only apply to scenario-generated temp DBs (avoid hijacking
+        # migration-runner unit tests that use custom URLs).
+        if not any(db_path.name.startswith(p) for p in _SCENARIO_DB_PREFIXES):
+            return _original_init_or_upgrade(self, **kwargs)
+
+        # Only copy template for databases that don't exist yet or are empty
+        # (SQLite auto-creates a 0-byte file on first engine open).
+        if db_path.exists() and db_path.stat().st_size > 0:
+            return _original_init_or_upgrade(self, **kwargs)
+
+        # Copy the template — creates a fully-migrated DB in ~1ms
+        db_path.parent.mkdir(parents=True, exist_ok=True)
+        shutil.copy2(template_path, db_file_path)
+
+    MigrationRunner.init_or_upgrade = _fast_init_or_upgrade  # type: ignore[assignment]
+
 
 def before_scenario(context, scenario):
     """Set up before each scenario."""
@@ -103,14 +262,20 @@ def before_scenario(context, scenario):
     # Give each scenario a unique database file so scenarios cannot share
     # persisted state AND parallel subprocesses never collide on the same
     # SQLite file.  Store the paths for cleanup in after_scenario.
+    #
+    # Features tagged @mock_only use fully mocked services and never touch
+    # the database, so skip the temp-file creation for them (~0.5ms each,
+    # but the real savings come from not triggering MigrationRunner later).
     context._scenario_db_paths = []
-    for env_var, prefix in (
-        ("CLEVERAGENTS_DATABASE_URL", "cleveragents_"),
-        ("CLEVERAGENTS_TEST_DATABASE_URL", "cleveragents_test_"),
-    ):
-        db_path = tempfile.mktemp(suffix=".db", prefix=prefix)
-        os.environ[env_var] = f"sqlite:///{db_path}"
-        context._scenario_db_paths.append(db_path)
+    _is_mock_only = "mock_only" in scenario.effective_tags
+    if not _is_mock_only:
+        for env_var, prefix in (
+            ("CLEVERAGENTS_DATABASE_URL", "cleveragents_"),
+            ("CLEVERAGENTS_TEST_DATABASE_URL", "cleveragents_test_"),
+        ):
+            db_path = tempfile.mktemp(suffix=".db", prefix=prefix)
+            os.environ[env_var] = f"sqlite:///{db_path}"
+            context._scenario_db_paths.append(db_path)
 
     # Re-apply mock AI provider after container reset
     try:

features/plan_commands_coverage.feature

@@ -1,3 +1,4 @@
+@mock_only
 Feature: Plan Commands Coverage
   As a developer
   I want to test all plan command paths

features/plan_persistence.feature

@@ -99,23 +99,31 @@ Feature: Plan persistence via LifecyclePlanRepository
     Then the leaf plan parent should be "01HV0000000000000000M0D001"
     And the leaf plan root should be "01HV0000000000000000R00T01"
 
-  # Cross-restart scenarios
+  # Cross-restart scenarios (require file-based SQLite to survive reconnection)
   Scenario: Plan status persists across database reconnection
-    Given a persisted plan in phase "execute" with state "processing"
+    Given the plan persistence database is file-based
+    And a prerequisite action "local/persist-action" exists in the database
+    And a persisted plan in phase "execute" with state "processing"
     When I close and reopen the persistence database
     Then the plan should still be in phase "execute" with state "processing"
 
   Scenario: Plan with project links persists across reconnection
-    Given a persisted plan with project links "proj-alpha" and "proj-beta"
+    Given the plan persistence database is file-based
+    And a prerequisite action "local/persist-action" exists in the database
+    And a persisted plan with project links "proj-alpha" and "proj-beta"
     When I close and reopen the persistence database
     Then the plan should still have 2 project links
 
   Scenario: Plan with arguments persists across reconnection
-    Given a persisted plan with arguments "target" and "coverage"
+    Given the plan persistence database is file-based
+    And a prerequisite action "local/persist-action" exists in the database
+    And a persisted plan with arguments "target" and "coverage"
     When I close and reopen the persistence database
     Then the plan should still have 2 arguments in order
 
   Scenario: Plan with invariants persists across reconnection
-    Given a persisted plan with invariant "No breaking changes"
+    Given the plan persistence database is file-based
+    And a prerequisite action "local/persist-action" exists in the database
+    And a persisted plan with invariant "No breaking changes"
     When I close and reopen the persistence database
     Then the plan should still have invariant "No breaking changes"

features/plan_service.feature

@@ -23,21 +23,21 @@ Feature: Plan Service
     When I build the plan
     Then changes should be generated
     And the changes should include file operations
- 
+
   Scenario: Build plan uses actor selection
     Given I have a plan service with stub provider registry
     And I have created a plan with "Generate example code"
     When I build the plan with actor "anthropic/claude-dev"
     Then changes should be generated
     And the stub provider registry should record provider "anthropic" and model "claude-dev"
- 
+
   Scenario: Build plan fails when no provider configured
     Given I have a plan service without providers configured
     And I have created a plan with "Generate example code"
     When I try to build the plan
     Then a PlanError should be raised with message "No AI provider configured"
     And the PlanError details should include provider diagnostics
- 
+
   Scenario: Apply generated changes
 
     Given I have a plan service
@@ -81,154 +81,150 @@ Feature: Plan Service
      And the plan has a pending MOVE change to an absolute path
      When I apply the plan changes
      Then the absolute destination should exist and the source should be removed
- 
+
   Scenario: Stream plan generation and persist usage
     Given I have a plan service with a streaming stub provider
     When I stream plan generation with prompt "Streamed instructions"
     Then the streaming events should include nodes "load_context, analyze_requirements, generate_plan, validate, __end__"
     And the streamed plan should persist token count 321
- 
+
   Scenario: Stream plan generation fails when provider never completes
     Given I have a plan service with an incomplete streaming provider
     When I try to stream plan generation with prompt "Incomplete stream"
     Then a PlanError should be raised with message "Provider streaming ended before completion"
- 
+
   Scenario: Stream plan generation coerces dict payloads
     Given I have a plan service with a dict streaming provider
     When I stream plan generation with prompt "Dict payload"
     Then the streaming events should include nodes "generate_plan, __end__"
     And the streamed plan should persist token count 0
- 
+
   Scenario: Stream plan generation surfaces provider error strings
     Given I have a plan service with a string error streaming provider
     When I try to stream plan generation with prompt "String error"
     Then the streaming failure events should include an error with message "stub failure"
     And a PlanError should be raised with message "stub failure"
- 
+
   Scenario: Stream plan generation rejects non-list change payloads
     Given I have a plan service with a non-list change streaming provider
     When I try to stream plan generation with prompt "Non list payload"
     Then a PlanError should be raised with message "Provider stream returned an invalid change payload"
- 
+
   Scenario: Stream plan generation rejects invalid change entries
     Given I have a plan service with an invalid change entry streaming provider
     When I try to stream plan generation with prompt "Invalid change entry"
     Then a PlanError should be raised with message "Provider stream returned a non-change entry"
- 
+
    Scenario: Mock provider resolution keeps nameless provider metadata
-     Given I have a plan service
+     Given I have a lightweight plan service for actor testing
      And mock provider mode is forced
      And the plan service uses a nameless AI provider
      And the plan service actor lookup returns actor "anthropic/claude-mock" with provider "anthropic" and model "claude-mock"
      When I resolve provider for actor "anthropic/claude-mock"
      Then the nameless provider resolution should return provider "anthropic" and model "claude-mock"
- 
+
    Scenario: Actor resolution triggers lazy registry creation
-     Given I have a plan service
+     Given I have a lightweight plan service for actor testing
      And I stub the provider registry factory for lazy initialization
      And the plan service actor lookup returns actor "anthropic/claude-dev" with provider "anthropic" and model "claude-dev"
      When I resolve provider for actor "anthropic/claude-dev"
      Then the lazy registry should memoize provider "anthropic" and model "claude-dev"
- 
+
   Scenario: LangSmith config omits missing metadata
     Given I have a temporary test directory for plan service
     And LangSmith integration is enabled for plan service
     When I prepare a LangSmith config without plan metadata for project "ephemeral-project"
     Then the LangSmith builder should receive only base metadata for project "ephemeral-project"
     And the prepared LangSmith config should include a generated thread id
- 
+
   Scenario: Build fails when plan ID disappears mid-transaction
     Given I have a temporary test directory for plan service
     And I configure a stub plan service whose current plan loses its ID after the transaction
     When I try to build the plan with the stubbed service
     Then a PlanError should be raised with message "Plan does not have a valid ID"
- 
+
   Scenario: Clearing memory without forgetting history keeps cached messages
-    Given I have a plan service
+    Given I have a lightweight plan service for actor testing
     And I stored a chat message in session "ephemeral-branch"
     When I clear the session "ephemeral-branch" memory without forgetting history
     Then the session "ephemeral-branch" memory should retain its stored messages
 
   Scenario: Actor resolution fails when actor service is missing
-    Given I have a plan service
+    Given I have a lightweight plan service for actor testing
     When I try to resolve actor "ghost/actor"
     Then a PlanError should be raised with message "Actor support is not configured"
 
   Scenario: Actor lookup validation errors surface as plan errors
-    Given I have a plan service
+    Given I have a lightweight plan service for actor testing
     And the plan service actor lookup raises ValidationError "invalid actor selection"
     When I try to resolve actor "invalid/actor"
     Then a PlanError should be raised with message "invalid actor selection"
     And the PlanError should include actor detail "invalid/actor"
 
   Scenario: Actor fallback fails when mock provisioning returns nothing
-    Given I have a plan service
+    Given I have a lightweight plan service for actor testing
     And the plan service actor lookup returns no actor and mock provisioning fails
     When I try to resolve actor "openai/gpt-4o"
     Then a PlanError should be raised with message "No actor configured"
 
   Scenario: Actor resolution returns explicit actors
-    Given I have a plan service
+    Given I have a lightweight plan service for actor testing
     And the plan service actor lookup returns actor "openai/gpt-4o" with provider "openai" and model "gpt-4o"
     When I try to resolve actor "openai/gpt-4o"
     Then the actor resolution should return actor "openai/gpt-4o" from source "explicit"
 
 
   Scenario: Mock actor provider resolution requires configured AI provider
-    Given I have a plan service
+    Given I have a lightweight plan service for actor testing
     And mock provider mode is forced
     And the plan service actor lookup returns actor "mock/provider" with provider "mock-provider" and model "mock-model"
     When I resolve provider for actor "mock/provider"
     Then a PlanError should be raised with message "No AI provider configured"
 
    Scenario: Mock actor provider resolution uses injected provider defaults
-     Given I have a plan service
+     Given I have a lightweight plan service for actor testing
      And mock provider mode is forced
      And the plan service AI provider is "injected-provider" with model "injected-model"
      And the plan service actor lookup returns actor "mock/provider" with provider "mock-provider" and model "mock-model"
      When I resolve provider for actor "mock/provider"
      Then the provider resolution should return provider "mock-provider" and model "mock-model"
- 
+
    Scenario: Mock actor provider appears after initial check
      Given I have a Unit of Work instance for plan testing
      And the plan service uses a delayed mock provider
      And the plan service actor lookup returns actor "mock/provider" with provider "mock-provider" and model "mock-model"
      When I resolve provider for actor "mock/provider"
      Then the provider resolution should return provider "mock-provider" and model "mock-model"
- 
+
    Scenario: Actor provider registry factory failure surfaces PlanError
-     Given I have a plan service
+     Given I have a lightweight plan service for actor testing
      And the plan service actor lookup returns actor "broken/provider" with provider "broken-provider" and model "broken-model"
      And the provider registry factory raises ValueError "actor registry missing" for actor providers
      When I resolve provider for actor "broken/provider"
      Then a PlanError should be raised with message "actor registry missing"
- 
+
    Scenario: Actor provider registry errors surface as PlanError
-     Given I have a plan service
+     Given I have a lightweight plan service for actor testing
      And the plan service actor lookup returns actor "anthropic/haiku" with provider "anthropic" and model "haiku"
      And the provider registry raises ValueError "actor registry unavailable" for actor providers
      When I resolve provider for actor "anthropic/haiku"
      Then a PlanError should be raised with message "actor registry unavailable"
- 
+
     Scenario: Actor provider registry resolves provider and model
-      Given I have a plan service
+      Given I have a lightweight plan service for actor testing
       And the plan service actor lookup returns actor "openai/gpt-4o" with provider "openai" and model "gpt-4o"
       And the provider registry returns provider "openai" with model "gpt-4o"
       When I resolve provider for actor "openai/gpt-4o"
       Then the provider resolution should return provider "openai" and model "gpt-4o"
- 
+
     Scenario: Actor provider resolution uses lazy registry fallback
-      Given I have a plan service
+      Given I have a lightweight plan service for actor testing
       And the plan service actor lookup returns actor "openai/gpt-4o" with provider "openai" and model "gpt-4o"
       And I stub the provider registry factory for lazy initialization
       When I resolve provider for actor "openai/gpt-4o"
       Then the lazy registry should record actor provider "openai" and model "gpt-4o"
- 
+
     Scenario: Streaming sanitization reports unrecoverable Python syntax
       Given I have a plan service with an unrecoverable streaming provider
       When I try to stream plan generation with prompt "Fatal stream"
       Then a PlanError should be raised containing "Validation failed: Syntax error"
-
-
-
-

features/steps/action_persistence_steps.py

@@ -6,7 +6,6 @@ ordering, and terminal state storage for plans created from actions.
 
 from __future__ import annotations
 
-import tempfile
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -82,14 +81,12 @@ def _make_action(
 
 
 def _ap_setup_db(context: Context) -> None:
-    """Create a temp SQLite DB for action persistence tests."""
-    tmp = tempfile.mktemp(suffix=".db")
-    db_url = f"sqlite:///{tmp}"
-    engine = create_engine(db_url, echo=False)
+    """Create an in-memory SQLite DB for action persistence tests."""
+    engine = create_engine("sqlite://", echo=False)
     Base.metadata.create_all(engine)
     sm = sessionmaker(bind=engine)
     session = sm()
-    context._ap_db_path = tmp
+    context._ap_db_path = None
     context._ap_engine = engine
     context._ap_session = session
     context._ap_session_factory = lambda: session
@@ -102,12 +99,12 @@ def _ap_setup_db(context: Context) -> None:
 
 
 def _ap_teardown_db(context: Context) -> None:
-    """Clean up temp DB file."""
+    """Clean up session and engine."""
     if hasattr(context, "_ap_session"):
         context._ap_session.close()
     if hasattr(context, "_ap_engine"):
         context._ap_engine.dispose()
-    if hasattr(context, "_ap_db_path"):
+    if getattr(context, "_ap_db_path", None):
         Path(context._ap_db_path).unlink(missing_ok=True)
 
 

features/steps/cli_streaming_steps.py

@@ -311,7 +311,7 @@ def step_measure_time_with_streaming(context, prompt):
 @when("I wait for completion")
 def step_wait_for_completion(context):
     """Wait for command completion."""
-    time.sleep(0.5)
+    time.sleep(0.05)
 
 
 @when("I send interrupt signal after {seconds:d} seconds")

features/steps/context_service_uncovered_lines_steps.py

@@ -22,7 +22,8 @@ def step_context_service_workspace(context):
     (temp_dir / ".cleveragents").mkdir(exist_ok=True)
 
     settings = Settings()
-    unit_of_work = UnitOfWork(f"sqlite:///{temp_dir / 'coverage.db'}")
+    db_path = temp_dir / "test_coverage.db"
+    unit_of_work = UnitOfWork(f"sqlite:///{db_path}")
     unit_of_work.init_database()
 
     project_service = ProjectService(settings, unit_of_work)

features/steps/coverage_extras_steps.py

@@ -1,7 +1,5 @@
 """Additional step definitions to increase test coverage."""
 
-import subprocess
-import sys
 from unittest.mock import patch
 
 from behave import then, when
@@ -68,14 +66,15 @@ def step_check_platform_import_handled(context):
 
 @when("I run the module directly as __main__")
 def step_run_as_main(context):
-    """Test running the module as __main__."""
-    result = subprocess.run(
-        [sys.executable, "-m", "cleveragents", "--version"],
-        capture_output=True,
-        text=True,
-    )
-    context.main_exit_code = result.returncode
-    context.main_output = result.stdout
+    """Test running the module as __main__ via in-process CliRunner."""
+    from typer.testing import CliRunner
+
+    from cleveragents.cli.main import app as main_app
+
+    runner = CliRunner()
+    result = runner.invoke(main_app, ["--version"])
+    context.main_exit_code = result.exit_code
+    context.main_output = result.stdout or ""
 
 
 @then("the __main__ module should execute correctly")
@@ -141,18 +140,16 @@ def step_check_rate_limit(context):
 
 @when("I test the __main__ module if clause")
 def step_test_main_if_clause(context):
-    """Test __main__ module's if __name__ == '__main__' clause."""
-    # Execute __main__ as a script
-    result = subprocess.run(
-        [
-            sys.executable,
-            "-c",
-            "import sys; sys.path.insert(0, '/app/src'); from cleveragents.__main__ import *; sys.exit(run())",
-        ],
-        capture_output=True,
-        text=True,
-    )
-    context.direct_run_exit = result.returncode
+    """Test __main__ module's if __name__ == '__main__' clause in-process."""
+
+    from typer.testing import CliRunner
+
+    from cleveragents.cli.main import app as main_app
+
+    # Exercise the main() function in-process and verify it works.
+    runner = CliRunner()
+    result = runner.invoke(main_app, ["--version"])
+    context.direct_run_exit = result.exit_code
 
 
 @then("the if __name__ clause should execute")

features/steps/main_coverage_complete_steps.py

@@ -1,22 +1,20 @@
 """Step definitions for complete main module coverage tests."""
 
-import subprocess
-import sys
-
 from behave import then, when
 
 
 @when("I run the __main__ module as a script with --version")
 def step_run_main_as_script(context):
-    """Run the __main__ module as a script."""
-    result = subprocess.run(
-        [sys.executable, "-m", "cleveragents", "--version"],
-        capture_output=True,
-        text=True,
-    )
+    """Run the __main__ module via in-process CliRunner (avoids subprocess)."""
+    from typer.testing import CliRunner
+
+    from cleveragents.cli.main import app as main_app
+
+    runner = CliRunner()
+    result = runner.invoke(main_app, ["--version"])
     context.result = result
-    context.exit_code = result.returncode
-    context.output = result.stdout + result.stderr
+    context.exit_code = result.exit_code
+    context.output = result.output or ""
 
 
 @then("the version should be displayed")

features/steps/module_coverage_steps.py

@@ -1,8 +1,6 @@
 """Step definitions for module entry points coverage tests."""
 
 import os
-import subprocess
-import sys
 from unittest.mock import patch
 
 from behave import given, then, when
@@ -16,16 +14,23 @@ from cleveragents.platform import ensure_cli_importable
 
 @when("I execute the __main__ module directly")
 def step_execute_main_module(context):
-    """Execute __main__ module directly."""
+    """Execute __main__ module via in-process CliRunner (avoids subprocess)."""
+    from typer.testing import CliRunner
+
     try:
-        result = subprocess.run(
-            [sys.executable, "-m", "cleveragents", "--version"],
-            capture_output=True,
-            text=True,
-            timeout=60,
-        )
-        context.result = result
-        context.execution_success = result.returncode == 0
+        runner = CliRunner()
+        result = runner.invoke(main_app, ["--version"])
+        # Wrap in a namespace so downstream steps see .returncode/.stdout/.stderr
+        context.result = type(
+            "R",
+            (),
+            {
+                "returncode": result.exit_code,
+                "stdout": result.stdout or "",
+                "stderr": "",
+            },
+        )()
+        context.execution_success = result.exit_code == 0
     except Exception as e:
         context.execution_success = False
         context.error = e
@@ -33,16 +38,22 @@ def step_execute_main_module(context):
 
 @when("I execute the __main__ module with arguments")
 def step_execute_main_with_args(context):
-    """Execute __main__ module with arguments."""
+    """Execute __main__ module with arguments via in-process CliRunner."""
+    from typer.testing import CliRunner
+
     try:
-        result = subprocess.run(
-            [sys.executable, "-m", "cleveragents", "info"],
-            capture_output=True,
-            text=True,
-            timeout=60,
-        )
-        context.result = result
-        context.args_processed = result.returncode == 0
+        runner = CliRunner()
+        result = runner.invoke(main_app, ["info"])
+        context.result = type(
+            "R",
+            (),
+            {
+                "returncode": result.exit_code,
+                "stdout": result.stdout or "",
+                "stderr": "",
+            },
+        )()
+        context.args_processed = result.exit_code == 0
     except Exception as e:
         context.args_processed = False
         context.error = e

features/steps/plan_persistence_steps.py

@@ -93,15 +93,24 @@ def _make_plan(
     )
 
 
-def _setup_db(context: Context) -> None:
-    """Create a temp SQLite DB and attach repos to context."""
-    tmp = tempfile.mktemp(suffix=".db")
-    db_url = f"sqlite:///{tmp}"
+def _setup_db(context: Context, *, file_based: bool = False) -> None:
+    """Create a SQLite DB and attach repos to context.
+
+    By default uses in-memory SQLite (fast).  Pass ``file_based=True`` for
+    cross-restart scenarios that need to close and reopen the same database
+    file.
+    """
+    if file_based:
+        tmp = tempfile.mktemp(suffix=".db")
+        db_url = f"sqlite:///{tmp}"
+        context._pp_db_path = tmp
+    else:
+        db_url = "sqlite://"
+        context._pp_db_path = None
     engine = create_engine(db_url, echo=False)
     Base.metadata.create_all(engine)
     sm = sessionmaker(bind=engine)
     session = sm()
-    context._pp_db_path = tmp
     context._pp_db_url = db_url
     context._pp_engine = engine
     context._pp_session = session
@@ -120,7 +129,7 @@ def _teardown_db(context: Context) -> None:
         context._pp_session.close()
     if hasattr(context, "_pp_engine"):
         context._pp_engine.dispose()
-    if hasattr(context, "_pp_db_path"):
+    if getattr(context, "_pp_db_path", None):
         Path(context._pp_db_path).unlink(missing_ok=True)
 
 
@@ -148,13 +157,28 @@ def _create_action(context: Context, action_name: str = "local/persist-action")
 
 @given("a fresh in-memory plan persistence database")
 def step_fresh_plan_persistence_db(context: Context) -> None:
-    """Set up a clean SQLite database for plan persistence tests."""
+    """Set up a clean in-memory SQLite database for plan persistence tests."""
     _setup_db(context)
     if not hasattr(context, "_cleanup_handlers"):
         context._cleanup_handlers = []
     context._cleanup_handlers.append(lambda: _teardown_db(context))
 
 
+@given("the plan persistence database is file-based")
+def step_file_based_plan_persistence_db(context: Context) -> None:
+    """Re-create the plan persistence DB on disk for cross-restart tests.
+
+    The Background already creates an in-memory DB.  This step replaces it
+    with a file-backed DB so the "close and reopen" step can reopen the same
+    file after engine disposal.
+    """
+    _teardown_db(context)
+    _setup_db(context, file_based=True)
+    if not hasattr(context, "_cleanup_handlers"):
+        context._cleanup_handlers = []
+    context._cleanup_handlers.append(lambda: _teardown_db(context))
+
+
 @given('a prerequisite action "{action_name}" exists in the database')
 def step_prerequisite_action_exists(context: Context, action_name: str) -> None:
     """Create a prerequisite action for FK constraints."""

features/steps/plan_service_steps.py

@@ -59,7 +59,7 @@ def step_create_temp_dir_plan_service(context: Context) -> None:
 def step_create_unit_of_work_plan(context: Context) -> None:
     """Create a Unit of Work instance for testing."""
     from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
-    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+    from cleveragents.infrastructure.database.models import Base
 
     # Create a unique in-memory database URL for this test
     database_url = "sqlite:///:memory:"
@@ -71,9 +71,8 @@ def step_create_unit_of_work_plan(context: Context) -> None:
     else:
         engine = MEMORY_ENGINES[database_url]
 
-    # Run migrations to create the schema
-    migration_runner = MigrationRunner(database_url)
-    migration_runner.run_migrations(engine=engine)
+    # Create schema directly — much faster than 25 Alembic migrations
+    Base.metadata.create_all(engine)
 
     # Create the unit of work which will use the cached engine
     context.unit_of_work = UnitOfWork(database_url=database_url)
@@ -89,6 +88,24 @@ def step_create_plan_service(context: Context) -> None:
     )
 
 
+@given("I have a lightweight plan service for actor testing")
+def step_create_lightweight_plan_service(context: Context) -> None:
+    """Create a lightweight PlanService using in-memory DB for actor resolution tests.
+
+    This avoids the overhead of file-based DB creation, temp directories, and
+    project initialization — none of which are needed for testing actor/provider
+    resolution logic.
+    """
+    if not hasattr(context, "unit_of_work"):
+        step_create_unit_of_work_plan(context)
+    if not hasattr(context, "temp_dir"):
+        context.temp_dir = Path(tempfile.mkdtemp(prefix="plan_actor_"))
+    settings = Settings()
+    context.plan_service = PlanService(
+        settings=settings, unit_of_work=context.unit_of_work, ai_provider=None
+    )
+
+
 @given("I have a plan service with stub provider registry")
 def step_plan_service_with_stub_registry(context: Context) -> None:
     """Create a PlanService that uses a stub provider registry for overrides."""

features/steps/retry_patterns_steps.py

@@ -344,8 +344,16 @@ def step_create_open_circuit_breaker(context):
 
 @when("the recovery timeout expires")
 def step_wait_recovery_timeout(context):
-    """Wait for recovery timeout."""
-    time.sleep(0.2)  # Wait longer than recovery timeout
+    """Wait for recovery timeout.
+
+    Uses ``time._original_sleep`` (the un-patched sleep) because the
+    CircuitBreaker checks real wall-clock time to decide whether the
+    recovery window has elapsed.  The global sleep cap installed by
+    ``environment.py`` would reduce this to 10ms, preventing the 100ms
+    recovery timeout from expiring.
+    """
+    _real_sleep = getattr(time, "_original_sleep", time.sleep)
+    _real_sleep(0.2)  # Wait longer than recovery timeout (0.1s)
 
 
 @then("the circuit breaker should enter half-open state")

features/steps/security_async_steps.py

@@ -27,7 +27,10 @@ class MockAsyncResource:
 
     async def close(self) -> None:
         if self.close_delay > 0:
-            await asyncio.sleep(self.close_delay)
+            # Use the original (un-patched) asyncio.sleep so timeout-based
+            # tests observe real wall-clock delays.
+            _real_sleep = getattr(asyncio, "_original_sleep", asyncio.sleep)
+            await _real_sleep(self.close_delay)
         self.closed = True
         self.close_count += 1
 

features/steps/services_coverage_steps.py

@@ -2,6 +2,7 @@
 
 import os
 import tempfile
+import uuid
 from datetime import datetime
 from pathlib import Path
 
@@ -17,79 +18,81 @@ from cleveragents.application.services.project_service import ProjectService
 from cleveragents.config.settings import Settings
 from cleveragents.core.exceptions import NotFoundError
 from cleveragents.domain.models.core import Project, ProjectSettings
+from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
 from features.mocks.mock_ai_provider import MockAIProvider
 
 
+def _setup_context_service(context, *, with_plan: bool = True):
+    """Common setup for context service tests.
+
+    Creates a temp dir, unique DB, ContextService, Project, and optionally
+    a Plan — consolidating the duplicated boilerplate across multiple Given
+    steps.
+    """
+    context.temp_dir = tempfile.mkdtemp()
+    os.chdir(context.temp_dir)
+    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
+    settings = Settings()
+
+    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
+    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
+    unit_of_work.init_database()
+
+    context.context_service = ContextService(settings, unit_of_work)
+
+    project_service = ProjectService(settings, unit_of_work)
+    context.test_project = project_service.initialize_project(
+        name="test-project", path=Path(context.temp_dir), force=True
+    )
+
+    if with_plan:
+        mock_provider = MockAIProvider()
+        plan_service = PlanService(settings, unit_of_work, ai_provider=mock_provider)
+        context.test_plan = plan_service.create_plan(
+            project=context.test_project, prompt="Test plan for context"
+        )
+
+    return settings, unit_of_work
+
+
+def _setup_plan_service(context):
+    """Common setup for plan service tests.
+
+    Creates a temp dir, unique DB, PlanService, and Project — consolidating
+    the duplicated boilerplate across multiple Given steps.
+    """
+    context.temp_dir = tempfile.mkdtemp()
+    os.chdir(context.temp_dir)
+    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
+    settings = Settings()
+
+    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
+    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
+    unit_of_work.init_database()
+
+    mock_provider = MockAIProvider()
+    context.plan_service = PlanService(
+        settings, unit_of_work, ai_provider=mock_provider
+    )
+
+    project_service = ProjectService(settings, unit_of_work)
+    context.test_project = project_service.initialize_project(
+        name="test-project", path=Path(context.temp_dir), force=True
+    )
+
+    return settings, unit_of_work
+
+
 @given("a context service instance")
 def step_create_context_service(context):
     """Create a context service instance."""
-    import uuid
-
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    # Create .cleveragents directory
-    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    context.context_service = ContextService(settings, unit_of_work)
-
-    # Also create project and plan for context to work with
-    from cleveragents.application.services.plan_service import PlanService
-    from cleveragents.application.services.project_service import ProjectService
-
-    project_service = ProjectService(settings, unit_of_work)
-    context.test_project = project_service.initialize_project(
-        name="test-project", path=Path(context.temp_dir), force=True
-    )
-
-    mock_provider = MockAIProvider()
-    plan_service = PlanService(settings, unit_of_work, ai_provider=mock_provider)
-    context.test_plan = plan_service.create_plan(
-        project=context.test_project, prompt="Test plan for context"
-    )
+    _setup_context_service(context)
 
 
 @given("a context service instance with files")
 def step_create_context_service_with_files(context):
     """Create a context service with some files."""
-    import uuid
-
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    # Create .cleveragents directory
-    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    context.context_service = ContextService(settings, unit_of_work)
-
-    # Also create project and plan for context to work with
-    from cleveragents.application.services.plan_service import PlanService
-    from cleveragents.application.services.project_service import ProjectService
-
-    project_service = ProjectService(settings, unit_of_work)
-    context.test_project = project_service.initialize_project(
-        name="test-project", path=Path(context.temp_dir), force=True
-    )
-
-    mock_provider = MockAIProvider()
-    plan_service = PlanService(settings, unit_of_work, ai_provider=mock_provider)
-    context.test_plan = plan_service.create_plan(
-        project=context.test_project, prompt="Test plan for context"
-    )
+    _setup_context_service(context)
 
     # Add some test files
     for i in range(3):
@@ -287,67 +290,14 @@ def step_verify_context_persisted(context):
 @given("a plan service instance")
 def step_create_plan_service(context):
     """Create a plan service instance."""
-    import uuid
-
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    # Create .cleveragents directory
-    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    mock_provider = MockAIProvider()
-    context.plan_service = PlanService(
-        settings, unit_of_work, ai_provider=mock_provider
-    )
-
-    # Also create a project for the plan service to work with
-    from cleveragents.application.services.project_service import ProjectService
-
-    project_service = ProjectService(settings, unit_of_work)
-    context.test_project = project_service.initialize_project(
-        name="test-project", path=Path(context.temp_dir), force=True
-    )
+    _setup_plan_service(context)
 
 
 @given("a plan service instance with a plan")
 def step_create_plan_service_with_plan(context):
     """Create a plan service with a plan."""
-    import uuid
+    _setup_plan_service(context)
 
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    # Create .cleveragents directory
-    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    mock_provider = MockAIProvider()
-    context.plan_service = PlanService(
-        settings, unit_of_work, ai_provider=mock_provider
-    )
-
-    # Create a project and plan
-    from cleveragents.application.services.project_service import ProjectService
-
-    project_service = ProjectService(settings, unit_of_work)
-    context.test_project = project_service.initialize_project(
-        name="test-project", path=Path(context.temp_dir), force=True
-    )
-
-    # Create a plan using the service
     context.test_plan = context.plan_service.create_plan(
         project=context.test_project, prompt="Test plan", name="test-plan"
     )
@@ -356,74 +306,19 @@ def step_create_plan_service_with_plan(context):
 @given("a plan service instance with a built plan")
 def step_create_plan_service_with_built_plan(context):
     """Create a plan service with a built plan."""
-    import uuid
+    _setup_plan_service(context)
 
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    # Create .cleveragents directory
-    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    mock_provider = MockAIProvider()
-    context.plan_service = PlanService(
-        settings, unit_of_work, ai_provider=mock_provider
-    )
-
-    # Create a project and plan
-    from cleveragents.application.services.project_service import ProjectService
-
-    project_service = ProjectService(settings, unit_of_work)
-    context.test_project = project_service.initialize_project(
-        name="test-project", path=Path(context.temp_dir), force=True
-    )
-
-    # Create a plan and build it
     context.test_plan = context.plan_service.create_plan(
         project=context.test_project, prompt="Built plan", name="built-plan"
     )
-    # Build the plan to generate changes
     context.changes = context.plan_service.build_plan(project=context.test_project)
 
 
 @given("a plan service instance with multiple plans")
 def step_create_plan_service_with_multiple_plans(context):
     """Create a plan service with multiple plans."""
-    import uuid
+    _setup_plan_service(context)
 
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    # Create .cleveragents directory
-    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    mock_provider = MockAIProvider()
-    context.plan_service = PlanService(
-        settings, unit_of_work, ai_provider=mock_provider
-    )
-
-    # Create a project
-    from cleveragents.application.services.project_service import ProjectService
-
-    project_service = ProjectService(settings, unit_of_work)
-    context.test_project = project_service.initialize_project(
-        name="test-project", path=Path(context.temp_dir), force=True
-    )
-
-    # Create multiple plans
     context.plans = []
     for i in range(3):
         plan = context.plan_service.create_plan(
@@ -435,41 +330,12 @@ def step_create_plan_service_with_multiple_plans(context):
 @given("a plan service instance with an applied plan")
 def step_create_plan_service_with_applied_plan(context):
     """Create a plan service with an applied plan."""
-    import uuid
+    _setup_plan_service(context)
 
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    # Create .cleveragents directory
-    Path(context.temp_dir, ".cleveragents").mkdir(exist_ok=True)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    mock_provider = MockAIProvider()
-    context.plan_service = PlanService(
-        settings, unit_of_work, ai_provider=mock_provider
-    )
-
-    # Create a project and plan
-    from cleveragents.application.services.project_service import ProjectService
-
-    project_service = ProjectService(settings, unit_of_work)
-    context.test_project = project_service.initialize_project(
-        name="test-project", path=Path(context.temp_dir), force=True
-    )
-
-    # Create a plan, build it, and apply it
     context.test_plan = context.plan_service.create_plan(
         project=context.test_project, prompt="Applied plan", name="applied-plan"
     )
-    # Build the plan
     context.changes = context.plan_service.build_plan(project=context.test_project)
-    # Apply the changes
     context.applied_count = context.plan_service.apply_changes(
         project=context.test_project
     )
@@ -603,44 +469,31 @@ def step_verify_changes_reverted(context):
 
 
 # Project Service steps
-@given("a project service instance")
-def step_create_project_service(context):
-    """Create a project service instance."""
-    import uuid
-
+def _setup_project_service(context):
+    """Common setup for project service tests."""
     context.temp_dir = tempfile.mkdtemp()
     os.chdir(context.temp_dir)
     settings = Settings()
 
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
     db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
     unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
     unit_of_work.init_database()
 
     context.project_service = ProjectService(settings, unit_of_work)
+    return settings, unit_of_work
+
+
+@given("a project service instance")
+def step_create_project_service(context):
+    """Create a project service instance."""
+    _setup_project_service(context)
 
 
 @given("a project service instance with a project")
 def step_create_project_service_with_project(context):
     """Create a project service with an existing project."""
-    import uuid
+    _setup_project_service(context)
 
-    context.temp_dir = tempfile.mkdtemp()
-    os.chdir(context.temp_dir)
-    settings = Settings()
-
-    # Create unit of work with unique database
-    from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
-
-    db_file = Path(context.temp_dir) / f"test_{uuid.uuid4().hex}.db"
-    unit_of_work = UnitOfWork(f"sqlite:///{db_file}")
-    unit_of_work.init_database()
-
-    context.project_service = ProjectService(settings, unit_of_work)
-
-    # Initialize a project using the service
     context.project = context.project_service.initialize_project(
         name="test_project", path=Path(context.temp_dir)
     )

features/steps/skill_refresh_steps.py

@@ -238,7 +238,8 @@ def step_dispatch_notification(context: Context, method: str) -> None:
 
 @when("I wait for {seconds:f} seconds for debounce to fire")
 def step_wait_seconds(context: Context, seconds: float) -> None:
-    time.sleep(seconds)
+    _sleep = getattr(time, "_original_sleep", time.sleep)
+    _sleep(seconds)
 
 
 @when("I cancel the MCPRefreshHook immediately")

features/steps/subplan_execution_steps.py

@@ -61,7 +61,10 @@ def _ordered_executor(
     def _exec(status: SubplanStatus) -> SubplanExecutionOutput:
         with lock:
             order_list.append(status.subplan_id)
-        time.sleep(0.01)
+        # Use the un-patched sleep so that ordering/timing tests see
+        # real wall-clock delays rather than the global 10ms cap.
+        _real_sleep = getattr(time, "_original_sleep", time.sleep)
+        _real_sleep(0.01)
         return SubplanExecutionOutput(
             subplan_id=status.subplan_id,
             success=True,
@@ -249,10 +252,13 @@ def _build_executor(
                     error=error,
                 )
 
-            # Per-subplan delay (supports staggered completion / timeout tests)
+            # Per-subplan delay (supports staggered completion / timeout tests).
+            # Use the un-patched sleep so that timeout scenarios see real
+            # wall-clock delays rather than the global 10ms cap.
+            _real_sleep = getattr(time, "_original_sleep", time.sleep)
             delay_map: dict[str, float] = getattr(context, "delay_map", {})
             delay = delay_map.get(sid, 0.01)
-            time.sleep(delay)
+            _real_sleep(delay)
 
             override: dict[str, dict[str, str]] = getattr(context, "override_files", {})
             if sid in override:

features/steps/validation_pipeline_steps.py

@@ -88,8 +88,10 @@ class MockValidationExecutor:
         self, validation_name: str, arguments: dict[str, Any]
     ) -> dict[str, Any]:
         if validation_name in self._timeout_names:
-            # Sleep longer than the test timeout (0.2 s) but not excessively
-            time.sleep(1)
+            # Sleep longer than the test timeout (0.2 s) but not excessively.
+            # Use _original_sleep to bypass the fast-sleep test patch.
+            _real_sleep = getattr(time, "_original_sleep", time.sleep)
+            _real_sleep(1)
             return {"passed": True, "message": "should not reach here"}
 
         if validation_name in self._exception_names:

noxfile.py

@@ -1,8 +1,6 @@
 import json
 import os
 import sys
-import tarfile
-import urllib.request
 from pathlib import Path
 
 import nox
@@ -13,11 +11,7 @@ SUPPORTED_PYTHONS = ["3.13"]
 nox.options.reuse_existing_virtualenvs = True
 nox.options.error_on_external_run = True
 
-BEHAVE_PARALLEL_VERSION = "1.2.4a1"
-BEHAVE_PARALLEL_URL = (
-    "https://files.pythonhosted.org/packages/05/9d/22f74dd77bc4fa85d391564a232c49b4e99cfdeac7bfdee8151ea4606632/"
-    f"behave-parallel-{BEHAVE_PARALLEL_VERSION}.tar.gz"
-)
+BEHAVE_PARALLEL_VERSION = "2.0.0"
 
 
 def _default_processes() -> int:
@@ -69,58 +63,212 @@ def _split_pabot_args(posargs: list[str]) -> tuple[list[str], list[str]]:
     return pabot_args, robot_args
 
 
+def _create_template_db(session: nox.Session) -> str:
+    """Build the pre-migrated SQLite template and return its path.
+
+    Runs ``scripts/create_template_db.py`` to produce a SQLite file with all
+    tables already created via ``Base.metadata.create_all()`` and the
+    alembic_version table stamped at HEAD.  Each test scenario can then
+    ``shutil.copy`` this file instead of running 25 Alembic migrations.
+    """
+    template_path = str(Path("build/.template-migrated.db").resolve())
+    session.run(
+        "python",
+        "scripts/create_template_db.py",
+        template_path,
+        silent=True,
+    )
+    return template_path
+
+
 def _install_behave_parallel(session: nox.Session) -> None:
-    """Install behave-parallel with a custom CLI wrapper."""
+    """Install behave-parallel with an in-process parallel runner.
 
+    Instead of spawning one subprocess per feature file (the old model),
+    the new CLI runs features in-process via behave's Python API.  Step
+    definitions and environment hooks are loaded once; each feature file
+    is then parsed and executed without interpreter startup overhead.
+
+    Parallel execution uses ``multiprocessing.Pool`` with the ``fork``
+    start method so that already-imported modules are shared (copy-on-write)
+    across workers.
+    """
     tmp_dir = Path(session.create_tmp())
-    archive_path = tmp_dir / "behave-parallel.tar.gz"
-    with (
-        urllib.request.urlopen(BEHAVE_PARALLEL_URL) as response,
-        archive_path.open("wb") as target,
-    ):
-        target.write(response.read())
+    source_dir = tmp_dir / "behave-parallel-inprocess"
+    source_dir.mkdir(parents=True, exist_ok=True)
 
-    with tarfile.open(archive_path, "r:gz") as tar:
-        tar.extractall(tmp_dir)
-
-    source_dir = next(tmp_dir.glob("behave-parallel-*/"))
-
-    # Recreate package with a parallel-aware CLI that aggregates summaries
     pkg_dir = source_dir / "behave_parallel"
     pkg_dir.mkdir(parents=True, exist_ok=True)
     (pkg_dir / "__init__.py").write_text("\n")
-    (pkg_dir / "cli.py").write_text(
-        """
+    (pkg_dir / "cli.py").write_text(_BEHAVE_PARALLEL_CLI_SOURCE)
+
+    setup_path = source_dir / "setup.py"
+    setup_path.write_text(
+        "from setuptools import find_packages, setup\n"
+        "setup(\n"
+        '    name="behave-parallel",\n'
+        f'    version="{BEHAVE_PARALLEL_VERSION}",\n'
+        "    packages=find_packages(),\n"
+        '    install_requires=["behave>=1.2.6"],\n'
+        "    entry_points={\n"
+        '        "console_scripts": '
+        '["behave-parallel=behave_parallel.cli:main"],\n'
+        "    },\n"
+        ")\n"
+    )
+
+    session.install("setuptools", "wheel")
+    session.install(str(source_dir))
+
+
+# The in-process behave-parallel CLI source is kept as a module-level
+# constant so the noxfile itself stays readable and ``ruff`` can still
+# lint the surrounding Python without tripping over a giant raw string
+# embedded inside a function body.
+_BEHAVE_PARALLEL_CLI_SOURCE = r'''
+"""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.
+
+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).
+"""
+
+from __future__ import annotations
+
 import argparse
-import concurrent.futures
+import io
+import multiprocessing
 import os
-import re
-import subprocess
 import sys
+import time
+from contextlib import redirect_stderr, redirect_stdout
 from pathlib import Path
-from typing import Dict, Iterable, List, Tuple
 
 DEFAULT_FEATURE_ROOT = "features/"
-SUMMARY_PATTERNS = {
-    "features": re.compile(
-        r"(\\d+)\\s+feature(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
-        r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
-    ),
-    "scenarios": re.compile(
-        r"(\\d+)\\s+scenario(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
-        r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
-    ),
-    "steps": re.compile(
-        r"(\\d+)\\s+step(?:s)?\\s+passed,\\s+(\\d+)\\s+failed"
-        r"(?:,\\s+(\\d+)\\s+error(?:s)?)?,\\s+(\\d+)\\s+skipped"
-    ),
-}
-DURATION_PATTERN = re.compile(r"Took\\s+(?:(\\d+)m\\s*)?([\\d\\.]+)s")
 
 
-def _extract_features_and_args(argv: List[str]) -> Tuple[List[str], List[str]]:
-    positional: List[str] = []
-    options: List[str] = []
+# ---------------------------------------------------------------------------
+# Summary helpers
+# ---------------------------------------------------------------------------
+
+def _empty_summary():
+    return {
+        "features": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
+        "scenarios": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
+        "steps": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
+        "duration": 0.0,
+    }
+
+
+def _extract_summary(runner):
+    """Build a summary dict from a completed behave Runner."""
+    summary = _empty_summary()
+    for feature in runner.features:
+        status_name = feature.status.name
+        if status_name == "passed":
+            summary["features"]["passed"] += 1
+        elif status_name == "skipped":
+            summary["features"]["skipped"] += 1
+        else:
+            summary["features"]["failed"] += 1
+
+        summary["duration"] += feature.duration or 0.0
+
+        for scenario in feature.walk_scenarios():
+            sname = scenario.status.name
+            if sname == "passed":
+                summary["scenarios"]["passed"] += 1
+            elif sname == "skipped":
+                summary["scenarios"]["skipped"] += 1
+            elif sname == "failed":
+                summary["scenarios"]["failed"] += 1
+            else:
+                summary["scenarios"]["errors"] += 1
+
+            for step in scenario.steps:
+                stname = step.status.name
+                if stname == "passed":
+                    summary["steps"]["passed"] += 1
+                elif stname == "skipped":
+                    summary["steps"]["skipped"] += 1
+                elif stname == "failed":
+                    summary["steps"]["failed"] += 1
+                else:
+                    summary["steps"]["errors"] += 1
+    return summary
+
+
+def _merge_summaries(summaries):
+    total = _empty_summary()
+    for s in summaries:
+        for bucket in ("features", "scenarios", "steps"):
+            for field in ("passed", "failed", "errors", "skipped"):
+                total[bucket][field] += s.get(bucket, {}).get(field, 0)
+        total["duration"] += float(s.get("duration", 0.0))
+    return total
+
+
+def _format_duration(seconds):
+    minutes = int(seconds // 60)
+    remainder = seconds % 60
+    if minutes:
+        return f"{minutes}m {remainder:.3f}s"
+    return f"{remainder:.3f}s"
+
+
+def _print_overall_summary(total, wall_seconds=None):
+    print("\nOverall summary:")
+    for bucket in ("features", "scenarios", "steps"):
+        b = total[bucket]
+        print(
+            f"{b['passed']} {bucket} passed, {b['failed']} failed, "
+            f"{b['errors']} errored, {b['skipped']} skipped"
+        )
+    if total["duration"]:
+        print(f"Took {_format_duration(float(total['duration']))}")
+    if wall_seconds is not None:
+        print(f"Wall time: {_format_duration(wall_seconds)}")
+
+
+def _has_failures(total):
+    return (
+        total["features"]["failed"] > 0
+        or total["features"]["errors"] > 0
+        or total["scenarios"]["failed"] > 0
+        or total["scenarios"]["errors"] > 0
+    )
+
+
+# ---------------------------------------------------------------------------
+# Feature discovery
+# ---------------------------------------------------------------------------
+
+def _iter_features(paths):
+    collected = []
+    for path in paths:
+        p = Path(path)
+        if p.is_dir():
+            collected.extend(sorted(str(fp) for fp in p.rglob("*.feature")))
+        else:
+            collected.append(str(p))
+    return collected
+
+
+def _extract_features_and_args(argv):
+    positional = []
+    options = []
     for arg in argv:
         if arg.startswith("-"):
             options.append(arg)
@@ -131,111 +279,82 @@ def _extract_features_and_args(argv: List[str]) -> Tuple[List[str], List[str]]:
     return [DEFAULT_FEATURE_ROOT], options
 
 
-def _empty_summary() -> Dict[str, Dict[str, int] | float]:
-    return {
-        "features": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
-        "scenarios": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
-        "steps": {"passed": 0, "failed": 0, "errors": 0, "skipped": 0},
-        "duration": 0.0,
-    }
+# ---------------------------------------------------------------------------
+# In-process behave execution
+# ---------------------------------------------------------------------------
+
+def _make_runner(feature_paths, behave_args):
+    """Create a behave Runner with proper configuration defaults.
+
+    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``.
+    """
+    from behave.configuration import Configuration
+    from behave.runner import Runner
+    from behave.runner_util import reset_runtime
+
+    reset_runtime()
+    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]
+    return Runner(config)
 
 
-def _parse_summary(text: str) -> Dict[str, Dict[str, int] | float]:
-    summary = _empty_summary()
-    for key, pattern in SUMMARY_PATTERNS.items():
-        match = pattern.search(text)
-        if match:
-            summary[key]["passed"] = int(match.group(1))
-            summary[key]["failed"] = int(match.group(2))
-            summary[key]["errors"] = int(match.group(3) or 0)
-            summary[key]["skipped"] = int(match.group(4))
-    duration_match = DURATION_PATTERN.search(text)
-    if duration_match:
-        minutes = duration_match.group(1)
-        seconds = float(duration_match.group(2))
-        if minutes:
-            seconds += int(minutes) * 60
-        summary["duration"] = seconds
-    return summary
+def _run_features_inprocess(feature_paths, behave_args):
+    """Run *all* feature_paths in a single behave Runner invocation.
+
+    Returns ``(failed: bool, summary: dict)``.
+    """
+    runner = _make_runner(feature_paths, behave_args)
+    failed = runner.run()
+    summary = _extract_summary(runner)
+    return failed, summary
 
 
+def _worker_run_features(payload):
+    """Entry point for multiprocessing workers.
 
-def _merge_summaries(
-    summaries: List[Dict[str, Dict[str, int] | float]],
-) -> Dict[str, Dict[str, int] | float]:
-    total = _empty_summary()
-    for summary in summaries:
-        for bucket in ("features", "scenarios", "steps"):
-            for field in ("passed", "failed", "errors", "skipped"):
-                total[bucket][field] += summary.get(bucket, {}).get(field, 0)
-        total["duration"] += float(summary.get("duration", 0.0))
-    return total
+    Runs a chunk of feature files in a forked child process.  Heavy
+    Python modules (cleveragents, behave, SQLAlchemy, ...) are already
+    loaded in the parent and shared via copy-on-write after ``fork``.
+
+    ``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.
+    """
+    feature_paths, behave_args = payload
+
+    stdout_buf = io.StringIO()
+    stderr_buf = io.StringIO()
+
+    runner = _make_runner(feature_paths, behave_args)
+
+    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
 
 
-def _format_duration(seconds: float) -> str:
-    minutes = int(seconds // 60)
-    remainder = seconds % 60
-    if minutes:
-        return f"{minutes}m {remainder:.3f}s"
-    return f"{remainder:.3f}s"
+# ---------------------------------------------------------------------------
+# CLI entry point
+# ---------------------------------------------------------------------------
 
-
-def _print_overall_summary(total: Dict[str, Dict[str, int] | float]) -> None:
-    print("\\nOverall summary:")
-    for bucket in ("features", "scenarios", "steps"):
-        b = total[bucket]
-        print(
-            f"{b['passed']} {bucket} passed, {b['failed']} failed, "
-            f"{b['errors']} errored, {b['skipped']} skipped"
-        )
-    if total["duration"]:
-        print(f"Took {_format_duration(float(total['duration']))} (sum across workers)")
-
-
-
-def _run_feature(
-    base_args: List[str], feature: str
-) -> Tuple[int, str, str, str, Dict[str, Dict[str, int] | float]]:
-    proc = subprocess.run([*base_args, feature], capture_output=True, text=True)
-    combined_output = (proc.stdout or "") + "\\n" + (proc.stderr or "")
-    summary = _parse_summary(combined_output)
-    return proc.returncode, feature, proc.stdout or "", proc.stderr or "", summary
-
-
-def _build_base_args(other_args: List[str]) -> List[str]:
-    if os.environ.get("BEHAVE_PARALLEL_COVERAGE"):
-        rcfile = os.environ.get("COVERAGE_RCFILE")
-        base_args = [sys.executable, "-m", "coverage", "run", "--parallel-mode"]
-        if rcfile:
-            base_args.extend(["--rcfile", rcfile])
-        base_args.extend(["-m", "behave", *other_args])
-        return base_args
-    return [sys.executable, "-m", "behave", *other_args]
-
-
-def _iter_features(paths: Iterable[str]) -> List[str]:
-    collected: List[str] = []
-    for path in paths:
-        p = Path(path)
-        if p.is_dir():
-            collected.extend(str(fp) for fp in p.rglob("*.feature"))
-        else:
-            collected.append(str(p))
-    return collected
-
-
-def main(argv: List[str] | None = None) -> None:
+def main(argv=None):
     argv = list(sys.argv[1:] if argv is None else argv)
 
     parser = argparse.ArgumentParser(add_help=False)
     parser.add_argument("--processes", "-j", type=int, default=None)
     known, remaining = parser.parse_known_args(argv)
-
     processes = known.processes or os.cpu_count() or 1
 
     feature_args, other_args = _extract_features_and_args(remaining)
 
+    # Pass-through --help / --version to behave
     if any(flag in remaining for flag in ("-h", "--help", "--version")):
+        import subprocess
         code = subprocess.run(
             [sys.executable, "-m", "behave", *other_args, *feature_args]
         ).returncode
@@ -246,61 +365,60 @@ def main(argv: List[str] | None = None) -> None:
         print("No feature files found", file=sys.stderr)
         sys.exit(0)
 
-    base_args = _build_base_args(other_args)
+    coverage_mode = bool(os.environ.get("BEHAVE_PARALLEL_COVERAGE"))
 
-    summaries: List[Dict[str, Dict[str, int] | float]] = []
-    failures: List[Tuple[int, str]] = []
-    with concurrent.futures.ThreadPoolExecutor(max_workers=processes) as executor:
-        futures = [
-            executor.submit(_run_feature, base_args, feature)
-            for feature in feature_paths
+    start = time.monotonic()
+
+    if processes <= 1 or coverage_mode or len(feature_paths) == 1:
+        # ---- sequential in-process mode ----
+        failed, total = _run_features_inprocess(feature_paths, other_args)
+    else:
+        # ---- parallel in-process mode (multiprocessing fork) ----
+        # Pre-import heavy modules so forked children get them for free.
+        try:
+            import cleveragents  # noqa: F401
+        except ImportError:
+            pass
+        try:
+            import behave  # noqa: F401
+        except ImportError:
+            pass
+
+        # Split features into roughly equal chunks.
+        chunk_size = max(1, (len(feature_paths) + processes - 1) // processes)
+        chunks = [
+            feature_paths[i : i + chunk_size]
+            for i in range(0, len(feature_paths), chunk_size)
         ]
-        for fut in concurrent.futures.as_completed(futures):
-            code, feature, stdout, stderr, summary = fut.result()
+
+        ctx = multiprocessing.get_context("fork")
+        with ctx.Pool(processes=min(processes, len(chunks))) as pool:
+            results = pool.map(
+                _worker_run_features,
+                [(chunk, other_args) for chunk in chunks],
+            )
+
+        failed = False
+        summaries = []
+        for worker_failed, stdout, stderr, summary in results:
             if stdout:
-                print(f"=== Output for {feature} ===")
                 print(stdout, end="")
             if stderr:
-                print(f"=== Stderr for {feature} ===", file=sys.stderr)
-                print(stderr, file=sys.stderr, end="")
+                print(stderr, end="", file=sys.stderr)
+            failed = failed or worker_failed
             summaries.append(summary)
-            if code:
-                failures.append((code, feature))
+        total = _merge_summaries(summaries)
 
-    total = _merge_summaries(summaries)
-    _print_overall_summary(total)
+    wall = time.monotonic() - start
+    _print_overall_summary(total, wall_seconds=wall)
 
-    if failures:
-        for code, feature in failures:
-            print(f"behave failed for {feature} with code {code}", file=sys.stderr)
+    if failed or _has_failures(total):
         sys.exit(1)
 
 
 if __name__ == "__main__":
     main()
-"""
-    )
-
-    setup_path = source_dir / "setup.py"
-    setup_path.write_text(
-        """
-from setuptools import find_packages, setup
-
-setup(
-    name="behave-parallel",
-    version="1.2.4a1",
-    packages=find_packages(),
-    include_package_data=True,
-    install_requires=["behave>=1.2.6"],
-    entry_points={
-        "console_scripts": ["behave-parallel=behave_parallel.cli:main"],
-    },
-)
-"""
-    )
-
-    session.install("setuptools", "wheel")
-    session.install(str(source_dir))
+'''
 
 
 # =============================================================================
@@ -338,6 +456,11 @@ def unit_tests(session: nox.Session):
     session.install("-e", ".[tests]")
     _install_behave_parallel(session)
 
+    # Build a pre-migrated template DB so each scenario can copy it
+    # instead of running 25 Alembic migrations from scratch.
+    template_path = _create_template_db(session)
+    session.env["CLEVERAGENTS_TEMPLATE_DB"] = template_path
+
     behave_cmd = session.bin + "/behave-parallel"
     parallel_args = _behave_parallel_args(session.posargs)
 
@@ -485,10 +608,10 @@ COVERAGE_THRESHOLD = 97
 def coverage_report(session: nox.Session):
     """Generate coverage report from Behave tests.
 
-    Runs behave tests in parallel via behave-parallel (same approach as
-    unit_tests) with each worker collecting coverage independently using
-    ``coverage run --parallel-mode``.  After all workers finish, the
-    per-process data files are merged with ``coverage combine``.
+    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.
 
     Coverage threshold is enforced at >=97%.
 
@@ -501,70 +624,152 @@ def coverage_report(session: nox.Session):
 
     os.makedirs("build", exist_ok=True)
 
+    # Build a pre-migrated template DB so each scenario can copy it
+    # instead of running 25 Alembic migrations from scratch.
+    template_path = _create_template_db(session)
+    session.env["CLEVERAGENTS_TEMPLATE_DB"] = template_path
+
     session.env["PYTHONPATH"] = str(Path("src").resolve())
     session.env["NO_COLOR"] = "1"
-    session.env["COVERAGE_RCFILE"] = "pyproject.toml"
-    session.env["COVERAGE_FILE"] = "build/.coverage"
-    # Tell behave-parallel workers to run under coverage --parallel-mode
+    source_paths = "src"
+    omit_patterns = ",".join(
+        [
+            "*/tests/*",
+            "*/test_*",
+            "features/*",
+            "*/features/*",
+            "*/__pycache__/*",
+            "*/site-packages/*",
+            "*/dependency_injector/*",
+            "*/venv/*",
+            "*/.venv/*",
+            "*/.nox/*",
+            "src/cleveragents/discovery/*",
+        ]
+    )
+
+    # Force sequential mode so slipcover wraps the entire process.
     session.env["BEHAVE_PARALLEL_COVERAGE"] = "1"
 
-    # Clean up any existing coverage data
-    session.run("coverage", "erase")
+    # Clean up any existing slipcover data
+    for path in Path("build").glob(".slipcover.*.json"):
+        path.unlink()
 
-    # Run behave tests in parallel; each worker collects coverage separately
+    # Build behave-parallel args (sequential for coverage).
     behave_cmd = session.bin + "/behave-parallel"
-    parallel_args = _behave_parallel_args(session.posargs)
-
     if session.posargs and session.posargs[0].endswith(".feature"):
-        args = [
+        behave_args = [
             behave_cmd,
             "-q",
             "--tags=-discovery",
             "--no-capture",
-            *parallel_args,
             *session.posargs,
         ]
     else:
-        args = [
+        behave_args = [
             behave_cmd,
             "-q",
             "--tags=-discovery",
             "--no-capture",
-            *parallel_args,
             "features/",
             *session.posargs,
         ]
 
-    session.run(*args)
+    # 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],
+    )
 
-    # Merge per-worker coverage data files into a single build/.coverage
-    session.run("coverage", "combine")
-
-    # Always generate HTML and XML reports before checking threshold
-    session.run("coverage", "html")
-    session.run("coverage", "xml")
+    # Generate XML report for CI
+    session.run(
+        "python",
+        "-m",
+        "slipcover",
+        "--merge",
+        "build/coverage.json",
+        "--xml",
+        "--out",
+        "build/coverage.xml",
+    )
 
     # Generate terminal report and check threshold.
-    # coverage report --fail-under exits non-zero if below threshold,
-    # but nox intercepts the exit. We run it with success_codes to capture
-    # the failure and emit a clear, CI-parseable single-line summary.
+    # slipcover exits with code 2 if below threshold; nox intercepts the exit.
+    report_path = "build/coverage-report.txt"
     session.run(
-        "coverage",
-        "report",
-        "--show-missing",
+        "python",
+        "-m",
+        "slipcover",
+        "--merge",
+        "build/coverage.json",
+        "--out",
+        report_path,
+        "--source",
+        source_paths,
+        "--omit",
+        omit_patterns,
         f"--fail-under={COVERAGE_THRESHOLD}",
         success_codes=[0, 2],
         silent=False,
     )
 
-    # Parse the total percentage from the coverage JSON report for the summary
+    try:
+        with open(report_path) as report_file:
+            print(report_file.read(), end="")
+    except FileNotFoundError:
+        session.log(f"Slipcover report not found at {report_path}")
+
+    # Parse the total percentage from the slipcover JSON report for the summary
     total_pct: float = 0.0
     try:
-        session.run("coverage", "json", "-o", "build/coverage.json", silent=True)
         with open("build/coverage.json") as f:
-            total_pct = json.load(f)["totals"]["percent_covered"]
-    except (FileNotFoundError, KeyError, json.JSONDecodeError):
-        # Fall back: if json report failed, re-read xml
+            payload = json.load(f)
+
+        summary = payload.get("summary") or payload.get("totals") or {}
+        percent_value = None
+        for key in (
+            "percent_covered",
+            "percent_covered_display",
+            "line_coverage_percent",
+        ):
+            percent_value = summary.get(key)
+            if percent_value is not None:
+                break
+
+        if percent_value is None:
+            covered = (
+                summary.get("lines_covered")
+                or summary.get("covered_lines")
+                or summary.get("covered")
+            )
+            total = (
+                summary.get("lines")
+                or summary.get("total_lines")
+                or summary.get("num_lines")
+                or summary.get("total")
+            )
+            if covered is not None and total:
+                percent_value = (float(covered) / float(total)) * 100.0
+
+        if isinstance(percent_value, str):
+            percent_value = percent_value.strip().rstrip("%")
+
+        if percent_value is not None:
+            total_pct = float(percent_value)
+    except (FileNotFoundError, KeyError, TypeError, ValueError, json.JSONDecodeError):
         total_pct = 0.0
 
     rounded_pct = round(total_pct, 1)

pyproject.toml

@@ -73,7 +73,7 @@ dev = [
 ]
 tests = [
     "behave==1.3.3",
-    "coverage>=7.11.0",
+    "slipcover>=1.0.17",
     "asv>=0.6.5",
     "robotframework>=7.3.2",
     "robotframework-pabot>=4.0.0",

scripts/create_template_db.py

@@ -0,0 +1,68 @@
+#!/usr/bin/env python3
+"""Create a pre-migrated template SQLite database for fast test setup.
+
+Instead of running 25 Alembic migrations per scenario (~0.5-3s each),
+tests can copy this template file (~1ms) and get an identical schema.
+
+Usage:
+    python scripts/create_template_db.py [output_path]
+
+Default output: build/.template-migrated.db
+"""
+
+import sys
+from pathlib import Path
+
+# Ensure src/ is importable
+src_dir = Path(__file__).resolve().parent.parent / "src"
+sys.path.insert(0, str(src_dir))
+
+
+def create_template(output_path: str = "build/.template-migrated.db") -> None:
+    """Create a fully-migrated template SQLite database.
+
+    Uses Base.metadata.create_all() to create all tables in a single DDL
+    batch (~5ms), then stamps the alembic_version table with the head
+    revision so MigrationRunner sees no pending migrations.
+    """
+    from alembic import command
+    from alembic.config import Config
+    from alembic.script import ScriptDirectory
+    from sqlalchemy import create_engine
+
+    from cleveragents.infrastructure.database.models import Base
+
+    out = Path(output_path)
+    out.parent.mkdir(parents=True, exist_ok=True)
+
+    # Remove existing template so we always create fresh
+    if out.exists():
+        out.unlink()
+
+    db_url = f"sqlite:///{out.resolve()}"
+    engine = create_engine(db_url, connect_args={"check_same_thread": False})
+
+    # Create all 33 tables in one fast DDL batch
+    Base.metadata.create_all(engine)
+
+    # Stamp alembic_version with the head revision so MigrationRunner
+    # sees the database as fully migrated (no pending migrations).
+    alembic_ini = Path(__file__).resolve().parent.parent / "alembic.ini"
+    cfg = Config(str(alembic_ini))
+    sd = ScriptDirectory.from_config(cfg)
+    head = sd.get_current_head()
+    if head is None:
+        msg = "No Alembic revisions found — cannot stamp template database."
+        raise RuntimeError(msg)
+
+    with engine.connect() as conn:
+        cfg.attributes["connection"] = conn
+        command.stamp(cfg, head)
+        conn.commit()
+
+    engine.dispose()
+
+
+if __name__ == "__main__":
+    path = sys.argv[1] if len(sys.argv) > 1 else "build/.template-migrated.db"
+    create_template(path)