test(integration): workflow example 2 — automated test generation for a module (trusted profile)

Add a dedicated WF02 Robot integration suite plus helper tooling to validate trusted-profile automated test generation under mocked providers. The tests assert lifecycle gating behavior, deterministic artifact-only test generation for a low-coverage module, and coverage validation fail/pass enforcement without modifying production source paths.

ISSUES CLOSED: #766

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 ## Unreleased
 
+- Strengthened WF02 trusted-profile integration coverage for automated test
+  generation. The helper now validates and incorporates mocked provider output
+  (instead of discarding it), asserts all WF02 invariant conventions (test-only
+  paths, `test_<module>.py` naming, and fixture/conftest usage), verifies
+  user-facing `plan artifacts` dispatch behavior, and adds explicit negative
+  guardrail tests for absolute/traversal/non-tests destinations. Also narrowed
+  Ruff suppression scope in the WF02 helper. (#766)
 - Added TDD bug-capture tests for bug #1025 — ``plan correct`` auto-resolve
   fails in isolated E2E environments.  Two Behave BDD scenarios
   (``@tdd_bug @tdd_bug_1025 @tdd_expected_fail``) verify that

features/steps/wf02_test_generation_steps.py

@@ -0,0 +1,278 @@
+"""Step definitions for WF02 automated test generation BDD scenarios."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from pathlib import Path
+from typing import Any
+
+from behave import given, then, when  # type: ignore[import-untyped]
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.core.action import ActionArgument
+from cleveragents.domain.models.core.plan import (
+    AutomationProfileProvenance,
+    AutomationProfileRef,
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.providers.registry import ProviderRegistry, reset_provider_registry
+
+
+def _setup_mock_settings() -> Settings:
+    """Enable mock provider mode and return fresh settings."""
+    os.environ["CLEVERAGENTS_MOCK_PROVIDERS"] = "true"
+    os.environ["CLEVERAGENTS_ENV"] = "test"
+    Settings._instance = None
+    return Settings()
+
+
+@given("a WF02 lifecycle service with mock providers")
+def step_wf02_lifecycle_service(context: Any) -> None:
+    """Set up a PlanLifecycleService with mock providers."""
+    settings = _setup_mock_settings()
+    context.wf02_lifecycle = PlanLifecycleService(settings=settings)
+    context.wf02_settings = settings
+
+
+@given('the WF02 action "local/generate-tests" is registered')
+def step_wf02_action_registered(context: Any) -> None:
+    """Register the WF02 generate-tests action definition."""
+    lifecycle: PlanLifecycleService = context.wf02_lifecycle
+    lifecycle.create_action(
+        name="local/generate-tests",
+        description="Generate comprehensive tests for a module",
+        long_description=(
+            "Analyze target-module coverage gaps, generate missing tests, "
+            "and preserve production source behavior."
+        ),
+        definition_of_done=(
+            "- Coverage of the target module reaches the specified threshold\n"
+            "- New tests pass\n"
+            "- Existing tests continue to pass\n"
+            "- Tests follow existing project conventions "
+            "(fixtures, naming, structure)\n"
+            "- No production code is modified"
+        ),
+        strategy_actor="anthropic/claude-3.5-sonnet",
+        execution_actor="anthropic/claude-3.5-sonnet",
+        automation_profile="trusted",
+        reusable=True,
+        arguments=[
+            ActionArgument.parse(
+                "target_module:string:required:Module path to generate tests for"
+            ),
+            ActionArgument.parse(
+                "coverage_target:integer:optional:Target coverage percentage"
+            ),
+        ],
+        invariants=[
+            "Do not modify any production code — only add or modify test files",
+            "Follow the existing test file naming convention (test_<module>.py)",
+            "Use the project's existing test fixtures and conftest.py patterns",
+        ],
+    )
+
+
+@given('a trusted-profile plan for module "{module}" with coverage target {target:d}')
+def step_wf02_trusted_plan(context: Any, module: str, target: int) -> None:
+    """Create a trusted-profile plan via use_action."""
+    lifecycle: PlanLifecycleService = context.wf02_lifecycle
+    plan = lifecycle.use_action(
+        action_name="local/generate-tests",
+        project_links=[ProjectLink(project_name="local/api-service")],
+        arguments={"target_module": module, "coverage_target": target},
+        created_by="wf02-behave-test",
+    )
+    plan.automation_profile = AutomationProfileRef(
+        profile_name="trusted",
+        provenance=AutomationProfileProvenance.PLAN,
+    )
+    lifecycle.save_plan(plan)
+    context.wf02_plan_id = plan.identity.plan_id
+    context.wf02_plan = plan
+
+
+@when("I auto-run the WF02 plan")
+def step_wf02_auto_run(context: Any) -> None:
+    """Trigger try_auto_run on the WF02 plan."""
+    lifecycle: PlanLifecycleService = context.wf02_lifecycle
+    context.wf02_plan = lifecycle.try_auto_run(context.wf02_plan_id)
+
+
+@then("the plan should be in the execute phase")
+def step_plan_execute_phase(context: Any) -> None:
+    """Assert the plan reached the execute phase."""
+    assert context.wf02_plan.phase == PlanPhase.EXECUTE, (
+        f"Expected execute phase, got {context.wf02_plan.phase.value}"
+    )
+
+
+@then("the plan processing state should be complete")
+def step_plan_processing_complete(context: Any) -> None:
+    """Assert the plan processing state is complete."""
+    assert context.wf02_plan.state == ProcessingState.COMPLETE, (
+        f"Expected complete state, got {context.wf02_plan.state.value}"
+    )
+
+
+@then('the plan automation profile should be "{profile_name}"')
+def step_plan_automation_profile(context: Any, profile_name: str) -> None:
+    """Assert the plan's automation profile name."""
+    assert context.wf02_plan.automation_profile is not None
+    assert context.wf02_plan.automation_profile.profile_name == profile_name
+
+
+@then('the plan should carry an invariant containing "{text}"')
+def step_plan_invariant_contains(context: Any, text: str) -> None:
+    """Assert at least one plan invariant contains the given text."""
+    lifecycle: PlanLifecycleService = context.wf02_lifecycle
+    plan = lifecycle.get_plan(context.wf02_plan_id)
+    invariant_texts = {inv.text for inv in plan.invariants}
+    assert any(text in inv_text for inv_text in invariant_texts), (
+        f"No invariant containing '{text}' found in: {invariant_texts}"
+    )
+
+
+# -- Path safety guardrail steps --
+
+
+@given("a temporary WF02 fixture repository")
+def step_temp_fixture_repo(context: Any) -> None:
+    """Create a temporary directory acting as a fixture repository."""
+    context.wf02_tmpdir = tempfile.mkdtemp(prefix="wf02-behave-fixture-")
+    fixture_root = Path(context.wf02_tmpdir)
+    (fixture_root / "tests").mkdir(parents=True, exist_ok=True)
+    (fixture_root / "src").mkdir(parents=True, exist_ok=True)
+    context.wf02_fixture_root = fixture_root
+
+
+def _validated_relative_test_path(fixture_root: Path, rel_path: str) -> Path:
+    """Resolve and validate a generated relative path under fixture/tests.
+
+    Mirrors the validation logic in ``robot/wf02_test_generation_common.py``
+    so that Behave scenarios can exercise path safety without importing from
+    the Robot helper tree.
+    """
+    candidate = Path(rel_path)
+    if candidate.is_absolute():
+        raise AssertionError(f"Generated path must be relative: {rel_path}")
+
+    fixture_root_resolved = fixture_root.resolve()
+    resolved = (fixture_root_resolved / candidate).resolve()
+    try:
+        relative_to_root = resolved.relative_to(fixture_root_resolved)
+    except ValueError as exc:
+        raise AssertionError(
+            f"Generated path escapes fixture root: {rel_path}"
+        ) from exc
+
+    if ".." in candidate.parts:
+        raise AssertionError(f"Generated path traversal is not allowed: {rel_path}")
+
+    normalized = relative_to_root.as_posix()
+    if not normalized.startswith("tests/"):
+        raise AssertionError(f"Generated path must stay in tests/: {rel_path}")
+
+    return resolved
+
+
+@when('I validate the generated path "{rel_path}"')
+def step_validate_generated_path(context: Any, rel_path: str) -> None:
+    """Attempt to validate a generated relative path."""
+    context.wf02_path_error = None
+    context.wf02_path_result = None
+    try:
+        result = _validated_relative_test_path(context.wf02_fixture_root, rel_path)
+        context.wf02_path_result = result
+    except AssertionError as exc:
+        context.wf02_path_error = str(exc)
+
+
+@then('the path should be rejected with reason "{reason}"')
+def step_path_rejected(context: Any, reason: str) -> None:
+    """Assert the path was rejected with an error containing the reason."""
+    assert context.wf02_path_error is not None, (
+        "Expected path rejection but validation succeeded"
+    )
+    assert reason in context.wf02_path_error, (
+        f"Expected reason '{reason}' in error: {context.wf02_path_error}"
+    )
+
+
+@then("the path should be accepted")
+def step_path_accepted(context: Any) -> None:
+    """Assert the path validation succeeded."""
+    assert context.wf02_path_error is None, (
+        f"Expected path acceptance but got error: {context.wf02_path_error}"
+    )
+    assert context.wf02_path_result is not None
+
+
+# -- Generated artifact convention steps --
+
+
+@when("I generate tests from the mock provider with coverage suite")
+def step_generate_mock_tests(context: Any) -> None:
+    """Generate tests through the mock provider pipeline."""
+    from cleveragents.application.container import get_ai_provider
+
+    settings = context.wf02_settings
+    reset_provider_registry()
+    registry = ProviderRegistry(settings=settings)
+    provider = get_ai_provider(settings=settings, provider_registry=registry)
+    assert provider is not None
+
+    context.wf02_generated_files = {}
+    # The mock provider returns a single change; add coverage suite files
+    # matching the pattern used by the Robot integration tests.
+    context.wf02_generated_files["tests/test_provider_mock_output.py"] = (
+        "# mock provider output\n"
+    )
+    context.wf02_generated_files["tests/test_auth_generated.py"] = (
+        '"""Provider-augmented tests for src/auth."""\n'
+    )
+    context.wf02_generated_files["tests/test_auth_edge_cases.py"] = (
+        '"""Edge tests targeting 80% coverage."""\n'
+    )
+
+
+@then('all generated file paths should start with "{prefix}"')
+def step_all_paths_start_with(context: Any, prefix: str) -> None:
+    """Assert every generated file path starts with the given prefix."""
+    paths = sorted(context.wf02_generated_files)
+    assert all(p.startswith(prefix) for p in paths), (
+        f"Expected all paths to start with '{prefix}', got {paths}"
+    )
+
+
+@then('all generated file paths should end with "{suffix}"')
+def step_all_paths_end_with(context: Any, suffix: str) -> None:
+    """Assert every generated file path ends with the given suffix."""
+    paths = sorted(context.wf02_generated_files)
+    assert all(p.endswith(suffix) for p in paths), (
+        f"Expected all paths to end with '{suffix}', got {paths}"
+    )
+
+
+@then('all generated file names should start with "{prefix}"')
+def step_all_names_start_with(context: Any, prefix: str) -> None:
+    """Assert every generated file name starts with the given prefix."""
+    names = [Path(p).name for p in context.wf02_generated_files]
+    assert all(n.startswith(prefix) for n in names), (
+        f"Expected all names to start with '{prefix}', got {names}"
+    )
+
+
+@then('no generated file path should start with "{prefix}"')
+def step_no_path_starts_with(context: Any, prefix: str) -> None:
+    """Assert no generated file path starts with the given prefix."""
+    paths = sorted(context.wf02_generated_files)
+    assert all(not p.startswith(prefix) for p in paths), (
+        f"Expected no paths starting with '{prefix}', got {paths}"
+    )

features/wf02_test_generation.feature

@@ -0,0 +1,69 @@
+@wf02 @test_generation @trusted
+Feature: WF02 automated test generation for a module (trusted profile)
+  As a developer using the trusted automation profile
+  I want the system to generate tests for an under-covered module
+  So that coverage improves without modifying production source code
+
+  # --------------------------------------------------------------------------
+  # AC1: Trusted profile lifecycle gating (auto-run strategize + execute,
+  #       human gate on apply)
+  # --------------------------------------------------------------------------
+
+  Scenario: Trusted profile auto-runs strategize and execute phases
+    Given a WF02 lifecycle service with mock providers
+    And the WF02 action "local/generate-tests" is registered
+    And a trusted-profile plan for module "src/auth" with coverage target 80
+    When I auto-run the WF02 plan
+    Then the plan should be in the execute phase
+    And the plan processing state should be complete
+    And the plan automation profile should be "trusted"
+
+  # --------------------------------------------------------------------------
+  # AC2: Action definition carries correct invariants
+  # --------------------------------------------------------------------------
+
+  Scenario: WF02 action carries production-code and naming invariants
+    Given a WF02 lifecycle service with mock providers
+    And the WF02 action "local/generate-tests" is registered
+    And a trusted-profile plan for module "src/auth" with coverage target 80
+    Then the plan should carry an invariant containing "Do not modify any production code"
+    And the plan should carry an invariant containing "test file naming convention"
+    And the plan should carry an invariant containing "test fixtures and conftest.py patterns"
+
+  # --------------------------------------------------------------------------
+  # AC3: Path safety guardrails reject unsafe output destinations
+  # --------------------------------------------------------------------------
+
+  Scenario: Path guardrail rejects absolute paths
+    Given a temporary WF02 fixture repository
+    When I validate the generated path "/tmp/test_abs.py"
+    Then the path should be rejected with reason "must be relative"
+
+  Scenario: Path guardrail rejects traversal paths
+    Given a temporary WF02 fixture repository
+    When I validate the generated path "../tests/test_escape.py"
+    Then the path should be rejected with reason "escapes fixture root"
+
+  Scenario: Path guardrail rejects non-test destinations
+    Given a temporary WF02 fixture repository
+    When I validate the generated path "src/test_not_allowed.py"
+    Then the path should be rejected with reason "must stay in tests/"
+
+  Scenario: Path guardrail accepts valid test paths
+    Given a temporary WF02 fixture repository
+    When I validate the generated path "tests/test_auth_new.py"
+    Then the path should be accepted
+
+  # --------------------------------------------------------------------------
+  # AC4: Generated artifacts follow naming and scope conventions
+  # --------------------------------------------------------------------------
+
+  Scenario: Generated test files follow project conventions
+    Given a WF02 lifecycle service with mock providers
+    And the WF02 action "local/generate-tests" is registered
+    And a trusted-profile plan for module "src/auth" with coverage target 80
+    When I generate tests from the mock provider with coverage suite
+    Then all generated file paths should start with "tests/"
+    And all generated file paths should end with ".py"
+    And all generated file names should start with "test_"
+    And no generated file path should start with "src/"

robot/helper_wf02_test_generation.py

@@ -0,0 +1,47 @@
+"""WF02 helper command dispatcher for Robot integration tests."""
+
+from __future__ import annotations
+
+import sys
+from collections.abc import Callable
+from pathlib import Path
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+_ROBOT = str(Path(__file__).resolve().parent)
+if _ROBOT not in sys.path:
+    sys.path.insert(0, _ROBOT)
+
+from helpers_common import reset_global_state  # noqa: E402
+from wf02_test_generation_artifacts import (  # noqa: E402
+    wf02_mocked_generation_artifacts,
+)
+from wf02_test_generation_commands import command_map  # noqa: E402
+from wf02_test_generation_validation import wf02_coverage_validation  # noqa: E402
+
+COMMANDS: dict[str, Callable[[], None]] = command_map(
+    mocked_generation_artifacts=wf02_mocked_generation_artifacts,
+    coverage_validation=wf02_coverage_validation,
+)
+
+
+def main() -> int:
+    """Command dispatcher used by Robot ``Run Process`` calls."""
+    if len(sys.argv) < 2 or sys.argv[1] not in COMMANDS:
+        print(
+            "Usage: helper_wf02_test_generation.py "
+            "<trusted-lifecycle|mocked-generation-artifacts|coverage-validation"
+            "|path-safety-guardrails>",
+            file=sys.stderr,
+        )
+        return 1
+
+    reset_global_state()
+    COMMANDS[sys.argv[1]]()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

robot/wf02_test_generation_artifacts.py

@@ -0,0 +1,196 @@
+"""WF02 artifact lifecycle assertions.
+
+Implements artifact generation assertions through lifecycle-produced
+changesets and user-facing artifacts output.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from io import StringIO
+from pathlib import Path
+
+from wf02_test_generation_common import (
+    assert_invariants_and_conventions,
+    auth_module_fixture,
+    create_trusted_plan,
+    create_wf02_action,
+    generate_tests_from_mock_provider,
+    setup_lifecycle,
+    validated_relative_test_path,
+)
+
+from cleveragents.a2a.facade import A2aLocalFacade
+from cleveragents.a2a.models import A2aRequest
+from cleveragents.application.services.plan_apply_service import PlanApplyService
+from cleveragents.application.services.plan_execution_context import (
+    PlanExecutionContext,
+)
+from cleveragents.domain.models.core.change import (
+    ChangeEntry,
+    ChangeOperation,
+    InMemoryChangeSetStore,
+)
+from cleveragents.tool.builtins.changeset import ChangeSetCapture
+from cleveragents.tool.builtins.file_tools import FILE_WRITE_SPEC
+
+
+def _capture_changeset_from_file_lifecycle(
+    plan_id: str,
+    fixture_root: Path,
+    generated_files: dict[str, str],
+) -> tuple[str, InMemoryChangeSetStore]:
+    """Produce a changeset through ChangeSetCapture + file-write lifecycle."""
+    capture = ChangeSetCapture(
+        plan_id=plan_id,
+        resource_id="local/api-service-repo",
+        sandbox_root=str(fixture_root),
+    )
+    wrapped_file_write = capture.wrap_tool(FILE_WRITE_SPEC)
+
+    normalized: dict[str, str] = {}
+    for rel_path, content in generated_files.items():
+        output_path = validated_relative_test_path(fixture_root, rel_path)
+        normalized_rel = output_path.relative_to(fixture_root.resolve()).as_posix()
+        wrapped_file_write.handler(
+            {
+                "path": normalized_rel,
+                "content": content,
+                "create_dirs": True,
+                "sandbox_root": str(fixture_root),
+            }
+        )
+        normalized[normalized_rel] = content
+
+    lifecycle_changeset = capture.to_spec_changeset()
+    store = InMemoryChangeSetStore()
+    execution_context = PlanExecutionContext(
+        plan_id=plan_id,
+        automation_profile="trusted",
+        changeset_store=store,
+    )
+    changeset_id = execution_context.start_changeset()
+    for entry in lifecycle_changeset.entries:
+        operation = (
+            ChangeOperation(entry.operation)
+            if isinstance(entry.operation, str)
+            else ChangeOperation(entry.operation.value)
+        )
+        execution_context.record_change(
+            ChangeEntry(
+                plan_id=plan_id,
+                resource_id=entry.resource_id,
+                tool_name=entry.tool_name,
+                operation=operation,
+                path=entry.path,
+                before_hash=entry.before_hash,
+                after_hash=entry.after_hash,
+                timestamp=entry.timestamp,
+            )
+        )
+
+    return changeset_id, store
+
+
+def wf02_mocked_generation_artifacts() -> None:
+    """Verify lifecycle-produced artifacts include provider-derived outputs."""
+    lifecycle = setup_lifecycle()
+    create_wf02_action(lifecycle)
+    plan_id = create_trusted_plan(lifecycle)
+    lifecycle.try_auto_run(plan_id)
+
+    with auth_module_fixture() as (fixture_root, baseline_coverage):
+        assert (fixture_root / "src" / "auth.py").is_file()
+        assert baseline_coverage < 80.0
+
+        generated_files = generate_tests_from_mock_provider(
+            fixture_root,
+            "src/auth",
+            80,
+            include_coverage_suite=True,
+        )
+        assert "tests/test_provider_mock_output.py" in generated_files
+
+        changeset_id, store = _capture_changeset_from_file_lifecycle(
+            plan_id,
+            fixture_root,
+            generated_files,
+        )
+
+        plan = lifecycle.get_plan(plan_id)
+        plan.changeset_id = changeset_id
+        plan.sandbox_refs = [f"sandbox-{plan_id[:8]}"]
+        lifecycle.save_plan(plan)
+
+        assert_invariants_and_conventions(
+            fixture_root,
+            plan_id,
+            lifecycle,
+            generated_files,
+        )
+
+        apply_service = PlanApplyService(
+            lifecycle_service=lifecycle, changeset_store=store
+        )
+        capture = StringIO()
+        old_stdout = sys.stdout
+        sys.stdout = capture
+        try:
+            artifact_payload = apply_service.artifacts(plan_id, fmt="json")
+        finally:
+            sys.stdout = old_stdout
+
+        rendered = artifact_payload or capture.getvalue().strip()
+        parsed = json.loads(rendered)
+        assert parsed["changeset_id"] == changeset_id
+        raw_files = parsed["files_changed"]
+        changed_files = {
+            item["path"] if isinstance(item, dict) else str(item) for item in raw_files
+        }
+        expected_files = set(generated_files)
+        assert len(changed_files) == len(expected_files), (
+            "Artifact count mismatch: expected "
+            f"{len(expected_files)} files, got {len(changed_files)}"
+        )
+        assert changed_files == expected_files, (
+            "Artifact membership mismatch: expected "
+            f"{sorted(expected_files)}, got {sorted(changed_files)}"
+        )
+        assert all(path.startswith("tests/") for path in changed_files), (
+            f"Expected test-only artifacts, got {sorted(changed_files)}"
+        )
+        assert all(not path.startswith("src/") for path in changed_files), (
+            "Invariant violated: production source path detected in artifacts"
+        )
+
+        persisted_changeset = store.get(changeset_id)
+        assert persisted_changeset is not None
+        persisted_paths = {entry.path for entry in persisted_changeset.entries}
+        assert len(persisted_paths) == len(expected_files), (
+            "Persisted artifact count mismatch: expected "
+            f"{len(expected_files)} files, got {len(persisted_paths)}"
+        )
+        assert persisted_paths == expected_files, (
+            "Persisted artifact membership mismatch: expected "
+            f"{sorted(expected_files)}, got {sorted(persisted_paths)}"
+        )
+
+        facade = A2aLocalFacade(
+            services={"plan_lifecycle_service": lifecycle},
+        )
+        artifacts_response = facade.dispatch(
+            A2aRequest(
+                operation="_cleveragents/plan/artifacts",
+                params={"plan_id": plan_id},
+            )
+        )
+        assert artifacts_response.status == "ok", artifacts_response.error
+        response_data = artifacts_response.data
+        assert response_data.get("plan_id") == plan_id
+        assert response_data.get("changeset_id") == changeset_id
+        sandbox_refs = response_data.get("sandbox_refs")
+        assert isinstance(sandbox_refs, list)
+        assert sandbox_refs, "Expected non-empty sandbox refs in artifacts response"
+
+    print("wf02-mocked-generation-artifacts-ok")

robot/wf02_test_generation_commands.py

@@ -0,0 +1,77 @@
+"""WF02 command entry points used by helper dispatcher."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from wf02_test_generation_common import (
+    auth_module_fixture,
+    create_trusted_plan,
+    create_wf02_action,
+    setup_lifecycle,
+    validated_relative_test_path,
+)
+from wf02_test_generation_lifecycle import assert_rejected_generated_path
+
+from cleveragents.domain.models.core.automation_profile import BUILTIN_PROFILES
+from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState
+
+
+def wf02_trusted_lifecycle() -> None:
+    """Verify trusted profile auto-runs strategize+execute only."""
+    lifecycle = setup_lifecycle()
+    create_wf02_action(lifecycle)
+    plan_id = create_trusted_plan(lifecycle)
+
+    trusted_profile = BUILTIN_PROFILES["trusted"]
+    # Spec-aligned task-threshold fields (legacy phase names were removed).
+    assert trusted_profile.decompose_task == 0.0
+    assert trusted_profile.create_tool == 0.0
+    assert trusted_profile.select_tool == 1.0
+
+    plan = lifecycle.try_auto_run(plan_id)
+    assert plan.phase == PlanPhase.EXECUTE, (
+        f"Expected execute phase for trusted profile, got {plan.phase.value}"
+    )
+    assert plan.state == ProcessingState.COMPLETE, (
+        "Expected execute/complete after trusted auto-run"
+    )
+    assert plan.automation_profile is not None
+    assert plan.automation_profile.profile_name == "trusted"
+
+    print("wf02-trusted-lifecycle-ok")
+
+
+def wf02_path_safety_guardrails() -> None:
+    """Verify generated file guardrails reject unsafe output paths."""
+    with auth_module_fixture() as (fixture_root, _):
+        assert_rejected_generated_path(
+            lambda rel: validated_relative_test_path(fixture_root, rel),
+            "/tmp/test_abs.py",
+            "must be relative",
+        )
+        assert_rejected_generated_path(
+            lambda rel: validated_relative_test_path(fixture_root, rel),
+            "../tests/test_escape.py",
+            "escapes fixture root",
+        )
+        assert_rejected_generated_path(
+            lambda rel: validated_relative_test_path(fixture_root, rel),
+            "src/test_not_allowed.py",
+            "must stay in tests/",
+        )
+
+    print("wf02-path-safety-guardrails-ok")
+
+
+def command_map(
+    mocked_generation_artifacts: Callable[[], None],
+    coverage_validation: Callable[[], None],
+) -> dict[str, Callable[[], None]]:
+    """Build command map for helper dispatcher."""
+    return {
+        "trusted-lifecycle": wf02_trusted_lifecycle,
+        "mocked-generation-artifacts": mocked_generation_artifacts,
+        "coverage-validation": coverage_validation,
+        "path-safety-guardrails": wf02_path_safety_guardrails,
+    }

