fix(plan-lifecycle): gate Apply phase on Definition-of-Done evaluation

Before transitioning to the Apply phase, PlanLifecycleService.apply_plan
now evaluates the plan's definition_of_done criteria using DoDEvaluator
(TextMatchEvaluator). If any required criteria fail, a DoDGatingError is
raised and the plan remains in Execute/COMPLETE state. The evaluation
result is stored in plan.validation_summary with dod_evaluated=True and
dod_all_passed reflecting the outcome. Plans with no DoD text skip
evaluation and proceed normally.

Adds DoDGatingError exception class and _evaluate_dod() helper method.
Adds BDD feature file and step definitions for DoD gating scenarios.

ISSUES CLOSED: #7927

CHANGELOG.md

@@ -753,6 +753,14 @@ uko-oo:Class` triple emission in `PythonAnalyzer._extract_class()` so that
   message listing available built-in profiles. The resolved profile name is also
   logged at debug level for observability.
 
+- **DoD Gating in Apply Phase** (#7927): `PlanLifecycleService.apply_plan` now
+  evaluates the plan's `definition_of_done` criteria before transitioning to the
+  Apply phase. If any required criteria fail, a `DoDGatingError` is raised and
+  the plan remains in Execute/COMPLETE state. The evaluation result is stored in
+  `plan.validation_summary` with `dod_evaluated=True` and `dod_all_passed`
+  reflecting the outcome. Plans with no DoD text skip evaluation and proceed
+  normally.
+
 ### Added
 
 - **ACMS Index Data Model and File Traversal Engine** (#9579): Implements the

CONTRIBUTORS.md

@@ -61,3 +61,4 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the actor compiler `actor_ref` field fix (issue #1429): corrected `_map_node()` and `compile_actor()` in `src/cleveragents/actor/compiler.py` to read `actor_ref` from the top-level `NodeDefinition.actor_ref` field instead of `node.config.get("actor_ref")`, resolving silent failures on all SUBGRAPH nodes where `subgraph_refs` was always empty and `NodeConfig.subgraph` was always `None`.
 * HAL 9000 has contributed the removal of the unsupported executable resource type (PR #3248 / issue #3077): removed `executable` from `LSP_RESOURCE_TYPES` and `BUILTIN_TYPE_NAMES`, updated `agents resource list` CLI table columns to the spec-required `[Name, ID, Type, Phys/Virt, Children, Projects]`, deleted orphaned `examples/resource-types/executable.yaml`, and updated related BDD test coverage.
 * HAL 9000 has contributed the alembic fileConfig error handling fix (PR #8288 / issue #7874): wrapped the `fileConfig()` call in `alembic/env.py` with a `try/except` block to catch malformed INI logging configuration and emit clear, actionable error messages to stderr.
+* HAL 9000 has contributed the Definition-of-Done gating feature for the Apply phase (PR #8299 / issue #7927): `PlanLifecycleService.apply_plan` now evaluates DoD criteria before transitioning to Apply, raising `DoDGatingError` when required criteria fail and storing evaluation results in `plan.validation_summary`.

features/plan_dod_gating.feature

@@ -0,0 +1,85 @@
+Feature: Definition-of-Done Gating in Apply Phase
+  As a developer
+  I want the Apply phase to evaluate Definition-of-Done criteria
+  So that plans cannot be applied until all required DoD criteria pass
+
+  Background:
+    Given I have a plan lifecycle service for DoD gating
+
+  # --- DoD evaluation is skipped when no DoD text is set ---
+
+  Scenario: Apply succeeds when plan has no definition of done
+    Given I have a plan with no definition of done in execute complete state
+    When I apply the plan with DoD gating
+    Then the plan should transition to apply phase
+    And the plan validation summary should be empty
+
+  Scenario: Apply succeeds when definition of done is empty string
+    Given I have a plan with empty definition of done in execute complete state
+    When I apply the plan with DoD gating
+    Then the plan should transition to apply phase
+    And the plan validation summary should be empty
+
+  # --- DoD evaluation runs and passes ---
+
+  Scenario: Apply succeeds when all DoD criteria pass via context match
+    Given I have a plan with definition of done "tests_passed" and argument "tests_passed" equal to "true"
+    When I apply the plan with DoD gating
+    Then the plan should transition to apply phase
+    And the plan validation summary should record dod_evaluated as true
+    And the plan validation summary should record dod_all_passed as true
+
+  Scenario: Apply succeeds when DoD criteria are skipped due to empty context
+    Given I have a plan with definition of done "all tests pass" and no arguments
+    When I apply the plan with DoD gating
+    Then the plan should transition to apply phase
+    And the plan validation summary should record dod_evaluated as true
+
+  # --- DoD evaluation blocks apply when criteria fail ---
+
+  Scenario: Apply is blocked when DoD criteria fail
+    Given I have a plan with definition of done "coverage_above_97" and argument "status" equal to "incomplete"
+    When I try to apply the plan with DoD gating
+    Then a DoD gating error should be raised
+    And the plan should remain in execute phase
+    And the plan validation summary should record dod_evaluated as true
+    And the plan validation summary should record dod_all_passed as false
+
+  Scenario: DoDGatingError contains correct failure counts
+    Given I have a plan with definition of done "criterion_alpha" and argument "status" equal to "incomplete"
+    When I try to apply the plan with DoD gating
+    Then a DoD gating error should be raised
+    And the DoD gating error should report 1 failed criteria out of 1 total
+
+  # --- DoD evaluation with template rendering ---
+
+  Scenario: DoD template placeholders are rendered with plan arguments
+    Given I have a plan with definition of done "{status}" and argument "status" equal to "complete"
+    When I apply the plan with DoD gating
+    Then the plan should transition to apply phase
+
+  # --- validation_summary is populated ---
+
+  Scenario: validation_summary is populated after DoD evaluation
+    Given I have a plan with definition of done "tests_passed" and argument "tests_passed" equal to "true"
+    When I apply the plan with DoD gating
+    Then the plan validation summary should contain key "dod_evaluated" with value true
+    And the plan validation summary should contain key "dod_all_passed" with value true
+    And the plan validation summary should contain key "total"
+    And the plan validation summary should contain key "required_passed"
+
+  # --- DoD gating integrates with full lifecycle ---
+
+  Scenario: Plan with passing DoD can complete full apply lifecycle
+    Given I have a plan with definition of done "tests_passed" and argument "tests_passed" equal to "true"
+    When I apply the plan with DoD gating
+    And I start the apply phase
+    And I complete the apply phase
+    Then the dod plan processing state should be "applied"
+
+  Scenario: Plan with failing DoD stays in execute phase after blocked apply
+    Given I have a plan with definition of done "coverage_above_97" and argument "status" equal to "incomplete"
+    When I try to apply the plan with DoD gating
+    Then a DoD gating error should be raised
+    And the dod plan phase should be "execute"
+    And the dod plan processing state should be "complete"

features/steps/plan_dod_gating_steps.py

@@ -0,0 +1,312 @@
+"""Step definitions for Definition-of-Done gating in the Apply phase.
+
+Tests that ``PlanLifecycleService.apply_plan`` evaluates DoD criteria
+before transitioning to the Apply phase and blocks the transition when
+any required criteria fail.
+"""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    DoDGatingError,
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.core.plan import (
+    PlanPhase,
+    ProjectLink,
+)
+
+# ---------------------------------------------------------------------------
+# Background / setup
+# ---------------------------------------------------------------------------
+
+
+@given("I have a plan lifecycle service for DoD gating")
+def step_create_dod_lifecycle_service(context: Context) -> None:
+    """Create a fresh PlanLifecycleService for DoD gating tests."""
+    settings = Settings()
+    context.dod_service = PlanLifecycleService(settings=settings)
+    context.dod_plan = None
+    context.dod_error = None
+
+
+# ---------------------------------------------------------------------------
+# Plan creation helpers
+# ---------------------------------------------------------------------------
+
+
+def _create_plan_in_execute_complete(
+    context: Context,
+    override_dod: str | None = None,
+    override_arguments: dict[str, str] | None = None,
+) -> None:
+    """Create an action + plan and drive it to Execute/COMPLETE state.
+
+    The action is always created with a valid DoD text.  After plan
+    creation, the plan's ``definition_of_done`` and ``arguments`` fields
+    are overridden directly so tests can exercise arbitrary DoD text and
+    argument combinations without needing to declare them on the action.
+    """
+    service: PlanLifecycleService = context.dod_service
+
+    action = service.create_action(
+        name="local/dod-test-action",
+        description="DoD gating test action",
+        definition_of_done="placeholder definition of done",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+    )
+
+    plan = service.use_action(
+        action_name=str(action.namespaced_name),
+        project_links=[ProjectLink(project_name="test-project")],
+    )
+
+    # Override DoD and arguments directly on the plan object so tests
+    # can exercise arbitrary combinations without action-level validation.
+    if override_dod is not None:
+        plan.definition_of_done = override_dod
+    if override_arguments is not None:
+        plan.arguments = override_arguments
+
+    # Drive to Execute/COMPLETE
+    service.start_strategize(plan.identity.plan_id)
+    service.complete_strategize(plan.identity.plan_id)
+    service.execute_plan(plan.identity.plan_id)
+    service.start_execute(plan.identity.plan_id)
+    service.complete_execute(plan.identity.plan_id)
+
+    context.dod_plan = service.get_plan(plan.identity.plan_id)
+
+
+@given("I have a plan with no definition of done in execute complete state")
+def step_plan_no_dod(context: Context) -> None:
+    """Create a plan with definition_of_done cleared to empty string."""
+    _create_plan_in_execute_complete(context, override_dod="")
+
+
+@given("I have a plan with empty definition of done in execute complete state")
+def step_plan_empty_dod(context: Context) -> None:
+    """Create a plan with definition_of_done set to whitespace only."""
+    _create_plan_in_execute_complete(context, override_dod="   ")
+
+
+@given(
+    'I have a plan with definition of done "{dod_text}" and argument "{arg_key}" equal to "{arg_value}"'
+)
+def step_plan_with_dod_and_arg(
+    context: Context, dod_text: str, arg_key: str, arg_value: str
+) -> None:
+    """Create a plan with a specific DoD text and a single argument."""
+    _create_plan_in_execute_complete(
+        context,
+        override_dod=dod_text,
+        override_arguments={arg_key: arg_value},
+    )
+
+
+@given('I have a plan with definition of done "{dod_text}" and no arguments')
+def step_plan_with_dod_no_args(context: Context, dod_text: str) -> None:
+    """Create a plan with a specific DoD text and no arguments."""
+    # Replace literal \n with actual newlines for multi-line DoD
+    actual_dod = dod_text.replace("\\n", "\n")
+    _create_plan_in_execute_complete(
+        context,
+        override_dod=actual_dod,
+        override_arguments={},
+    )
+
+
+# ---------------------------------------------------------------------------
+# Apply actions
+# ---------------------------------------------------------------------------
+
+
+@when("I apply the plan with DoD gating")
+def step_apply_plan_dod(context: Context) -> None:
+    """Apply the plan (expecting success)."""
+    service: PlanLifecycleService = context.dod_service
+    plan_id = context.dod_plan.identity.plan_id
+    context.dod_error = None
+    context.dod_plan = service.apply_plan(plan_id)
+
+
+@when("I try to apply the plan with DoD gating")
+def step_try_apply_plan_dod(context: Context) -> None:
+    """Attempt to apply the plan, capturing DoDGatingError if raised."""
+    service: PlanLifecycleService = context.dod_service
+    plan_id = context.dod_plan.identity.plan_id
+    context.dod_error = None
+    try:
+        context.dod_plan = service.apply_plan(plan_id)
+    except DoDGatingError as exc:
+        context.dod_error = exc
+        # Refresh plan state from service after the blocked attempt
+        context.dod_plan = service.get_plan(plan_id)
+
+
+@when("I start the apply phase")
+def step_start_apply_dod(context: Context) -> None:
+    """Start the Apply phase processing."""
+    service: PlanLifecycleService = context.dod_service
+    plan_id = context.dod_plan.identity.plan_id
+    context.dod_plan = service.start_apply(plan_id)
+
+
+@when("I complete the apply phase")
+def step_complete_apply_dod(context: Context) -> None:
+    """Complete the Apply phase."""
+    service: PlanLifecycleService = context.dod_service
+    plan_id = context.dod_plan.identity.plan_id
+    context.dod_plan = service.complete_apply(plan_id)
+
+
+# ---------------------------------------------------------------------------
+# Assertions
+# ---------------------------------------------------------------------------
+
+
+@then("the plan should transition to apply phase")
+def step_assert_plan_in_apply(context: Context) -> None:
+    """Assert the plan is now in the Apply phase."""
+    plan = context.dod_plan
+    assert plan.phase == PlanPhase.APPLY, (
+        f"Expected plan phase 'apply', got '{plan.phase.value}'"
+    )
+
+
+@then("the plan should remain in execute phase")
+def step_assert_plan_in_execute(context: Context) -> None:
+    """Assert the plan is still in the Execute phase."""
+    plan = context.dod_plan
+    assert plan.phase == PlanPhase.EXECUTE, (
+        f"Expected plan phase 'execute', got '{plan.phase.value}'"
+    )
+
+
+@then('the dod plan phase should be "{expected_phase}"')
+def step_assert_dod_plan_phase(context: Context, expected_phase: str) -> None:
+    """Assert the DoD plan phase matches the expected value."""
+    plan = context.dod_plan
+    assert plan.phase.value == expected_phase, (
+        f"Expected plan phase '{expected_phase}', got '{plan.phase.value}'"
+    )
+
+
+@then('the dod plan processing state should be "{expected_state}"')
+def step_assert_dod_plan_processing_state(
+    context: Context, expected_state: str
+) -> None:
+    """Assert the DoD plan processing state matches the expected value."""
+    plan = context.dod_plan
+    assert plan.processing_state.value == expected_state, (
+        f"Expected processing state '{expected_state}', "
+        f"got '{plan.processing_state.value}'"
+    )
+
+
+@then("the plan validation summary should be empty")
+def step_assert_validation_summary_empty(context: Context) -> None:
+    """Assert the plan's validation_summary is None (no DoD evaluated)."""
+    plan = context.dod_plan
+    assert plan.validation_summary is None, (
+        f"Expected validation_summary to be None, got {plan.validation_summary!r}"
+    )
+
+
+@then("the plan validation summary should record dod_evaluated as true")
+def step_assert_dod_evaluated_true(context: Context) -> None:
+    """Assert validation_summary has dod_evaluated=True."""
+    plan = context.dod_plan
+    assert plan.validation_summary is not None, (
+        "Expected validation_summary to be set, but it is None"
+    )
+    assert plan.validation_summary.get("dod_evaluated") is True, (
+        f"Expected dod_evaluated=True, got {plan.validation_summary!r}"
+    )
+
+
+@then("the plan validation summary should record dod_all_passed as true")
+def step_assert_dod_all_passed_true(context: Context) -> None:
+    """Assert validation_summary has dod_all_passed=True."""
+    plan = context.dod_plan
+    assert plan.validation_summary is not None, (
+        "Expected validation_summary to be set, but it is None"
+    )
+    assert plan.validation_summary.get("dod_all_passed") is True, (
+        f"Expected dod_all_passed=True, got {plan.validation_summary!r}"
+    )
+
+
+@then("the plan validation summary should record dod_all_passed as false")
+def step_assert_dod_all_passed_false(context: Context) -> None:
+    """Assert validation_summary has dod_all_passed=False."""
+    plan = context.dod_plan
+    assert plan.validation_summary is not None, (
+        "Expected validation_summary to be set, but it is None"
+    )
+    assert plan.validation_summary.get("dod_all_passed") is False, (
+        f"Expected dod_all_passed=False, got {plan.validation_summary!r}"
+    )
+
+
+@then("a DoD gating error should be raised")
+def step_assert_dod_gating_error(context: Context) -> None:
+    """Assert a DoDGatingError was raised."""
+    assert context.dod_error is not None, (
+        "Expected a DoDGatingError to be raised, but no error was captured"
+    )
+    assert isinstance(context.dod_error, DoDGatingError), (
+        f"Expected DoDGatingError, got {type(context.dod_error).__name__}"
+    )
+
+
+@then(
+    "the DoD gating error should report {failed:d} failed criteria out of {total:d} total"
+)
+def step_assert_dod_error_counts(context: Context, failed: int, total: int) -> None:
+    """Assert the DoDGatingError has the expected failure counts."""
+    error = context.dod_error
+    assert isinstance(error, DoDGatingError), (
+        f"Expected DoDGatingError, got {type(error).__name__}"
+    )
+    assert error.failed_count == failed, (
+        f"Expected failed_count={failed}, got {error.failed_count}"
+    )
+    assert error.total_count == total, (
+        f"Expected total_count={total}, got {error.total_count}"
+    )
+
+
+@then('the plan validation summary should contain key "{key}" with value true')
+def step_assert_validation_summary_key_true(context: Context, key: str) -> None:
+    """Assert validation_summary contains a key with value True."""
+    plan = context.dod_plan
+    assert plan.validation_summary is not None, (
+        "Expected validation_summary to be set, but it is None"
+    )
+    assert key in plan.validation_summary, (
+        f"Expected key '{key}' in validation_summary, "
+        f"got keys: {list(plan.validation_summary.keys())}"
+    )
+    assert plan.validation_summary[key] is True, (
+        f"Expected validation_summary['{key}']=True, "
+        f"got {plan.validation_summary[key]!r}"
+    )
+
+
+@then('the plan validation summary should contain key "{key}"')
+def step_assert_validation_summary_has_key(context: Context, key: str) -> None:
+    """Assert validation_summary contains a specific key."""
+    plan = context.dod_plan
+    assert plan.validation_summary is not None, (
+        "Expected validation_summary to be set, but it is None"
+    )
+    assert key in plan.validation_summary, (
+        f"Expected key '{key}' in validation_summary, "
+        f"got keys: {list(plan.validation_summary.keys())}"
+    )

src/cleveragents/application/services/plan_lifecycle_service.py

@@ -81,6 +81,13 @@ from cleveragents.domain.models.core.automation_profile import (
     AutomationProfile,
 )
 from cleveragents.domain.models.core.decision import ContextSnapshot, DecisionType
+from cleveragents.domain.models.core.definition_of_done import (
+    DoDEvaluator,
+    DoDSummary,
+    TextMatchEvaluator,
+    parse_dod_criteria,
+    render_dod_template,
+)
 from cleveragents.domain.models.core.plan import (
     AutomationProfileProvenance,
     AutomationProfileRef,
@@ -175,6 +182,26 @@ class ReconciliationBlockedError(BusinessRuleViolation):
         self.reason = reason
 
 
+class DoDGatingError(BusinessRuleViolation):
+    """Raised when Definition-of-Done evaluation blocks the Apply transition.
+
+    The plan's ``definition_of_done`` criteria were evaluated before
+    transitioning to the Apply phase and one or more required criteria
+    failed.  The plan cannot proceed to APPLIED until all required
+    criteria pass.
+    """
+
+    def __init__(self, plan_id: str, failed_count: int, total_count: int):
+        super().__init__(
+            f"Plan {plan_id} cannot be applied: "
+            f"{failed_count}/{total_count} Definition-of-Done criteria failed. "
+            "All required criteria must pass before apply is permitted."
+        )
+        self.plan_id = plan_id
+        self.failed_count = failed_count
+        self.total_count = total_count
+
+
 class PlanLifecycleService:
     """Service for managing v3 plan lifecycle.
 
@@ -198,6 +225,7 @@ class PlanLifecycleService:
         invariant_service: InvariantService | None = None,
         lock_service: LockService | None = None,
         autonomy_controller: AutonomyController | None = None,
+        dod_evaluator: DoDEvaluator | None = None,
     ):
         """Initialize the plan lifecycle service.
 
@@ -259,6 +287,7 @@ class PlanLifecycleService:
         self.invariant_service = invariant_service
         self._lock_service = lock_service
         self._autonomy_controller = autonomy_controller
+        self._dod_evaluator: DoDEvaluator = dod_evaluator or TextMatchEvaluator()
         self._logger = logger.bind(service="plan_lifecycle")
         self.preflight_guardrail = PlanPreflightGuardrail()
         self._subscribe_correction_reconciliation()
@@ -553,6 +582,110 @@ class PlanLifecycleService:
                 reason=str(exc),
             ) from exc
 
+    def _evaluate_dod(self, plan: Plan) -> DoDSummary | None:
+        """Evaluate the plan's Definition-of-Done criteria before Apply.
+
+        Parses the plan's ``definition_of_done`` text into individual
+        criteria and evaluates each one using the ``TextMatchEvaluator``
+        with the plan's arguments as context.
+
+        The evaluation result is stored in ``plan.validation_summary``
+        so that downstream consumers (CLI, audit log) can inspect it.
+
+        When the plan has no ``definition_of_done`` text (empty or
+        ``None``), evaluation is skipped and ``None`` is returned.
+
+        Args:
+            plan: The plan to evaluate DoD for.
+
+        Returns:
+            A ``DoDSummary`` with per-criterion results, or ``None``
+            when the plan has no DoD text.
+
+        Raises:
+            DoDGatingError: If any required DoD criteria fail.
+        """
+        dod_text = plan.definition_of_done
+        if not dod_text or not dod_text.strip():
+            return None
+
+        plan_id = plan.identity.plan_id
+
+        # Render template placeholders using plan arguments
+        try:
+            rendered = render_dod_template(dod_text, plan.arguments or {})
+        except ValueError:
+            self._logger.warning(
+                "dod_template_render_failed",
+                plan_id=plan_id,
+                exc_info=True,
+            )
+            rendered = dod_text
+
+        # Parse criteria
+        criteria = parse_dod_criteria(rendered)
+        if not criteria:
+            return None
+
+        # Evaluate using the injected DoDEvaluator (defaults to TextMatchEvaluator)
+        context: dict[str, object] = dict(plan.arguments or {})
+        results = self._dod_evaluator.evaluate(criteria, context)
+
+        summary = DoDSummary(
+            plan_id=plan_id,
+            definition_of_done=rendered,
+            results=results,
+        )
+
+        # Store evaluation results in plan.validation_summary
+        plan.validation_summary = summary.to_validation_summary()
+        plan.timestamps.updated_at = datetime.now()
+        self._commit_plan(plan)
+
+        self._logger.info(
+            "dod_evaluation_complete",
+            plan_id=plan_id,
+            total=summary.total,
+            passed=summary.passed,
+            failed=summary.failed,
+            skipped=summary.skipped,
+            all_passed=summary.all_passed,
+        )
+
+        # Emit DOD_EVALUATED event
+        if self.event_bus is not None:
+            try:
+                self.event_bus.emit(
+                    DomainEvent(
+                        event_type=EventType.PLAN_STATE_CHANGED,
+                        plan_id=plan_id,
+                        details={
+                            "dod_evaluated": True,
+                            "dod_all_passed": summary.all_passed,
+                            "dod_total": summary.total,
+                            "dod_passed": summary.passed,
+                            "dod_failed": summary.failed,
+                        },
+                    )
+                )
+            except Exception:
+                self._logger.warning(
+                    "event_bus_emit_failed",
+                    event_type="DOD_EVALUATED",
+                    plan_id=plan_id,
+                    exc_info=True,
+                )
+
+        # Gate: block apply if any criteria failed
+        if not summary.all_passed and summary.failed > 0:
+            raise DoDGatingError(
+                plan_id=plan_id,
+                failed_count=summary.failed,
+                total_count=summary.total,
+            )
+
+        return summary
+
     def _subscribe_correction_reconciliation(self) -> None:
         """Subscribe to CORRECTION_APPLIED events for post-correction reconciliation.
 
@@ -1919,6 +2052,11 @@ class PlanLifecycleService:
             # Invariant Reconciliation: verify invariants before Apply
             self._run_invariant_reconciliation(plan)
 
+            # Definition-of-Done gating: evaluate DoD criteria before Apply.
+            # Raises DoDGatingError if any required criteria fail, blocking
+            # the transition until all criteria pass.
+            self._evaluate_dod(plan)
+
             # Transition to Apply phase — set processing_state first so that
             # the phase-state validator sees QUEUED (valid in any phase) when
             # the phase assignment triggers re-validation.