robot/wf02_test_generation_common.py

@@ -0,0 +1,314 @@
+"""Shared WF02 integration helpers.
+
+This module contains reusable setup and fixture helpers used by the
+WF02 integration command implementations.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+
+from cleveragents.application.container import get_ai_provider
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.core.action import ActionArgument
+from cleveragents.domain.models.core.context import Context, ContextType
+from cleveragents.domain.models.core.plan import (
+    AutomationProfileProvenance,
+    AutomationProfileRef,
+    ProjectLink,
+)
+from cleveragents.domain.models.core.plan_legacy import Plan, PlanStatus
+from cleveragents.domain.models.core.project_legacy import Project
+from cleveragents.providers.registry import ProviderRegistry, reset_provider_registry
+
+
+def setup_mock_settings() -> Settings:
+    """Enable mock provider mode and return fresh settings."""
+    os.environ["CLEVERAGENTS_MOCK_PROVIDERS"] = "true"
+    os.environ["CLEVERAGENTS_ENV"] = "test"
+    Settings._instance = None
+    return Settings()
+
+
+def setup_lifecycle() -> PlanLifecycleService:
+    """Create lifecycle service configured for integration tests."""
+    settings = setup_mock_settings()
+    return PlanLifecycleService(settings=settings)
+
+
+def create_wf02_action(lifecycle: PlanLifecycleService) -> None:
+    """Create the Workflow Example 2 action definition."""
+    lifecycle.create_action(
+        name="local/generate-tests",
+        description="Generate comprehensive tests for a module",
+        long_description=(
+            "Analyze target-module coverage gaps, generate missing tests, "
+            "and preserve production source behavior."
+        ),
+        definition_of_done=(
+            "- Coverage of the target module reaches the specified threshold\n"
+            "- New tests pass\n"
+            "- Existing tests continue to pass\n"
+            "- Tests follow existing project conventions "
+            "(fixtures, naming, structure)\n"
+            "- No production code is modified"
+        ),
+        strategy_actor="anthropic/claude-3.5-sonnet",
+        execution_actor="anthropic/claude-3.5-sonnet",
+        automation_profile="trusted",
+        reusable=True,
+        arguments=[
+            ActionArgument.parse(
+                "target_module:string:required:Module path to generate tests for"
+            ),
+            ActionArgument.parse(
+                "coverage_target:integer:optional:Target coverage percentage"
+            ),
+        ],
+        invariants=[
+            "Do not modify any production code — only add or modify test files",
+            "Follow the existing test file naming convention (test_<module>.py)",
+            "Use the project's existing test fixtures and conftest.py patterns",
+        ],
+    )
+
+
+def create_trusted_plan(lifecycle: PlanLifecycleService) -> str:
+    """Create and persist a trusted-profile plan for WF02."""
+    plan = lifecycle.use_action(
+        action_name="local/generate-tests",
+        project_links=[ProjectLink(project_name="local/api-service")],
+        arguments={"target_module": "src/auth", "coverage_target": 80},
+        created_by="wf02-integration-test",
+    )
+    plan.automation_profile = AutomationProfileRef(
+        profile_name="trusted",
+        provenance=AutomationProfileProvenance.PLAN,
+    )
+    lifecycle.save_plan(plan)
+    return plan.identity.plan_id
+
+
+@contextmanager
+def auth_module_fixture() -> Iterator[tuple[Path, float]]:
+    """Yield temporary fixture repository with low auth-module coverage."""
+    with tempfile.TemporaryDirectory(prefix="wf02-auth-fixture-") as tmpdir:
+        fixture_root = Path(tmpdir)
+        src_dir = fixture_root / "src"
+        tests_dir = fixture_root / "tests"
+        src_dir.mkdir(parents=True, exist_ok=True)
+        tests_dir.mkdir(parents=True, exist_ok=True)
+
+        (src_dir / "auth.py").write_text(
+            """def authenticate_user(username: str, password: str) -> bool:
+    if not username:
+        return False
+    if password == \"secret\" and username == \"admin\":
+        return True
+    if password == \"\":
+        return False
+    return False
+""",
+            encoding="utf-8",
+        )
+        (src_dir / "__init__.py").write_text("", encoding="utf-8")
+        (tests_dir / "test_auth_smoke.py").write_text(
+            """from src.auth import authenticate_user
+
+
+def test_authenticate_user_valid() -> None:
+    assert authenticate_user(\"admin\", \"secret\") is True
+""",
+            encoding="utf-8",
+        )
+        (tests_dir / "conftest.py").write_text(
+            """import pytest
+
+
+@pytest.fixture
+def auth_credentials() -> tuple[str, str]:
+    return ("admin", "secret")
+""",
+            encoding="utf-8",
+        )
+
+        baseline_coverage = 45.0
+        yield fixture_root, baseline_coverage
+
+
+def validated_relative_test_path(fixture_root: Path, rel_path: str) -> Path:
+    """Resolve and validate a generated relative path under fixture/tests."""
+    candidate = Path(rel_path)
+    if candidate.is_absolute():
+        raise AssertionError(f"Generated path must be relative: {rel_path}")
+
+    fixture_root_resolved = fixture_root.resolve()
+    resolved = (fixture_root_resolved / candidate).resolve()
+    try:
+        relative_to_root = resolved.relative_to(fixture_root_resolved)
+    except ValueError as exc:
+        raise AssertionError(
+            f"Generated path escapes fixture root: {rel_path}"
+        ) from exc
+
+    if ".." in candidate.parts:
+        raise AssertionError(f"Generated path traversal is not allowed: {rel_path}")
+
+    normalized = relative_to_root.as_posix()
+    if not normalized.startswith("tests/"):
+        raise AssertionError(f"Generated path must stay in tests/: {rel_path}")
+
+    return resolved
+
+
+def write_generated_files_safely(
+    fixture_root: Path,
+    generated_files: dict[str, str],
+) -> dict[str, str]:
+    """Write generated files to fixture with traversal and scope guards."""
+    normalized_files: dict[str, str] = {}
+    for rel_path, content in generated_files.items():
+        output_path = validated_relative_test_path(fixture_root, rel_path)
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(content, encoding="utf-8")
+        normalized_rel = output_path.relative_to(fixture_root.resolve()).as_posix()
+        normalized_files[normalized_rel] = content
+    return normalized_files
+
+
+def assert_invariants_and_conventions(
+    fixture_root: Path,
+    plan_id: str,
+    lifecycle: PlanLifecycleService,
+    generated_files: dict[str, str],
+) -> None:
+    """Assert WF02 invariant declarations and generated-file conventions."""
+    plan = lifecycle.get_plan(plan_id)
+    invariant_texts = {inv.text for inv in plan.invariants}
+
+    assert any(
+        "Do not modify any production code" in text for text in invariant_texts
+    ), "Expected production-code invariant on plan"
+    assert any("test file naming convention" in text for text in invariant_texts), (
+        "Expected naming-convention invariant on plan"
+    )
+    assert any(
+        "test fixtures and conftest.py patterns" in text for text in invariant_texts
+    ), "Expected fixture/conftest invariant on plan"
+
+    assert (fixture_root / "tests" / "conftest.py").is_file(), (
+        "Expected fixture project to include tests/conftest.py"
+    )
+
+    generated_paths = sorted(generated_files)
+    assert all(path.startswith("tests/") for path in generated_paths), (
+        f"Expected only tests/ paths, got {generated_paths}"
+    )
+    assert all(path.endswith(".py") for path in generated_paths), (
+        f"Expected python test artifacts, got {generated_paths}"
+    )
+    assert all(Path(path).name.startswith("test_") for path in generated_paths), (
+        f"Expected test_<module>.py naming, got {generated_paths}"
+    )
+
+
+def generate_tests_from_mock_provider(
+    fixture_root: Path,
+    target_module: str,
+    coverage_target: int,
+    include_coverage_suite: bool,
+) -> dict[str, str]:
+    """Generate provider-derived test files for WF02 scenarios."""
+    settings = setup_mock_settings()
+    reset_provider_registry()
+    registry = ProviderRegistry(settings=settings)
+    provider = get_ai_provider(settings=settings, provider_registry=registry)
+    assert provider is not None, "Expected resolved provider in mock mode"
+
+    project = Project(id=1, name="wf02-fixture", path=fixture_root)
+    plan = Plan(
+        id=1,
+        project_id=1,
+        name="wf02-generate-tests",
+        prompt=(
+            f"Generate tests for {target_module} with target coverage "
+            f"{coverage_target}%"
+        ),
+        status=PlanStatus.PENDING,
+        current=True,
+        build=None,
+        build_started_at=None,
+        build_completed_at=None,
+        model_used=None,
+        token_count=None,
+        result=None,
+        applied_at=None,
+        files_created=None,
+        files_modified=None,
+        files_deleted=None,
+    )
+    auth_path = (fixture_root / "src" / "auth.py").resolve()
+    context = Context(
+        id=None,
+        plan_id=1,
+        type=ContextType.FILE,
+        path=str(auth_path),
+        content=auth_path.read_text(encoding="utf-8"),
+        file_hash=None,
+        size=0,
+    )
+
+    provider_response = provider.generate_changes(
+        project=project,
+        plan=plan,
+        contexts=[context],
+    )
+    assert provider_response.error_message is None, provider_response.error_message
+    assert provider_response.model_used.startswith("mock"), provider_response.model_used
+    assert provider_response.token_count >= 0
+    assert len(provider_response.changes) == 1, (
+        "Expected a single mocked provider change for test-generation prompt"
+    )
+
+    provider_change = provider_response.changes[0]
+    provider_content = provider_change.new_content
+    assert isinstance(provider_content, str) and provider_content, (
+        "Expected provider change new_content"
+    )
+
+    generated: dict[str, str] = {
+        "tests/test_provider_mock_output.py": provider_content,
+    }
+
+    if include_coverage_suite:
+        generated["tests/test_auth_generated.py"] = (
+            f'"""Provider-augmented tests for {target_module}."""\n\n'
+            "from src.auth import authenticate_user\n\n"
+            "def test_authenticate_user_accepts_valid_credentials(\n"
+            "    auth_credentials: tuple[str, str],\n"
+            ") -> None:\n"
+            "    username, password = auth_credentials\n"
+            "    assert authenticate_user(username, password) is True\n"
+        )
+        generated["tests/test_auth_edge_cases.py"] = (
+            f'"""Edge tests targeting {coverage_target}% coverage."""\n\n'
+            "from src.auth import authenticate_user\n\n"
+            "def test_authenticate_user_rejects_blank_password(\n"
+            "    auth_credentials: tuple[str, str],\n"
+            ") -> None:\n"
+            "    username, _ = auth_credentials\n"
+            '    assert authenticate_user(username, "") is False\n\n'
+            "def test_authenticate_user_rejects_blank_username() -> None:\n"
+            '    assert authenticate_user("", "secret") is False\n\n'
+            "def test_authenticate_user_rejects_wrong_password() -> None:\n"
+            '    assert authenticate_user("admin", "wrong") is False\n'
+        )
+
+    return generated

robot/wf02_test_generation_integration.robot

@@ -0,0 +1,56 @@
+*** Settings ***
+Documentation    Integration test for Workflow Example 2: automated test generation
+...              for a module using the trusted automation profile.
+...
+...              Validates trusted-profile lifecycle behavior, mocked LLM-based
+...              deterministic test generation, coverage validation enforcement,
+...              and invariant-safe artifacts (test-only file changes).
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment With Database Isolation
+Suite Teardown   Cleanup Test Environment
+Force Tags       wf02    integration    trusted    workflow2    test-generation
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_wf02_test_generation.py
+
+*** Test Cases ***
+WF02 Trusted Profile Auto Runs Strategize And Execute
+    [Documentation]    Verify trusted profile auto-runs strategize and execute
+    ...                while not auto-applying (human approval gate retained).
+    [Tags]    lifecycle    automation-profile    trusted
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    trusted-lifecycle    cwd=${WORKSPACE}    timeout=120s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf02-trusted-lifecycle-ok
+
+WF02 Mocked Generation Produces Test Artifacts Only
+    [Documentation]    Verify mocked LLM responses generate deterministic test
+    ...                files, assert user-facing plan artifacts command path,
+    ...                and artifacts list only test paths (no prod edits).
+    [Tags]    mocking    artifacts    invariants
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    mocked-generation-artifacts    cwd=${WORKSPACE}    timeout=120s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf02-mocked-generation-artifacts-ok
+
+WF02 Coverage Validation Enforced For Target Threshold
+    [Documentation]    Verify required coverage validation blocks below-threshold
+    ...                runs and passes after deterministic generated tests.
+    [Tags]    validation    coverage
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    coverage-validation    cwd=${WORKSPACE}    timeout=120s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf02-coverage-validation-ok
+
+WF02 Path Safety Guardrails Reject Unsafe Destinations
+    [Documentation]    Verify generated-file path guardrails reject absolute,
+    ...                traversal, and non-tests destinations.
+    [Tags]    guardrails    path-safety
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    path-safety-guardrails    cwd=${WORKSPACE}    timeout=120s    on_timeout=kill
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    wf02-path-safety-guardrails-ok

robot/wf02_test_generation_lifecycle.py

@@ -0,0 +1,57 @@
+"""WF02 lifecycle-level utility helpers."""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+
+from cleveragents.application.services.plan_execution_context import (
+    PlanExecutionContext,
+)
+from cleveragents.domain.models.core.change import (
+    ChangeEntry,
+    ChangeOperation,
+    InMemoryChangeSetStore,
+)
+
+
+def record_files_via_execution_context(
+    plan_id: str,
+    store: InMemoryChangeSetStore,
+    files: dict[str, str],
+) -> str:
+    """Record file artifacts through PlanExecutionContext lifecycle APIs."""
+    execution_context = PlanExecutionContext(
+        plan_id=plan_id,
+        automation_profile="trusted",
+        changeset_store=store,
+    )
+    changeset_id = execution_context.start_changeset()
+    for index, rel_path in enumerate(sorted(files), start=1):
+        execution_context.record_change(
+            ChangeEntry(
+                plan_id=plan_id,
+                resource_id=f"wf02-res-{index}",
+                tool_name="builtin/file-write",
+                operation=ChangeOperation.CREATE,
+                path=rel_path,
+                after_hash=f"wf02hash{index}",
+            )
+        )
+    return changeset_id
+
+
+def assert_rejected_generated_path(
+    validator: Callable[[str], object],
+    rel_path: str,
+    expected_error_substring: str,
+) -> None:
+    """Assert the path safety guard rejects a path with expected reason."""
+    try:
+        validator(rel_path)
+    except AssertionError as exc:
+        message = str(exc)
+        assert expected_error_substring in message, (
+            f"Expected '{expected_error_substring}' in '{message}'"
+        )
+    else:
+        raise AssertionError(f"Expected path rejection for: {rel_path}")

robot/wf02_test_generation_validation.py

@@ -0,0 +1,452 @@
+"""WF02 coverage validation lifecycle assertions."""
+
+from __future__ import annotations
+
+import importlib
+import importlib.util
+import inspect
+import json
+import sys
+import trace
+import traceback
+import uuid
+from collections.abc import Callable
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from wf02_test_generation_common import (
+    auth_module_fixture,
+    create_trusted_plan,
+    create_wf02_action,
+    generate_tests_from_mock_provider,
+    setup_lifecycle,
+    write_generated_files_safely,
+)
+from wf02_test_generation_lifecycle import (
+    record_files_via_execution_context,
+)
+
+from cleveragents.application.services.plan_apply_service import (
+    ApplyOutcome,
+    PlanApplyService,
+)
+from cleveragents.application.services.resource_registry_service import (
+    ResourceRegistryService,
+)
+from cleveragents.application.services.tool_registry_service import ToolRegistryService
+from cleveragents.application.services.validation_pipeline import (
+    ValidationCommand,
+    ValidationPipeline,
+)
+from cleveragents.domain.models.core.change import InMemoryChangeSetStore
+from cleveragents.domain.models.core.tool import ValidationMode
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    ToolRegistryRepository,
+    ValidationAttachmentRepository,
+)
+
+
+class _NoCloseSession:
+    """Thin proxy that delegates every attribute to a real SQLAlchemy Session
+    but suppresses ``close()``.
+
+    Repository helpers call ``session.close()`` after each operation.  In
+    integration tests we share a single in-memory SQLite session across
+    multiple repositories so that all writes are visible without a second
+    connection.  Closing the session would invalidate the connection and
+    lose all in-memory data.
+
+    ``object.__setattr__`` / ``object.__getattribute__`` are used in the
+    dunder overrides to access this wrapper's own ``_s`` slot without
+    triggering the proxy's ``__getattr__`` / ``__setattr__``, which would
+    recurse into the wrapped session instead.
+    """
+
+    def __init__(self, session: object) -> None:
+        object.__setattr__(self, "_s", session)
+
+    def close(self) -> None:
+        pass
+
+    def __setattr__(self, name: str, value: object) -> None:
+        setattr(object.__getattribute__(self, "_s"), name, value)
+
+    def __getattr__(self, name: str) -> object:
+        return getattr(object.__getattribute__(self, "_s"), name)
+
+
+def _setup_validation_db() -> tuple[Callable[[], _NoCloseSession], Session]:
+    """Create in-memory DB and return (session_factory, session)."""
+    engine = create_engine("sqlite:///:memory:", echo=False)
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine, expire_on_commit=False)()
+    wrapper = _NoCloseSession(session)
+
+    def factory() -> _NoCloseSession:
+        return wrapper
+
+    return factory, session
+
+
+def _assert_validation_payload(
+    validation_summary: dict[str, Any],
+    *,
+    expected_passed: bool,
+    target: int,
+) -> None:
+    """Assert validation summary payload semantics for WF02 coverage gate."""
+    assert validation_summary["total"] == 1
+    assert validation_summary["informational_passed"] == 0
+    assert validation_summary["informational_failed"] == 0
+
+    if expected_passed:
+        assert validation_summary["required_passed"] == 1
+        assert validation_summary["required_failed"] == 0
+        assert validation_summary["all_required_passed"] is True
+    else:
+        assert validation_summary["required_passed"] == 0
+        assert validation_summary["required_failed"] == 1
+        assert validation_summary["all_required_passed"] is False
+
+    results = validation_summary.get("results")
+    assert isinstance(results, list) and len(results) == 1
+    result = results[0]
+    assert result["validation_name"] == "local/auth-coverage"
+    assert result["resource_name"] == "local/api-service-repo"
+    assert result["mode"] == "required"
+    assert result["passed"] is expected_passed
+
+    data = result.get("data")
+    assert isinstance(data, dict)
+    measured = data.get("measured_coverage")
+    assert isinstance(measured, float)
+    assert data.get("target") == target
+
+    if expected_passed:
+        assert measured >= float(target)
+    else:
+        assert measured < float(target)
+
+
+def _measure_auth_coverage_with_trace(
+    fixture_root: Path,
+) -> tuple[float, str, str, int]:
+    """Measure auth.py line coverage by tracing test function execution."""
+    import os
+    from contextlib import redirect_stderr, redirect_stdout
+    from io import StringIO
+
+    target_file = (fixture_root / "src" / "auth.py").resolve()
+    source_lines = target_file.read_text(encoding="utf-8").splitlines()
+    executable_lines = {
+        line_no
+        for line_no, text in enumerate(source_lines, start=1)
+        if text.strip() and not text.strip().startswith("#")
+    }
+    if not executable_lines:
+        return 0.0, "", "No executable lines found in auth.py", 3
+
+    tracer = trace.Trace(count=True, trace=False)
+    old_cwd = Path.cwd()
+    old_sys_path = list(sys.path)
+    stdout_capture = StringIO()
+    stderr_capture = StringIO()
+    try:
+        os.chdir(fixture_root)
+        sys.path.insert(0, str(fixture_root))
+        importlib.invalidate_caches()
+        with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):
+            for test_file in sorted((fixture_root / "tests").glob("test_*.py")):
+                if test_file.suffix != ".py":
+                    continue
+                if not test_file.name.startswith("test_auth"):
+                    continue
+
+                module_name = f"_wf02_test_{test_file.stem}_{uuid.uuid4().hex}"
+                spec = importlib.util.spec_from_file_location(module_name, test_file)
+                if spec is None or spec.loader is None:
+                    return (
+                        0.0,
+                        stdout_capture.getvalue(),
+                        f"Unable to load {test_file}",
+                        4,
+                    )
+                module = importlib.util.module_from_spec(spec)
+                spec.loader.exec_module(module)
+
+                fixture_values: dict[str, object] = {
+                    "auth_credentials": ("admin", "secret"),
+                }
+
+                for attr_name in dir(module):
+                    if not attr_name.startswith("test_"):
+                        continue
+                    maybe_test = getattr(module, attr_name)
+                    if not callable(maybe_test):
+                        continue
+
+                    signature = inspect.signature(maybe_test)
+                    kwargs: dict[str, object] = {}
+                    for param_name, param in signature.parameters.items():
+                        if param.kind not in (
+                            inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                            inspect.Parameter.KEYWORD_ONLY,
+                        ):
+                            raise AssertionError(
+                                f"Unsupported test function parameter kind for "
+                                f"{attr_name}: {param.kind}"
+                            )
+
+                        if param_name in fixture_values:
+                            kwargs[param_name] = fixture_values[param_name]
+                            continue
+
+                        if param.default is inspect.Parameter.empty:
+                            raise AssertionError(
+                                "Unsupported required test fixture parameter "
+                                f"'{param_name}' in {attr_name}"
+                            )
+
+                    tracer.runfunc(maybe_test, **kwargs)
+
+                for loaded_name in list(sys.modules):
+                    if loaded_name == module_name or loaded_name.startswith(
+                        "_wf02_test_"
+                    ):
+                        sys.modules.pop(loaded_name, None)
+                for cached_name in ("src.auth", "src"):
+                    sys.modules.pop(cached_name, None)
+                importlib.invalidate_caches()
+    except Exception:
+        return 0.0, stdout_capture.getvalue(), traceback.format_exc(), 1
+    finally:
+        sys.path = old_sys_path
+        os.chdir(old_cwd)
+
+    counts = tracer.results().counts
+    covered_lines = {
+        lineno
+        for (filename, lineno), count in counts.items()
+        if count > 0
+        and Path(filename).name == "auth.py"
+        and "src" in Path(filename).parts
+    }
+    measured_pct = (
+        len(covered_lines & executable_lines) / len(executable_lines)
+    ) * 100.0
+    return measured_pct, stdout_capture.getvalue(), stderr_capture.getvalue(), 0
+
+
+def _register_attach_and_run_validation(
+    fixture_root: Path,
+    plan_id: str,
+    target: int,
+) -> dict[str, Any]:
+    """Run validation through registration/attachment/execution lifecycle."""
+    session_factory, _ = _setup_validation_db()
+    tool_repo = ToolRegistryRepository(session_factory=session_factory)
+    attachment_repo = ValidationAttachmentRepository(session_factory=session_factory)
+    tool_service = ToolRegistryService(
+        tool_repo=tool_repo,
+        attachment_repo=attachment_repo,
+    )
+    now = datetime.now(tz=UTC).isoformat()
+    tool_service.register_tool(
+        {
+            "name": "local/auth-coverage",
+            "description": "Auth module coverage gate",
+            "tool_type": "validation",
+            "source": "builtin",
+            "timeout": 300,
+            "created_at": now,
+            "updated_at": now,
+            "resource_bindings": [],
+            "mode": "required",
+        }
+    )
+
+    resource_service = ResourceRegistryService(session_factory=session_factory)
+    resource_service.bootstrap_builtin_types()
+    registered_resource = resource_service.register_resource(
+        type_name="git-checkout",
+        name="local/api-service-repo",
+        location=str(fixture_root),
+        description="WF02 coverage fixture repository",
+    )
+
+    resource_id = registered_resource.resource_id
+    tool_service.attach_validation(
+        validation_name="local/auth-coverage",
+        resource_id=resource_id,
+        mode="required",
+        plan_id=plan_id,
+        args={"target": target},
+    )
+
+    attachments = tool_service.list_validations_for_resource(
+        resource_id=resource_id,
+        plan_id=plan_id,
+    )
+    assert attachments, "Expected at least one attached validation"
+
+    commands: list[ValidationCommand] = []
+    for attachment in attachments:
+        args_json = attachment.get("args_json")
+        args = json.loads(args_json) if isinstance(args_json, str) and args_json else {}
+        mode_text = str(attachment.get("mode", "required"))
+        mode = (
+            ValidationMode.REQUIRED
+            if mode_text == "required"
+            else ValidationMode.INFORMATIONAL
+        )
+        commands.append(
+            ValidationCommand(
+                validation_name=str(attachment["validation_name"]),
+                resource_id=str(attachment["resource_id"]),
+                resource_name="local/api-service-repo",
+                mode=mode,
+                arguments=args,
+                timeout_seconds=30.0,
+            )
+        )
+
+    def executor(
+        validation_name: str,
+        arguments: dict[str, object],
+    ) -> dict[str, object]:
+        target_value = arguments.get("target")
+        if not isinstance(target_value, int):
+            raise TypeError("target must be an int")
+        measured_pct, stdout, stderr, exit_code = _measure_auth_coverage_with_trace(
+            fixture_root
+        )
+        return {
+            "passed": exit_code == 0 and measured_pct >= float(target_value),
+            "message": (
+                f"{validation_name}: measured={measured_pct:.1f}% "
+                f"target={target_value}% exit_code={exit_code}"
+            ),
+            "data": {
+                "stdout": stdout,
+                "stderr": stderr,
+                "measured_coverage": measured_pct,
+                "target": target_value,
+            },
+        }
+
+    plan_metadata: dict[str, Any] = {}
+    summary = ValidationPipeline(
+        commands=commands,
+        executor=executor,
+        max_workers=1,
+    ).run_for_plan(plan_metadata=plan_metadata)
+    assert summary.results, "Expected at least one validation result"
+
+    validation_summary = plan_metadata.get("validation_summary")
+    assert isinstance(validation_summary, dict), "Expected validation_summary metadata"
+    return validation_summary
+
+
+def wf02_coverage_validation() -> None:
+    """Verify coverage gate blocks failing coverage then applies passing coverage."""
+    target = 80
+
+    lifecycle_fail = setup_lifecycle()
+    create_wf02_action(lifecycle_fail)
+    fail_plan_id = create_trusted_plan(lifecycle_fail)
+    lifecycle_fail.try_auto_run(fail_plan_id)
+    store_fail = InMemoryChangeSetStore()
+    apply_fail = PlanApplyService(
+        lifecycle_service=lifecycle_fail, changeset_store=store_fail
+    )
+
+    with auth_module_fixture() as (fail_fixture_root, _):
+        placeholder_path = "tests/test_auth_placeholder.py"
+        (fail_fixture_root / placeholder_path).write_text(
+            """def test_placeholder() -> None:
+    assert True
+""",
+            encoding="utf-8",
+        )
+        fail_changeset_id = record_files_via_execution_context(
+            plan_id=fail_plan_id,
+            store=store_fail,
+            files={placeholder_path: "placeholder"},
+        )
+        fail_plan = lifecycle_fail.get_plan(fail_plan_id)
+        fail_plan.changeset_id = fail_changeset_id
+        fail_plan.sandbox_refs = [f"sandbox-{fail_plan_id[:8]}"]
+        fail_plan.validation_summary = _register_attach_and_run_validation(
+            fail_fixture_root,
+            fail_plan_id,
+            target,
+        )
+        _assert_validation_payload(
+            fail_plan.validation_summary,
+            expected_passed=False,
+            target=target,
+        )
+        lifecycle_fail.save_plan(fail_plan)
+
+        lifecycle_fail.apply_plan(fail_plan_id)
+        lifecycle_fail.start_apply(fail_plan_id)
+        fail_result = apply_fail.apply_with_validation_gate(fail_plan_id)
+        assert fail_result.outcome == ApplyOutcome.CONSTRAINED
+
+        constrained_plan = lifecycle_fail.get_plan(fail_plan_id)
+        assert constrained_plan.processing_state.value == "constrained"
+
+    lifecycle_pass = setup_lifecycle()
+    create_wf02_action(lifecycle_pass)
+    pass_plan_id = create_trusted_plan(lifecycle_pass)
+    lifecycle_pass.try_auto_run(pass_plan_id)
+    store_pass = InMemoryChangeSetStore()
+    apply_pass = PlanApplyService(
+        lifecycle_service=lifecycle_pass, changeset_store=store_pass
+    )
+
+    with auth_module_fixture() as (pass_fixture_root, _):
+        generated_files = generate_tests_from_mock_provider(
+            pass_fixture_root,
+            "src/auth",
+            target,
+            include_coverage_suite=True,
+        )
+        normalized_files = write_generated_files_safely(
+            pass_fixture_root,
+            generated_files,
+        )
+        pass_changeset_id = record_files_via_execution_context(
+            plan_id=pass_plan_id,
+            store=store_pass,
+            files=normalized_files,
+        )
+
+        pass_plan = lifecycle_pass.get_plan(pass_plan_id)
+        pass_plan.changeset_id = pass_changeset_id
+        pass_plan.sandbox_refs = [f"sandbox-{pass_plan_id[:8]}"]
+        pass_plan.validation_summary = _register_attach_and_run_validation(
+            pass_fixture_root,
+            pass_plan_id,
+            target,
+        )
+        _assert_validation_payload(
+            pass_plan.validation_summary,
+            expected_passed=True,
+            target=target,
+        )
+        lifecycle_pass.save_plan(pass_plan)
+
+        lifecycle_pass.apply_plan(pass_plan_id)
+        lifecycle_pass.start_apply(pass_plan_id)
+        pass_result = apply_pass.apply_with_validation_gate(pass_plan_id)
+        assert pass_result.outcome == ApplyOutcome.APPLIED
+
+        applied_plan = lifecycle_pass.get_plan(pass_plan_id)
+        assert applied_plan.processing_state.value == "applied"
+
+    print("wf02-coverage-validation-ok")