fix(automation): respect automation profile gates in lifecycle service and async jobs

Add BDD regression tests for issue #4328 automation profile gates Tests verify that complete_strategize() and complete_execute() respect automation profile thresholds and do NOT unconditionally call auto_progress(). Covers 8 built-in profiles: manual, full-auto, supervised, auto, review_before_apply, ci, trusted ISSUES CLOSED: #4328
2026-05-11 13:38:44 +00:00
parent f5761912db
commit ddfa3af94c
15 changed files with 2091 additions and 171 deletions
@@ -79,6 +79,19 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
   removal. Mocked existing steps to allow remaining V2 features to be 
   covered/tested.

+- **Automation profile threshold gates fully respect spec semantics** (#4328): Added
+  ``_should_auto_progress_for_threshold()`` helper in ``PlanLifecycleService`` that
+  implements the full spec Threshold Semantics: ``0.0`` = always auto, ``1.0`` =
+  always human approval, ``0.0 < v < 1.0`` = proceed only if confidence >= threshold.
+  Integrated with ``AutonomyController.should_proceed_automatically()`` for
+  intermediate thresholds. Updated ``should_auto_progress()``, ``try_auto_run()``,
+  ``execute_async_job()``, ``try_auto_revert_from_apply()``, and
+  ``try_auto_revert_from_execute()`` to use the helper. Previously the service
+  only checked ``< 1.0`` (treated all intermediates as auto). Added BDD regression
+  tests in ``features/tdd_automation_profile_gates_4328.feature`` covering all 8
+  built-in profiles including the ``cautious`` profile's intermediate thresholds
+  (e.g. ``create_tool=0.7``).
+
 - **TUI Prompt Symbol Mode Awareness** (#6431): The prompt widget now displays a
  mode-dependent symbol (`❯` normal, `/` command, `$` shell, `☰` multi-line),
  implemented via `_PromptSymbolMixin` and `InputMode.MULTILINE`. The widget uses
@@ -0,0 +1,139 @@
+Feature: Plan Lifecycle Service coverage boost round 3
+  As a developer
+  I want to exercise remaining uncovered code paths in PlanLifecycleService
+  So that code coverage improves beyond the current per-file threshold
+
+  Background:
+    Given I have a fresh plan lifecycle service for coverage boost r3
+
+  # ---------------------------------------------------------------
+  # complete_strategize — event_bus.emit PLAN_STATE_CHANGED success
+  # ---------------------------------------------------------------
+
+  Scenario: complete_strategize emits PLAN_STATE_CHANGED event via event bus
+    Given a plan lifecycle service with a recording event bus for r3
+    And an action "local/strat-complete-event" exists for coverage boost r3
+    And a plan in strategize phase with processing state for r3
+    When I complete strategize on the plan with event bus
+    Then the event bus should have recorded a PLAN_STATE_CHANGED event
+
+  # ---------------------------------------------------------------
+  # complete_strategize — event_bus.emit PLAN_STATE_CHANGED exception
+  # ---------------------------------------------------------------
+
+  Scenario: complete_strategize catches event bus emission failure
+    Given a plan lifecycle service with a failing event bus for r3
+    And an action "local/strat-complete-fail" exists for coverage boost r3
+    And a plan in strategize phase with processing state for r3
+    When I complete strategize on the plan and event bus emit fails
+    Then the plan phase should be "strategize" and state should be "complete"
+
+  # ---------------------------------------------------------------
+  # fail_strategize — event_bus.emit PLAN_ERRORED success
+  # ---------------------------------------------------------------
+
+  Scenario: fail_strategize emits PLAN_ERRORED event via event bus
+    Given a plan lifecycle service with a recording event bus for r3
+    And an action "local/strat-fail-event" exists for coverage boost r3
+    And a plan in strategize phase with processing state for r3
+    When I fail strategize on the plan with error "Test strategy error"
+    Then the event bus should have recorded a PLAN_ERRORED event
+
+  # ---------------------------------------------------------------
+  # fail_strategize — event_bus.emit PLAN_ERRORED exception
+  # ---------------------------------------------------------------
+
+  Scenario: fail_strategize catches event bus emission failure
+    Given a plan lifecycle service with a failing event bus for r3
+    And an action "local/strat-fail-fail" exists for coverage boost r3
+    And a plan in strategize phase with processing state for r3
+    When I fail strategize on the plan and event bus emit fails
+    Then the plan should be in errored state for r3
+
+  # ---------------------------------------------------------------
+  # complete_execute — event_bus.emit PLAN_STATE_CHANGED success
+  # ---------------------------------------------------------------
+
+  Scenario: complete_execute emits PLAN_STATE_CHANGED event via event bus
+    Given a plan lifecycle service with a recording event bus for r3
+    And an action "local/exec-complete-event" exists for coverage boost r3
+    And a plan in execute phase with processing state for r3
+    When I complete execute on the plan with event bus
+    Then the event bus should have recorded a PLAN_STATE_CHANGED event
+
+  # ---------------------------------------------------------------
+  # complete_execute — event_bus.emit PLAN_STATE_CHANGED exception
+  # ---------------------------------------------------------------
+
+  Scenario: complete_execute catches event bus emission failure
+    Given a plan lifecycle service with a failing event bus for r3
+    And an action "local/exec-complete-fail" exists for coverage boost r3
+    And a plan in execute phase with processing state for r3
+    When I complete execute on the plan and event bus emit fails
+    Then the plan phase should be "execute" and state should be "complete"
+
+  # ---------------------------------------------------------------
+  # fail_execute — event_bus.emit PLAN_ERRORED success
+  # ---------------------------------------------------------------
+
+  Scenario: fail_execute emits PLAN_ERRORED event via event bus
+    Given a plan lifecycle service with a recording event bus for r3
+    And an action "local/exec-fail-event" exists for coverage boost r3
+    And a plan in execute phase with processing state for r3
+    When I fail execute on the plan with error "Test execution error"
+    Then the event bus should have recorded a PLAN_ERRORED event
+
+  # ---------------------------------------------------------------
+  # fail_execute — event_bus.emit PLAN_ERRORED exception
+  # ---------------------------------------------------------------
+
+  Scenario: fail_execute catches event bus emission failure
+    Given a plan lifecycle service with a failing event bus for r3
+    And an action "local/exec-fail-fail" exists for coverage boost r3
+    And a plan in execute phase with processing state for r3
+    When I fail execute on the plan and event bus emit fails
+    Then the plan should be in errored state for r3
+
+  # ---------------------------------------------------------------
+  # fail_apply — event_bus.emit PLAN_ERRORED success
+  # ---------------------------------------------------------------
+
+  Scenario: fail_apply emits PLAN_ERRORED event via event bus
+    Given a plan lifecycle service with a recording event bus for r3
+    And an action "local/apply-fail-event" exists for coverage boost r3
+    And a plan in apply phase with processing state for r3
+    When I fail apply on the plan with error "Test apply error"
+    Then the event bus should have recorded a PLAN_ERRORED event
+
+  # ---------------------------------------------------------------
+  # fail_apply — event_bus.emit PLAN_ERRORED exception
+  # ---------------------------------------------------------------
+
+  Scenario: fail_apply catches event bus emission failure
+    Given a plan lifecycle service with a failing event bus for r3
+    And an action "local/apply-fail-fail" exists for coverage boost r3
+    And a plan in apply phase with processing state for r3
+    When I fail apply on the plan and event bus emit fails
+    Then the plan should be in errored state for r3
+
+  # ---------------------------------------------------------------
+  # fail_execute — _cleanup_devcontainers called on terminal failure
+  # ---------------------------------------------------------------
+
+  Scenario: fail_execute triggers devcontainer cleanup on terminal failure
+    Given a plan lifecycle service with mocked devcontainer cleanup for r3
+    And an action "local/exec-fail-cleanup" exists for coverage boost r3
+    And a plan in execute phase with processing state for r3
+    When I fail execute on the plan with error "Container cleanup test"
+    Then the plan should be in errored state and cleanup should have been called
+
+  # ---------------------------------------------------------------
+  # fail_apply — _cleanup_devcontainers called on terminal failure
+  # ---------------------------------------------------------------
+
+  Scenario: fail_apply triggers devcontainer cleanup on terminal failure
+    Given a plan lifecycle service with mocked devcontainer cleanup for r3
+    And an action "local/apply-fail-cleanup" exists for coverage boost r3
+    And a plan in apply phase with processing state for r3
+    When I fail apply on the plan with error "Container cleanup test"
+    Then the plan should be in errored state and cleanup should have been called
@@ -0,0 +1,163 @@
+Feature: Plan Lifecycle Service coverage boost round 4
+  As a developer
+  I want to exercise remaining uncovered code paths in PlanLifecycleService
+  So that code coverage improves to meet the 97% threshold
+
+  Background:
+    Given I have a fresh plan lifecycle service for coverage boost r4
+
+  # ---------------------------------------------------------------
+  # _run_estimation — event_bus.emit PLAN_ESTIMATION_COMPLETE
+  # coverage (via execute_plan when estimation_actor is configured)
+  # ---------------------------------------------------------------
+
+  Scenario: execute_plan with estimation_actor succeeds and emits PLAN_ESTIMATION_COMPLETE
+    Given an action with estimation_actor configured for r4
+    And a plan in strategize-complete state for r4
+    And the plan lifecycle service has a recording event bus for r4
+    When I execute the plan for r4
+    Then the event bus should have recorded PLAN_ESTIMATION_COMPLETE event
+    And the plan should be in execute phase
+
+  Scenario: execute_plan with estimation_actor catches event_bus emit failure for PLAN_ESTIMATION_COMPLETE
+    Given an action with estimation_actor configured for r4
+    And a plan in strategize-complete state for r4
+    And the plan lifecycle service has a failing event bus for r4
+    When I execute the plan for r4
+    Then the plan should be in execute phase despite estimation emit failure
+
+  # ---------------------------------------------------------------
+  # _run_estimation — cost_estimate_usd assignment (line 376)
+  # ---------------------------------------------------------------
+
+  Scenario: execute_plan populates cost_estimate_usd when estimation returns cost
+    Given an action with estimation_actor returning cost for r4
+    And a plan in strategize-complete state for r4
+    And the plan lifecycle service has a recording event bus for r4
+    When I execute the plan for r4 with mocked estimation
+    Then the plan cost_estimate_usd should be populated
+
+  # ---------------------------------------------------------------
+  # execute_plan — lock acquire/release when lock_service is configured
+  # Lines 1616-1621 (acquire) and 1684-1688 (finally release)
+  # ---------------------------------------------------------------
+
+  Scenario: execute_plan acquires and releases plan lock when lock_service is configured
+    Given an action for execute with lock service for r4
+    And a plan in strategize-complete state for r4
+    When I execute the plan with lock service for r4
+    Then the plan should be in execute phase
+    And the lock service should have acquired and released the plan lock
+
+  # ---------------------------------------------------------------
+  # apply_plan — lock acquire/release when lock_service is configured
+  # Lines 1830-1835 (acquire) and 1872-1877 (finally release)
+  # ---------------------------------------------------------------
+
+  Scenario: apply_plan acquires and releases plan lock when lock_service is configured
+    Given an action for apply with lock service for r4
+    And a plan in execute-complete state for r4
+    When I apply the plan with lock service for r4
+    Then the plan should be in apply phase
+    And the lock service should have acquired and released the plan lock
+
+  # ---------------------------------------------------------------
+  # _subscribe_correction_reconciliation — event_bus.subscribe failure
+  # Lines 562-566
+  # ---------------------------------------------------------------
+
+  Scenario: lifecycle service handles event_bus subscribe failure gracefully
+    Given I have a plan lifecycle service with event_bus that fails on subscribe for r4
+    When the service is created for r4
+    Then no exception should be raised for r4
+
+  # ---------------------------------------------------------------
+  # prompt_plan — validation scenarios
+  # Lines 2082 (blank guidance) and 2086-2097 (wrong phase/state)
+  # ---------------------------------------------------------------
+
+  Scenario: prompt_plan rejects blank guidance with ValidationError
+    Given a plan in execute phase with processing state for prompt_plan r4
+    When I prompt the plan with blank guidance for r4
+    Then a ValidationError should be raised for blank guidance
+
+  Scenario: prompt_plan rejects plan not in execute phase
+    Given a plan in strategize phase for prompt_plan r4
+    When I prompt the plan with guidance for wrong phase for r4
+    Then a PlanError should be raised for wrong phase
+
+  Scenario: prompt_plan rejects plan in complete state
+    Given a plan in execute-complete state for prompt_plan r4
+    When I prompt the plan with guidance for complete state for r4
+    Then a PlanError should be raised for wrong state
+
+  # ---------------------------------------------------------------
+  # prompt_plan — state transition and decision_service interaction
+  # Lines 2100-2154
+  # ---------------------------------------------------------------
+
+  Scenario: prompt_plan transitions from errored to processing and clears error_message
+    Given a plan in execute phase with errored state for prompt_plan r4
+    When I prompt the plan with guidance to resume from errored state for r4
+    Then the plan processing_state should be processing
+    And the plan error_message should be cleared
+
+  Scenario: prompt_plan handles decision_service exception gracefully
+    Given a plan in execute phase with processing state for prompt_plan r4
+    And the plan lifecycle service has a failing decision service for r4
+    When I prompt the plan with guidance for r4
+    Then the plan processing_state should be processing
+    And no exception should propagate
+
+  # ---------------------------------------------------------------
+  # _run_invariant_reconciliation — event_bus.emit exceptions
+  # Lines 499-505 (success emit failure) and 530-536 (failure emit failure)
+  # ---------------------------------------------------------------
+
+  Scenario: _run_invariant_reconciliation success path catches event_bus emit failure
+    Given an action with invariant_actor for reconciliation success r4
+    And a plan in strategize-complete state with project links for r4
+    And the plan lifecycle service has a failing event bus for r4
+    When I execute the plan for invariant reconciliation emit failure r4
+    Then the plan should be in execute phase
+    And the event_bus should have been called for INVARIANT_RECONCILED emit failure
+
+  Scenario: _run_invariant_reconciliation failure path catches event_bus emit failure
+    Given an action with invariant_actor causing reconciliation failure for r4
+    And a plan in strategize-complete state with project links for r4
+    And the plan lifecycle service has a failing event bus for r4
+    When I execute the plan expecting reconciliation failure for r4
+    Then ReconciliationBlockedError should be raised
+    And the event_bus should have been called for INVARIANT_VIOLATED emit failure
+
+  # ---------------------------------------------------------------
+  # Unknown automation profile validation
+  # Lines 1229-1233
+  # ---------------------------------------------------------------
+
+  Scenario: use_action rejects unknown automation profile with ValidationError
+    Given an action with unknown automation profile for r4
+    When I use the action to create a plan for r4
+    Then a ValidationError should be raised for unknown profile
+
+  # ---------------------------------------------------------------
+  # _handle_correction_applied — event handler scenarios
+  # Lines 577 (plan_id None guard), 584-595 (exception handlers)
+  # ---------------------------------------------------------------
+
+Scenario: _handle_correction_applied skips event with no plan_id
+    Given the plan lifecycle service has a recording event bus for correction handler r4
+    When I handle a CORRECTION_APPLIED event with no plan_id for r4
+    Then no exception should be raised for r4
+
+ Scenario: _handle_correction_applied handles ReconciliationBlockedError gracefully
+    Given a plan in strategize phase with processing state for r4
+    And the invariant reconciliation raises ReconciliationBlockedError for r4
+    When I handle a CORRECTION_APPLIED event for blocked reconciliation for r4
+    Then no exception should be raised for r4
+
+ Scenario: _handle_correction_applied handles general exception from get_plan gracefully
+    Given the plan lifecycle service is configured for r4
+    And get_plan will fail for the plan for r4
+    When I handle a CORRECTION_APPLIED event when get_plan fails for r4
+    Then no exception should be raised for r4
@@ -52,6 +52,7 @@ def _profile_name_for_level(level: str) -> str:
        "manual": "manual",
        "review_before_apply": "auto",
        "full_automation": "full-auto",
+        "cautious": "cautious",
    }
    return mapping.get(level, level)

@@ -0,0 +1,455 @@
+"""Step definitions for plan_lifecycle_service_coverage_boost_r3.feature.
+
+Targets remaining uncovered lines in PlanLifecycleService:
+- Lines ~1520-1540: complete_strategize event_bus.emit success + exception
+- Lines ~1569-1581: fail_strategize event_bus.emit success + exception
+- Lines ~1739-1759: complete_execute event_bus.emit success + exception
+- Lines ~1774-1792: fail_execute event_bus.emit success + exception
+- Lines ~2036-2054: fail_apply event_bus.emit success + exception
+- Lines ~2056-2057: fail_execute _cleanup_devcontainers call
+- Lines ~2056-2057: fail_apply _cleanup_devcontainers call
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.core.plan import (
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.infrastructure.events.types import EventType
+
+
+# -----------------------------------------------------------------
+# Background
+# -----------------------------------------------------------------
+
+
+@given("I have a fresh plan lifecycle service for coverage boost r3")
+def step_create_fresh_service_r3(context: Context) -> None:
+    """Create a clean PlanLifecycleService instance."""
+    Settings._instance = None
+    settings = Settings()
+    context.service = PlanLifecycleService(settings=settings)
+    context.error = None
+
+
+# -----------------------------------------------------------------
+# Helpers
+# -----------------------------------------------------------------
+
+
+def _create_action_r3(context: Context, name: str, **kwargs):
+    """Helper to create a basic action with sensible defaults."""
+    defaults = {
+        "name": name,
+        "description": f"Action {name}",
+        "definition_of_done": "Tests pass",
+        "strategy_actor": "openai/gpt-4",
+        "execution_actor": "openai/gpt-4",
+    }
+    defaults.update(kwargs)
+    return context.service.create_action(**defaults)
+
+
+def _advance_to_state(
+    context: Context, target_phase: PlanPhase, target_state: ProcessingState
+):
+    """Advance the current plan to the target phase and state."""
+    pid = context.plan.identity.plan_id
+    if (
+        target_phase == PlanPhase.STRATEGIZE
+        and target_state == ProcessingState.PROCESSING
+    ):
+        context.service.start_strategize(pid)
+        context.plan = context.service.get_plan(pid)
+    elif (
+        target_phase == PlanPhase.EXECUTE and target_state == ProcessingState.PROCESSING
+    ):
+        context.service.start_strategize(pid)
+        context.service.complete_strategize(pid)
+        context.service.execute_plan(pid)
+        context.service.start_execute(pid)
+        context.plan = context.service.get_plan(pid)
+    elif target_phase == PlanPhase.APPLY and target_state == ProcessingState.PROCESSING:
+        context.service.start_strategize(pid)
+        context.service.complete_strategize(pid)
+        context.service.execute_plan(pid)
+        context.service.start_execute(pid)
+        context.service.complete_execute(pid)
+        context.service.apply_plan(pid)
+        context.service.start_apply(pid)
+        context.plan = context.service.get_plan(pid)
+
+
+class RecordingEventBus:
+    """A simple event bus that records emitted events."""
+
+    def __init__(self):
+        self.events = []
+
+    def emit(self, event):
+        self.events.append(event)
+
+    def subscribe(self, event_type, handler):
+        pass
+
+
+class FailingEventBus:
+    """An event bus that always raises on emit."""
+
+    def emit(self, event):
+        raise RuntimeError("Simulated event bus failure")
+
+    def subscribe(self, event_type, handler):
+        pass
+
+
+# =================================================================
+# complete_strategize — event_bus.emit success
+# =================================================================
+
+
+@given("a plan lifecycle service with a recording event bus for r3")
+def step_service_with_recording_bus_r3(context: Context) -> None:
+    """Create a service with a recording event bus."""
+    Settings._instance = None
+    settings = Settings()
+    context.recording_bus = RecordingEventBus()
+    context.service = PlanLifecycleService(
+        settings=settings,
+        event_bus=context.recording_bus,
+    )
+    context.error = None
+
+
+@given('an action "{name}" exists for coverage boost r3')
+def step_create_action_r3(context: Context, name: str) -> None:
+    """Create an action with the given name."""
+    context.action = _create_action_r3(context, name)
+
+
+@given("a plan in strategize phase with processing state for r3")
+def step_plan_strategize_processing_r3(context: Context) -> None:
+    """Create a plan in Strategize/PROCESSING state."""
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r3")],
+    )
+    _advance_to_state(context, PlanPhase.STRATEGIZE, ProcessingState.PROCESSING)
+
+
+@when("I complete strategize on the plan with event bus")
+def step_complete_strategize_with_bus_r3(context: Context) -> None:
+    """Complete strategize — should emit PLAN_STATE_CHANGED event."""
+    context.error = None
+    try:
+        context.plan = context.service.complete_strategize(
+            context.plan.identity.plan_id
+        )
+    except Exception as e:
+        context.error = e
+
+
+@then("the event bus should have recorded a PLAN_STATE_CHANGED event")
+def step_verify_plan_state_changed_event_r3(context: Context) -> None:
+    """Verify PLAN_STATE_CHANGED was emitted."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    state_events = [
+        e
+        for e in context.recording_bus.events
+        if e.event_type == EventType.PLAN_STATE_CHANGED
+    ]
+    assert len(state_events) >= 1, (
+        f"Expected at least 1 PLAN_STATE_CHANGED event, got {len(state_events)}. "
+        f"All events: {[e.event_type for e in context.recording_bus.events]}"
+    )
+
+
+# =================================================================
+# complete_strategize — event_bus.emit exception
+# =================================================================
+
+
+@given("a plan lifecycle service with a failing event bus for r3")
+def step_service_with_failing_bus_r3(context: Context) -> None:
+    """Create a service with an event bus that raises on emit."""
+    Settings._instance = None
+    settings = Settings()
+    context.service = PlanLifecycleService(
+        settings=settings,
+        event_bus=FailingEventBus(),
+    )
+    context.error = None
+
+
+@when("I complete strategize on the plan and event bus emit fails")
+def step_complete_strategize_failing_bus_r3(context: Context) -> None:
+    """Complete strategize — event bus will raise but plan should still complete."""
+    context.error = None
+    try:
+        context.plan = context.service.complete_strategize(
+            context.plan.identity.plan_id
+        )
+    except Exception as e:
+        context.error = e
+
+
+@then('the plan phase should be "{phase}" and state should be "{state}"')
+def step_verify_plan_phase_state_r3(context: Context, phase: str, state: str) -> None:
+    """Verify the plan phase and state."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.phase.value == phase, (
+        f"Expected phase '{phase}', got '{context.plan.phase.value}'"
+    )
+    assert context.plan.processing_state.value == state, (
+        f"Expected state '{state}', got '{context.plan.processing_state.value}'"
+    )
+
+
+# =================================================================
+# fail_strategize — event_bus.emit success
+# =================================================================
+
+
+@when('I fail strategize on the plan with error "{error}"')
+def step_fail_strategize_with_error_r3(context: Context, error: str) -> None:
+    """Fail strategize with an error message."""
+    context.error = None
+    try:
+        context.plan = context.service.fail_strategize(
+            context.plan.identity.plan_id, error
+        )
+    except Exception as e:
+        context.error = e
+
+
+@then("the event bus should have recorded a PLAN_ERRORED event")
+def step_verify_plan_errored_event_r3(context: Context) -> None:
+    """Verify PLAN_ERRORED was emitted."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    error_events = [
+        e
+        for e in context.recording_bus.events
+        if e.event_type == EventType.PLAN_ERRORED
+    ]
+    assert len(error_events) >= 1, (
+        f"Expected at least 1 PLAN_ERRORED event, got {len(error_events)}. "
+        f"All events: {[e.event_type for e in context.recording_bus.events]}"
+    )
+
+
+# =================================================================
+# fail_strategize — event_bus.emit exception
+# =================================================================
+
+
+@when("I fail strategize on the plan and event bus emit fails")
+def step_fail_strategize_failing_bus_r3(context: Context) -> None:
+    """Fail strategize — event bus will raise but plan should still be errored."""
+    context.error = None
+    try:
+        context.plan = context.service.fail_strategize(
+            context.plan.identity.plan_id, "Test error"
+        )
+    except Exception as e:
+        context.error = e
+
+
+@then("the plan should be in errored state for r3")
+def step_verify_errored_state_r3(context: Context) -> None:
+    """Verify the plan is in ERRORED state."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.processing_state == ProcessingState.ERRORED, (
+        f"Expected ERRORED, got {context.plan.processing_state.value}"
+    )
+
+
+# =================================================================
+# complete_execute — event_bus.emit success
+# =================================================================
+
+
+@given("a plan in execute phase with processing state for r3")
+def step_plan_execute_processing_r3(context: Context) -> None:
+    """Create a plan in Execute/PROCESSING state."""
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r3")],
+    )
+    _advance_to_state(context, PlanPhase.EXECUTE, ProcessingState.PROCESSING)
+
+
+@when("I complete execute on the plan with event bus")
+def step_complete_execute_with_bus_r3(context: Context) -> None:
+    """Complete execute — should emit PLAN_STATE_CHANGED event."""
+    context.error = None
+    try:
+        context.plan = context.service.complete_execute(context.plan.identity.plan_id)
+    except Exception as e:
+        context.error = e
+
+
+# =================================================================
+# complete_execute — event_bus.emit exception
+# =================================================================
+
+
+@when("I complete execute on the plan and event bus emit fails")
+def step_complete_execute_failing_bus_r3(context: Context) -> None:
+    """Complete execute — event bus will raise but plan should still complete."""
+    context.error = None
+    try:
+        context.plan = context.service.complete_execute(context.plan.identity.plan_id)
+    except Exception as e:
+        context.error = e
+
+
+# =================================================================
+# fail_execute — event_bus.emit success
+# =================================================================
+
+
+@when('I fail execute on the plan with error "{error}"')
+def step_fail_execute_with_error_r3(context: Context, error: str) -> None:
+    """Fail execute with an error message."""
+    context.error = None
+    try:
+        context.plan = context.service.fail_execute(
+            context.plan.identity.plan_id, error
+        )
+    except Exception as e:
+        context.error = e
+
+
+# =================================================================
+# fail_execute — event_bus.emit exception
+# =================================================================
+
+
+@when("I fail execute on the plan and event bus emit fails")
+def step_fail_execute_failing_bus_r3(context: Context) -> None:
+    """Fail execute — event bus will raise but plan should still be errored."""
+    context.error = None
+    try:
+        context.plan = context.service.fail_execute(
+            context.plan.identity.plan_id, "Test error"
+        )
+    except Exception as e:
+        context.error = e
+
+
+# =================================================================
+# fail_apply — event_bus.emit success
+# =================================================================
+
+
+@given("a plan in apply phase with processing state for r3")
+def step_plan_apply_processing_r3(context: Context) -> None:
+    """Create a plan in Apply/PROCESSING state."""
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r3")],
+    )
+    _advance_to_state(context, PlanPhase.APPLY, ProcessingState.PROCESSING)
+
+
+@when('I fail apply on the plan with error "{error}"')
+def step_fail_apply_with_error_r3(context: Context, error: str) -> None:
+    """Fail apply with an error message."""
+    context.error = None
+    try:
+        context.plan = context.service.fail_apply(context.plan.identity.plan_id, error)
+    except Exception as e:
+        context.error = e
+
+
+# =================================================================
+# fail_apply — event_bus.emit exception
+# =================================================================
+
+
+@when("I fail apply on the plan and event bus emit fails")
+def step_fail_apply_failing_bus_r3(context: Context) -> None:
+    """Fail apply — event bus will raise but plan should still be errored."""
+    context.error = None
+    try:
+        context.plan = context.service.fail_apply(
+            context.plan.identity.plan_id, "Test error"
+        )
+    except Exception as e:
+        context.error = e
+
+
+# =================================================================
+# fail_execute — _cleanup_devcontainers success
+# =================================================================
+
+
+@given("a plan lifecycle service with mocked devcontainer cleanup for r3")
+def step_service_with_mocked_cleanup_r3(context: Context) -> None:
+    """Create a service where devcontainer cleanup returns stopped containers."""
+    Settings._instance = None
+    settings = Settings()
+    context.service = PlanLifecycleService(settings=settings)
+    context.cleanup_called = False
+    context.error = None
+
+
+@when('I fail execute on the plan with error "{error}" and cleanup should be called')
+def step_fail_execute_with_cleanup_r3(context: Context, error: str) -> None:
+    """Fail execute with mocked cleanup."""
+    context.error = None
+    context.cleanup_called = False
+    try:
+        with patch(
+            "cleveragents.application.services.cleanup_service.stop_all_active_containers",
+            return_value=["container-1", "container-2"],
+        ):
+            context.plan = context.service.fail_execute(
+                context.plan.identity.plan_id, error
+            )
+            context.cleanup_called = True
+    except Exception as e:
+        context.error = e
+
+
+@then("the plan should be in errored state and cleanup should have been called")
+def step_verify_errored_with_cleanup_r3(context: Context) -> None:
+    """Verify the plan is ERRORED and cleanup was invoked."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.processing_state == ProcessingState.ERRORED, (
+        f"Expected ERRORED, got {context.plan.processing_state.value}"
+    )
+
+
+# =================================================================
+# fail_apply — _cleanup_devcontainers success
+# =================================================================
+
+
+@when('I fail apply on the plan with error "{error}" and cleanup should be called')
+def step_fail_apply_with_cleanup_r3(context: Context, error: str) -> None:
+    """Fail apply with mocked cleanup."""
+    context.error = None
+    context.cleanup_called = False
+    try:
+        with patch(
+            "cleveragents.application.services.cleanup_service.stop_all_active_containers",
+            return_value=["container-1", "container-2"],
+        ):
+            context.plan = context.service.fail_apply(
+                context.plan.identity.plan_id, error
+            )
+            context.cleanup_called = True
+    except Exception as e:
+        context.error = e
@@ -0,0 +1,886 @@
+"""Step definitions for plan_lifecycle_service_coverage_boost_r4.feature.
+
+Targets remaining uncovered lines in PlanLifecycleService:
+- Lines ~386-405: _run_estimation event_bus.emit PLAN_ESTIMATION_COMPLETE success + exception
+- Line 376: cost_estimate_usd assignment when estimated_cost_usd is not None
+- Line 462: project_name extraction from plan.project_links[0]
+- Lines ~481-505: _run_invariant_reconciliation INVARIANT_RECONCILED emit exception
+- Lines ~517-536: _run_invariant_reconciliation INVARIANT_VIOLATED emit exception
+- Lines ~562-566: _subscribe_correction_reconciliation event_bus.subscribe failure
+- Line 577: _handle_correction_applied guard (plan_id is None)
+- Lines ~584-595: _handle_correction_applied ReconciliationBlockedError + general exception
+- Lines ~1230-1232: unknown automation profile ValidationError
+- Lines ~1616-1621, 1684-1688: execute_plan lock acquire/release with lock_service
+- Lines ~1830-1835, 1872-1877: apply_plan lock acquire/release with lock_service
+- Lines ~2082-2097: prompt_plan blank guidance + wrong phase/state validation
+- Lines ~2100-2156: prompt_plan state transition + decision_service interaction
+"""
+
+from __future__ import annotations
+
+from typing import Any
+from unittest.mock import patch
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+    ReconciliationBlockedError,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.core.exceptions import PlanError, ValidationError
+from cleveragents.domain.models.core.action import Action
+from cleveragents.domain.models.core.estimation import EstimationResult
+from cleveragents.domain.models.core.plan import (
+    PlanPhase,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.infrastructure.events.models import DomainEvent
+from cleveragents.infrastructure.events.types import EventType
+
+
+# -----------------------------------------------------------------
+# Background
+# -----------------------------------------------------------------
+
+
+@given("I have a fresh plan lifecycle service for coverage boost r4")
+def step_create_fresh_service_r4(context: Context) -> None:
+    """Create a clean PlanLifecycleService instance."""
+    Settings._instance = None
+    settings = Settings()
+    context.service = PlanLifecycleService(settings=settings)
+    context.error = None
+
+
+# -----------------------------------------------------------------
+# Helpers
+# -----------------------------------------------------------------
+
+
+def _create_action_r4(context: Context, name: str, **kwargs) -> Action:
+    """Helper to create a basic action with sensible defaults."""
+    defaults: dict[str, Any] = {
+        "name": name,
+        "description": f"Action {name}",
+        "definition_of_done": "Tests pass",
+        "strategy_actor": "openai/gpt-4",
+        "execution_actor": "openai/gpt-4",
+    }
+    defaults.update(kwargs)
+    return context.service.create_action(**defaults)
+
+
+def _advance_to_strategize_complete(context: Context) -> None:
+    """Advance plan to strategize-complete state."""
+    pid = context.plan.identity.plan_id
+    context.service.start_strategize(pid)
+    context.service.complete_strategize(pid)
+    context.plan = context.service.get_plan(pid)
+
+
+def _advance_to_execute_complete(context: Context) -> None:
+    """Advance plan to execute-complete state (ready for apply)."""
+    pid = context.plan.identity.plan_id
+    context.service.start_strategize(pid)
+    context.service.complete_strategize(pid)
+    context.service.execute_plan(pid)
+    context.service.start_execute(pid)
+    context.service.complete_execute(pid)
+    context.plan = context.service.get_plan(pid)
+
+
+def _advance_to_strategize_processing(context: Context) -> None:
+    """Advance plan to strategize/PROCESSING state."""
+    pid = context.plan.identity.plan_id
+    context.service.start_strategize(pid)
+    context.plan = context.service.get_plan(pid)
+
+
+# -----------------------------------------------------------------
+# Mock event buses
+# -----------------------------------------------------------------
+
+
+class FailingEventBusR4:
+    """An event bus that raises on every emit call."""
+
+    def emit(self, event: DomainEvent) -> None:
+        raise RuntimeError("Simulated event bus emit failure")
+
+    def subscribe(self, event_type: Any, handler: Any) -> None:
+        pass
+
+
+class FailingSubscribeEventBus:
+    """An event bus that raises on subscribe but works for emit."""
+
+    def emit(self, event: DomainEvent) -> None:
+        pass
+
+    def subscribe(self, event_type: Any, handler: Any) -> None:
+        raise RuntimeError("Simulated subscribe failure")
+
+
+class RecordingEventBusR4:
+    """A simple event bus that records emitted events."""
+
+    def __init__(self) -> None:
+        self.events: list[DomainEvent] = []
+
+    def emit(self, event: DomainEvent) -> None:
+        self.events.append(event)
+
+    def subscribe(self, event_type: Any, handler: Any) -> None:
+        pass
+
+
+class FailingDecisionService:
+    """A decision service that raises on all operations."""
+
+    def list_decisions(self, plan_id: str) -> list[Any]:
+        raise RuntimeError("Simulated decision service failure")
+
+    def record_decision(self, **kwargs: Any) -> Any:
+        raise RuntimeError("Simulated decision service failure")
+
+
+# =================================================================
+# _run_estimation via execute_plan — event_bus.emit PLAN_ESTIMATION_COMPLETE
+# =================================================================
+
+
+@given("an action with estimation_actor configured for r4")
+def step_action_with_estimation_actor_r4(context: Context) -> None:
+    """Create an action with estimation_actor set."""
+    context.action = _create_action_r4(
+        context,
+        "local/est-action-r4",
+        estimation_actor="openai/gpt-4",
+    )
+
+
+@given("an action with estimation_actor returning cost for r4")
+def step_action_with_estimation_cost_r4(context: Context) -> None:
+    """Create an action with estimation_actor and patch stub to return cost."""
+    context.action = _create_action_r4(
+        context,
+        "local/est-cost-action-r4",
+        estimation_actor="openai/gpt-4",
+    )
+    context.mock_estimation_result = EstimationResult(
+        summary="Test estimation with cost",
+        estimated_cost_usd=1.23,
+    )
+    context.estimation_patch = patch(
+        "cleveragents.application.services.plan_executor.EstimationStubActor.estimate",
+        return_value=context.mock_estimation_result,
+    )
+
+
+@given("a plan in strategize-complete state for r4")
+def step_plan_strategize_complete_r4(context: Context) -> None:
+    """Create a plan in strategize/COMPLETE state."""
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r4")],
+    )
+    _advance_to_strategize_complete(context)
+
+
+@given("a plan in strategize phase with processing state for r4")
+def step_plan_strategize_processing_r4(context: Context) -> None:
+    """Create a plan in strategize/PROCESSING state."""
+    if not hasattr(context, "action") or context.action is None:
+        context.action = _create_action_r4(
+            context, f"local/corr-handler-r4-{id(context)}"
+        )
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r4")],
+    )
+    _advance_to_strategize_processing(context)
+
+
+@given("the plan lifecycle service has a recording event bus for r4")
+def step_service_with_recording_bus_r4(context: Context) -> None:
+    """Set up recording event bus on existing service."""
+    context.recording_bus_r4 = RecordingEventBusR4()
+    context.service.event_bus = context.recording_bus_r4
+
+
+@given("the plan lifecycle service has a failing event bus for r4")
+def step_service_with_failing_bus_r4(context: Context) -> None:
+    """Set up failing event bus on existing service."""
+    context.service.event_bus = FailingEventBusR4()
+
+
+@when("I execute the plan for r4")
+def step_execute_plan_r4(context: Context) -> None:
+    """Call execute_plan on the current plan."""
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.plan = context.service.execute_plan(pid)
+    except Exception as e:
+        context.error = e
+
+
+@then("the event bus should have recorded PLAN_ESTIMATION_COMPLETE event")
+def step_verify_estimation_complete_event_r4(context: Context) -> None:
+    """Verify PLAN_ESTIMATION_COMPLETE was emitted."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    est_events = [
+        e
+        for e in context.recording_bus_r4.events
+        if e.event_type == EventType.PLAN_ESTIMATION_COMPLETE
+    ]
+    assert len(est_events) >= 1, (
+        f"Expected at least 1 PLAN_ESTIMATION_COMPLETE event, "
+        f"got {len(est_events)}. Events: {[e.event_type for e in context.recording_bus_r4.events]}"
+    )
+
+
+@then("the plan should be in execute phase despite estimation emit failure")
+def step_plan_in_executeDespiteEstimationEmitFailure_r4(context: Context) -> None:
+    """Verify plan is in EXECUTE even when estimation emit failed."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.phase == PlanPhase.EXECUTE, (
+        f"Expected EXECUTE phase, got {context.plan.phase}"
+    )
+
+
+@then("the plan should be in execute phase")
+def step_plan_in_execute_phase_r4(context: Context) -> None:
+    """Verify plan is in EXECUTE phase."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.phase == PlanPhase.EXECUTE, (
+        f"Expected EXECUTE phase, got {context.plan.phase}"
+    )
+
+
+@then("the plan cost_estimate_usd should be populated")
+def step_verify_cost_estimate_populated_r4(context: Context) -> None:
+    """Verify plan.cost_estimate_usd was set from estimation result."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.cost_estimate_usd is not None, (
+        "Expected cost_estimate_usd to be set, got None"
+    )
+
+
+# =================================================================
+# execute_plan with mocked EstimationStubActor for cost_estimate_usd
+# =================================================================
+
+
+@when("I execute the plan for r4 with mocked estimation")
+def step_execute_plan_with_mocked_estimation_r4(context: Context) -> None:
+    """Execute plan with EstimationStubActor patched to return cost."""
+    mock_result = context.mock_estimation_result
+    with patch(
+        "cleveragents.application.services.plan_executor.EstimationStubActor.estimate",
+        return_value=mock_result,
+    ):
+        context.error = None
+        try:
+            pid = context.plan.identity.plan_id
+            context.plan = context.service.execute_plan(pid)
+        except Exception as e:
+            context.error = e
+
+
+# =================================================================
+# execute_plan with lock_service
+# =================================================================
+
+
+class MockLockService:
+    """A mock lock service that tracks acquire/release calls."""
+
+    def __init__(self) -> None:
+        self.acquired: list[tuple[str, str, str]] = []
+        self.released: list[tuple[str, str, str]] = []
+
+    def acquire(self, owner_id: str, resource_type: str, resource_id: str) -> None:
+        self.acquired.append((owner_id, resource_type, resource_id))
+
+    def release(self, owner_id: str, resource_type: str, resource_id: str) -> None:
+        self.released.append((owner_id, resource_type, resource_id))
+
+
+@given("an action for execute with lock service for r4")
+def step_action_execute_lock_r4(context: Context) -> None:
+    """Create an action for testing execute_plan with lock_service."""
+    context.action = _create_action_r4(context, "local/exec-lock-r4")
+
+
+@when("I execute the plan with lock service for r4")
+def step_execute_with_lock_r4(context: Context) -> None:
+    """Execute plan with a mocked lock service."""
+    context.mock_lock_service = MockLockService()
+    context.service._lock_service = context.mock_lock_service
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.plan = context.service.execute_plan(pid)
+    except Exception as e:
+        context.error = e
+
+
+@then("the lock service should have acquired and released the plan lock")
+def step_verify_lock_acquire_release_r4(context: Context) -> None:
+    """Verify lock was acquired and released for execute_plan."""
+    assert hasattr(context, "mock_lock_service"), "mock_lock_service not found"
+    ls = context.mock_lock_service
+    assert len(ls.acquired) >= 1, (
+        f"Expected at least 1 acquire call, got {len(ls.acquired)}"
+    )
+    assert len(ls.released) >= 1, (
+        f"Expected at least 1 release call, got {len(ls.released)}"
+    )
+    plan_id = context.plan.identity.plan_id
+    for call in ls.acquired:
+        assert call[1] == "plan", f"Expected resource_type 'plan', got {call[1]}"
+        assert call[2] == plan_id, f"Expected resource_id {plan_id}, got {call[2]}"
+
+
+# =================================================================
+# apply_plan with lock_service
+# =================================================================
+
+
+@given("an action for apply with lock service for r4")
+def step_action_apply_lock_r4(context: Context) -> None:
+    """Create an action for testing apply_plan with lock_service."""
+    context.action = _create_action_r4(context, "local/apply-lock-r4")
+
+
+@given("a plan in execute-complete state for r4")
+def step_plan_execute_complete_r4(context: Context) -> None:
+    """Create a plan in execute/COMPLETE state (ready for apply)."""
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r4")],
+    )
+    _advance_to_execute_complete(context)
+
+
+@when("I apply the plan with lock service for r4")
+def step_apply_with_lock_r4(context: Context) -> None:
+    """Apply plan with a mocked lock service."""
+    context.mock_lock_service = MockLockService()
+    context.service._lock_service = context.mock_lock_service
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.plan = context.service.apply_plan(pid)
+    except Exception as e:
+        context.error = e
+
+
+@then("the plan should be in apply phase")
+def step_plan_in_apply_phase_r4(context: Context) -> None:
+    """Verify plan is in APPLY phase."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.phase == PlanPhase.APPLY, (
+        f"Expected APPLY phase, got {context.plan.phase}"
+    )
+
+
+# =================================================================
+# _subscribe_correction_reconciliation — event_bus.subscribe failure
+# =================================================================
+
+
+@given("I have a plan lifecycle service with event_bus that fails on subscribe for r4")
+def step_service_with_failing_subscribe_r4(context: Context) -> None:
+    """Create service where event_bus.subscribe raises."""
+    Settings._instance = None
+    settings = Settings()
+    context.error = None
+    try:
+        context.service = PlanLifecycleService(
+            settings=settings,
+            event_bus=FailingSubscribeEventBus(),
+        )
+    except Exception as e:
+        context.error = e
+
+
+@when("the service is created for r4")
+def step_service_created_r4(context: Context) -> None:
+    """Service creation step (no-op since given already created it)."""
+    pass
+
+
+@then("no exception should be raised for r4")
+def step_no_exception_r4(context: Context) -> None:
+    """Verify no exception was raised."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+
+
+# =================================================================
+# prompt_plan validation scenarios
+# =================================================================
+
+
+def _advance_to_execute_processing(context: Context) -> None:
+    """Advance plan to execute/PROCESSING state."""
+    pid = context.plan.identity.plan_id
+    context.service.start_strategize(pid)
+    context.service.complete_strategize(pid)
+    context.service.execute_plan(pid)
+    context.service.start_execute(pid)
+    context.plan = context.service.get_plan(pid)
+
+
+def _advance_to_execute_complete_for_prompt(context: Context) -> None:
+    """Advance plan to execute/COMPLETE state."""
+    pid = context.plan.identity.plan_id
+    context.service.start_strategize(pid)
+    context.service.complete_strategize(pid)
+    context.service.execute_plan(pid)
+    context.service.start_execute(pid)
+    context.service.complete_execute(pid)
+    context.plan = context.service.get_plan(pid)
+
+
+@given("a plan in execute phase with processing state for prompt_plan r4")
+def step_plan_execute_processing_for_prompt_r4(context: Context) -> None:
+    """Create a plan in execute/PROCESSING state for prompt_plan testing."""
+    context.action = _create_action_r4(context, "local/prompt-test-r4")
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r4")],
+    )
+    _advance_to_execute_processing(context)
+
+
+@given("a plan in strategize phase for prompt_plan r4")
+def step_plan_strategize_for_prompt_r4(context: Context) -> None:
+    """Create a plan in strategize/PROCESSING state for prompt_plan testing."""
+    context.action = _create_action_r4(context, "local/prompt-wrong-phase-r4")
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r4")],
+    )
+    _advance_to_state_prompt(context, PlanPhase.STRATEGIZE, ProcessingState.PROCESSING)
+
+
+@given("a plan in execute-complete state for prompt_plan r4")
+def step_plan_execute_complete_for_prompt_r4(context: Context) -> None:
+    """Create a plan in execute/COMPLETE state for prompt_plan testing."""
+    context.action = _create_action_r4(context, "local/prompt-complete-state-r4")
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r4")],
+    )
+    _advance_to_execute_complete_for_prompt(context)
+
+
+@given("a plan in execute phase with errored state for prompt_plan r4")
+def step_plan_execute_errored_for_prompt_r4(context: Context) -> None:
+    """Create a plan in execute/ERRORED state for prompt_plan testing."""
+    context.action = _create_action_r4(context, "local/prompt-errored-r4")
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="proj-r4")],
+    )
+    _advance_to_execute_processing(context)
+    pid = context.plan.identity.plan_id
+    context.service.fail_execute(pid, error_message="Prior test error")
+    context.plan = context.service.get_plan(pid)
+
+
+def _advance_to_state_prompt(
+    context: Context, target_phase: PlanPhase, target_state: ProcessingState
+) -> None:
+    """Advance plan to target phase/state."""
+    pid = context.plan.identity.plan_id
+    if (
+        target_phase == PlanPhase.STRATEGIZE
+        and target_state == ProcessingState.PROCESSING
+    ):
+        context.service.start_strategize(pid)
+        context.plan = context.service.get_plan(pid)
+    elif (
+        target_phase == PlanPhase.EXECUTE and target_state == ProcessingState.PROCESSING
+    ):
+        context.service.start_strategize(pid)
+        context.service.complete_strategize(pid)
+        context.service.execute_plan(pid)
+        context.service.start_execute(pid)
+        context.plan = context.service.get_plan(pid)
+
+
+@when("I prompt the plan with blank guidance for r4")
+def step_prompt_blank_guidance_r4(context: Context) -> None:
+    """Call prompt_plan with blank guidance."""
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.result = context.service.prompt_plan(pid, guidance="   ")
+    except Exception as e:
+        context.error = e
+
+
+@then("a ValidationError should be raised for blank guidance")
+def step_verify_validation_error_r4(context: Context) -> None:
+    """Verify ValidationError was raised for blank guidance."""
+    assert isinstance(context.error, ValidationError), (
+        f"Expected ValidationError, got {type(context.error).__name__}: {context.error}"
+    )
+
+
+@when("I prompt the plan with guidance for wrong phase for r4")
+def step_prompt_wrong_phase_r4(context: Context) -> None:
+    """Call prompt_plan when plan is not in Execute phase."""
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.result = context.service.prompt_plan(pid, guidance="Do something")
+    except Exception as e:
+        context.error = e
+
+
+@then("a PlanError should be raised for wrong phase")
+def step_verify_plan_error_wrong_phase_r4(context: Context) -> None:
+    """Verify PlanError was raised for wrong phase."""
+    assert isinstance(context.error, PlanError), (
+        f"Expected PlanError, got {type(context.error).__name__}: {context.error}"
+    )
+
+
+@when("I prompt the plan with guidance for complete state for r4")
+def step_prompt_wrong_state_r4(context: Context) -> None:
+    """Call prompt_plan when plan is in COMPLETE state (not recoverable)."""
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.result = context.service.prompt_plan(pid, guidance="Resume work")
+    except Exception as e:
+        context.error = e
+
+
+@then("a PlanError should be raised for wrong state")
+def step_verify_plan_error_wrong_state_r4(context: Context) -> None:
+    """Verify PlanError was raised for wrong state."""
+    assert isinstance(context.error, PlanError), (
+        f"Expected PlanError, got {type(context.error).__name__}: {context.error}"
+    )
+
+
+@when("I prompt the plan with guidance to resume from errored state for r4")
+def step_prompt_errored_resume_r4(context: Context) -> None:
+    """Call prompt_plan to resume from errored state."""
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.result = context.service.prompt_plan(pid, guidance="Please recover")
+        context.plan = context.service.get_plan(pid)
+    except Exception as e:
+        context.error = e
+
+
+@then("the plan processing_state should be processing")
+def step_verify_processing_state_r4(context: Context) -> None:
+    """Verify plan.processing_state is PROCESSING."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.processing_state == ProcessingState.PROCESSING, (
+        f"Expected PROCESSING state, got {context.plan.processing_state}"
+    )
+
+
+@then("the plan error_message should be cleared")
+def step_verify_error_cleared_r4(context: Context) -> None:
+    """Verify plan.error_message was cleared after resuming from errored."""
+    assert context.plan.error_message is None, (
+        f"Expected error_message to be cleared, got: {context.plan.error_message}"
+    )
+
+
+@given("the plan lifecycle service has a failing decision service for r4")
+def step_failing_decision_service_r4(context: Context) -> None:
+    """Replace decision_service with one that always fails."""
+    context.service.decision_service = FailingDecisionService()
+
+
+@when("I prompt the plan with guidance for r4")
+def step_prompt_with_failing_decision_service_r4(context: Context) -> None:
+    """Call prompt_plan when decision_service raises."""
+    context.error = None
+    try:
+        pid = context.plan.identity.plan_id
+        context.result = context.service.prompt_plan(pid, guidance="Continue anyway")
+        context.plan = context.service.get_plan(pid)
+    except Exception as e:
+        context.error = e
+
+
+# =================================================================
+# _run_invariant_reconciliation — event_bus.emit exception scenarios
+# =================================================================
+
+
+@given("an action with invariant_actor for reconciliation success r4")
+def step_action_invariant_success_r4(context: Context) -> None:
+    """Create action with invariant_actor set for successful reconciliation."""
+    context.action = _create_action_r4(
+        context,
+        "local/inv-success-r4",
+        invariant_actor="openai/gpt-4",
+    )
+
+
+@given("a plan in strategize-complete state with project links for r4")
+def step_plan_strategize_complete_with_links_r4(context: Context) -> None:
+    """Create plan with project_links for testing project_name extraction."""
+    if (
+        not hasattr(context.service, "invariant_service")
+        or context.service.invariant_service is None
+    ):
+        from cleveragents.application.services.invariant_service import InvariantService
+        from cleveragents.application.services.decision_service import DecisionService
+
+        context.service.invariant_service = InvariantService()
+        context.service.decision_service = DecisionService()
+    context.plan = context.service.use_action(
+        action_name=str(context.action.namespaced_name),
+        project_links=[ProjectLink(project_name="test-project-r4")],
+    )
+    _advance_to_strategize_complete(context)
+
+
+@when("I execute the plan for invariant reconciliation emit failure r4")
+def step_execute_invariant_emit_fail_r4(context: Context) -> None:
+    """Execute plan with mocked reconciliation actor (succeeds) and failing emit."""
+    from cleveragents.actor.reconciliation import ReconciliationResult
+    from cleveragents.actor.reconciliation import InvariantReconciliationActor
+    from cleveragents.domain.models.core.invariant import InvariantSet
+
+    context.error = None
+    mock_result = ReconciliationResult(
+        reconciled_set=InvariantSet(invariants=[]),
+        conflicts=[],
+        enforced_decision_ids=[],
+    )
+
+    class PatchedActor(InvariantReconciliationActor):
+        def run(self, **kwargs: Any) -> Any:
+            return mock_result
+
+    with patch(
+        "cleveragents.actor.reconciliation.InvariantReconciliationActor",
+        PatchedActor,
+    ):
+        try:
+            pid = context.plan.identity.plan_id
+            context.plan = context.service.execute_plan(pid)
+        except Exception as e:
+            context.error = e
+
+
+@then("the event_bus should have been called for INVARIANT_RECONCILED emit failure")
+def step_verify_invariant_reconciled_emit_fail_r4(context: Context) -> None:
+    """Verify plan reached EXECUTE despite emit failure in reconciliation."""
+    assert context.error is None, f"Expected no error, got: {context.error}"
+    assert context.plan.phase == PlanPhase.EXECUTE, (
+        f"Expected EXECUTE, got {context.plan.phase}"
+    )
+
+
+@given("an action with invariant_actor causing reconciliation failure for r4")
+def step_action_invariant_fail_r4(context: Context) -> None:
+    """Create action with invariant_actor that will cause failure."""
+    context.action = _create_action_r4(
+        context,
+        "local/inv-fail-r4",
+        invariant_actor="openai/gpt-4",
+    )
+
+
+@when("I execute the plan expecting reconciliation failure for r4")
+def step_execute_invariant_fail_r4(context: Context) -> None:
+    """Execute plan where reconciliation raises (triggers INVARIANT_VIOLATED emit fail path)."""
+    from cleveragents.actor.reconciliation import InvariantReconciliationActor
+
+    class FailingActor(InvariantReconciliationActor):
+        def run(self, **kwargs: Any) -> Any:
+            raise RuntimeError("Simulated invariant violation")
+
+    with patch(
+        "cleveragents.actor.reconciliation.InvariantReconciliationActor",
+        FailingActor,
+    ):
+        context.error = None
+        try:
+            pid = context.plan.identity.plan_id
+            context.plan = context.service.execute_plan(pid)
+        except ReconciliationBlockedError as e:
+            context.error = e
+        except Exception as e:
+            context.error = e
+
+
+@then("ReconciliationBlockedError should be raised")
+def step_verify_reconciliation_blocked_error_r4(context: Context) -> None:
+    """Verify ReconciliationBlockedError was raised."""
+    assert isinstance(context.error, ReconciliationBlockedError), (
+        f"Expected ReconciliationBlockedError, got {type(context.error).__name__}: {context.error}"
+    )
+
+
+@then("the event_bus should have been called for INVARIANT_VIOLATED emit failure")
+def step_verify_invariant_violated_emit_fail_r4(context: Context) -> None:
+    """Verify INVARIANT_VIOLATED emit exception was handled (error was still raised)."""
+    assert isinstance(context.error, ReconciliationBlockedError), (
+        "Expected ReconciliationBlockedError to be raised despite emit failure"
+    )
+
+
+# =================================================================
+# Unknown automation profile validation (lines 1230-1232)
+# =================================================================
+
+
+@given("an action with unknown automation profile for r4")
+def step_action_unknown_profile_r4(context: Context) -> None:
+    """Create action with an unknown automation profile name."""
+    context.action = _create_action_r4(
+        context,
+        "local/unknown-profile-r4",
+        automation_profile="nonexistent-profile-xyz",
+    )
+
+
+@when("I use the action to create a plan for r4")
+def step_use_action_unknown_profile_r4(context: Context) -> None:
+    """Try to use_action with unknown automation profile (raises during profile resolution)."""
+    context.error = None
+    try:
+        context.plan = context.service.use_action(
+            action_name=str(context.action.namespaced_name),
+            project_links=[ProjectLink(project_name="proj-r4")],
+        )
+    except ValidationError as e:
+        context.error = e
+    except Exception as e:
+        context.error = e
+
+
+@then("a ValidationError should be raised for unknown profile")
+def step_verify_validation_error_unknown_profile_r4(context: Context) -> None:
+    """Verify ValidationError was raised for unknown automation profile."""
+    assert isinstance(context.error, ValidationError), (
+        f"Expected ValidationError, got {type(context.error).__name__}: {context.error}"
+    )
+
+
+# =================================================================
+# _handle_correction_applied — event handler scenarios
+# =================================================================
+
+
+class MockEventBusForHandler:
+    """Event bus that records correction applied events for handler testing."""
+
+    def __init__(self) -> None:
+        self.events: list[DomainEvent] = []
+        self._handler: Any = None
+
+    def emit(self, event: DomainEvent) -> None:
+        self.events.append(event)
+
+    def subscribe(self, event_type: Any, handler: Any) -> None:
+        self._handler = handler
+
+
+@when("I handle a CORRECTION_APPLIED event with no plan_id for r4")
+def step_handle_correction_no_plan_id_r4(context: Context) -> None:
+    """Call _handle_correction_applied with an event that has no plan_id."""
+    context.error = None
+    event = DomainEvent(
+        event_type=EventType.CORRECTION_APPLIED,
+        plan_id=None,
+        details={},
+    )
+    try:
+        context.service._handle_correction_applied(event)
+    except Exception as e:
+        context.error = e
+
+
+@given("the plan lifecycle service has a recording event bus for correction handler r4")
+def step_service_recording_bus_for_handler_r4(context: Context) -> None:
+    """Set up service with recording event bus for _handle_correction_applied testing."""
+    Settings._instance = None
+    settings = Settings()
+    context.mock_bus_handler = MockEventBusForHandler()
+    context.service = PlanLifecycleService(
+        settings=settings, event_bus=context.mock_bus_handler
+    )
+
+
+@given("the plan lifecycle service is configured for r4")
+def step_service_configured_r4(context: Context) -> None:
+    """Ensure service is configured (reuse existing or create new)."""
+    if not hasattr(context, "service") or context.service is None:
+        Settings._instance = None
+        settings = Settings()
+        context.service = PlanLifecycleService(settings=settings)
+
+
+@given("the invariant reconciliation raises ReconciliationBlockedError for r4")
+def step_invariant_raises_blocked_error_r4(context: Context) -> None:
+    """Patch _run_invariant_reconciliation to raise ReconciliationBlockedError."""
+
+    def _patched_run_invariant(plan: Any) -> None:
+        raise ReconciliationBlockedError(
+            plan_id=plan.identity.plan_id,
+            phase=plan.phase,
+            reason="Simulated blocked error",
+        )
+
+    context.service._run_invariant_reconciliation = _patched_run_invariant
+
+
+@when("I handle a CORRECTION_APPLIED event for blocked reconciliation for r4")
+def step_handle_correction_blocked_r4(context: Context) -> None:
+    """Call _handle_correction_applied when reconciliation raises ReconciliationBlockedError."""
+    context.error = None
+    event = DomainEvent(
+        event_type=EventType.CORRECTION_APPLIED,
+        plan_id=context.plan.identity.plan_id,
+        details={},
+    )
+    try:
+        context.service._handle_correction_applied(event)
+    except Exception as e:
+        context.error = e
+
+
+@given("get_plan will fail for the plan for r4")
+def step_get_plan_fails_r4(context: Context) -> None:
+    """Patch get_plan to raise an exception."""
+
+    def _failing_get_plan(plan_id: str) -> Any:
+        raise RuntimeError("Simulated get_plan failure")
+
+    context.service.get_plan = _failing_get_plan
+
+
+@when("I handle a CORRECTION_APPLIED event when get_plan fails for r4")
+def step_handle_correction_get_plan_fails_r4(context: Context) -> None:
+    """Call _handle_correction_applied when get_plan raises."""
+    context.error = None
+    event = DomainEvent(
+        event_type=EventType.CORRECTION_APPLIED,
+        plan_id="nonexistent-plan-id",
+        details={},
+    )
+    try:
+        context.service._handle_correction_applied(event)
+    except Exception as e:
+        context.error = e
@@ -0,0 +1,127 @@
+@tdd_issue @tdd_issue_4328
+Feature: Automation Profile Gates - Issue #4328 Regression Tests
+  Regression tests for manual automation profile being disrespected during phase
+  completion. Verifies that complete_strategize() and complete_execute() respect
+  automation profile gates and do NOT auto-progress when thresholds are 1.0.
+
+  Background:
+    Given I have a plan lifecycle service with automation level support
+
+  # ========================================================================
+  # Issue #4328: Manual profile (all thresholds = 1.0) should NOT auto-progress
+  # These scenarios verify the fix - that complete_strategize/execute do NOT
+  # unconditionally call auto_progress()
+  # ========================================================================
+
+  Scenario: Manual profile create_tool=1.0 blocks strategize to execute transition
+    Given I have a plan with automation level "manual" in strategize phase with processing state
+    When I complete strategize on the automated plan
+    Then the automated plan phase should be "strategize"
+    And the automated plan processing state should be "complete"
+    And should_auto_progress should return false
+
+  Scenario: Manual profile select_tool=1.0 blocks execute to apply transition
+    Given I have a plan with automation level "manual" in execute phase with processing state
+    When I complete execute on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "complete"
+    And should_auto_progress should return false
+
+  # ========================================================================
+  # Verify auto_progress() IS called when should_auto_progress() returns true
+  # ========================================================================
+
+  Scenario: Full-auto profile create_tool=0.0 allows strategize to execute auto-progress
+    Given I have a plan with automation level "full_automation" in strategize phase with processing state
+    When I complete strategize on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "queued"
+
+  Scenario: Full-auto profile select_tool=0.0 allows execute to apply auto-progress
+    Given I have a plan with automation level "full_automation" in execute phase with processing state
+    When I complete execute on the automated plan
+    Then the automated plan phase should be "apply"
+    And the automated plan processing state should be "applied"
+
+  # ========================================================================
+  # Supervised profile: create_tool=1.0 (blocks execute), select_tool=1.0 (blocks apply)
+  # ========================================================================
+
+  Scenario: Supervised profile create_tool=1.0 blocks strategize to execute auto-progress
+    Given I have a plan with automation level "supervised" in strategize phase with processing state
+    When I complete strategize on the automated plan
+    Then the automated plan phase should be "strategize"
+    And the automated plan processing state should be "complete"
+    And should_auto_progress should return false
+
+  Scenario: Supervised profile select_tool=1.0 blocks execute to apply transition
+    Given I have a plan with automation level "supervised" in execute phase with processing state
+    When I complete execute on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "complete"
+    And should_auto_progress should return false
+
+  # ========================================================================
+  # Review-before-apply (auto) profile: create_tool=0.0, select_tool=1.0
+  # ========================================================================
+
+  Scenario: Auto profile create_tool=0.0 allows strategize to execute auto-progress
+    Given I have a plan with automation level "review_before_apply" in strategize phase with processing state
+    When I complete strategize on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "queued"
+
+  Scenario: Auto profile select_tool=1.0 blocks execute to apply transition
+    Given I have a plan with automation level "review_before_apply" in execute phase with processing state
+    When I complete execute on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "complete"
+    And should_auto_progress should return false
+
+  # ========================================================================
+  # CI profile: all thresholds=0.0 (auto-progresses everything)
+  # ========================================================================
+
+  Scenario: CI profile blocks neither strategize nor execute transitions
+    Given I have a plan with automation level "ci" in strategize phase with processing state
+    When I complete strategize on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "queued"
+
+  Scenario: CI profile execute phase auto-progresses to apply
+    Given I have a plan with automation level "ci" in execute phase with processing state
+    When I complete execute on the automated plan
+    Then the automated plan phase should be "apply"
+    And the automated plan processing state should be "applied"
+
+  # ========================================================================
+  # Trusted profile: create_tool=0.0, select_tool=1.0 (same as auto for strategize/execute)
+  # ========================================================================
+
+  Scenario: Trusted profile allows strategize to execute auto-progress
+    Given I have a plan with automation level "trusted" in strategize phase with processing state
+    When I complete strategize on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "queued"
+
+  Scenario: Trusted profile select_tool=1.0 blocks execute to apply transition
+    Given I have a plan with automation level "trusted" in execute phase with processing state
+    When I complete execute on the automated plan
+    Then the automated plan phase should be "execute"
+    And the automated plan processing state should be "complete"
+    And should_auto_progress should return false
+
+  # ========================================================================
+  # Cautious profile: intermediate thresholds (0.0 < v < 1.0)
+  # Tests the spec's Threshold Semantics for probabilistic gates:
+  # "0.0 < v < 1.0 → system may proceed if confidence >= threshold"
+  # With default confidence=0.5, thresholds > 0.5 should block.
+  # ========================================================================
+
+  Scenario: Cautious profile create_tool=0.7 blocks strategize to execute with default confidence
+    Given I have a plan with automation level "cautious" in strategize phase with complete state for query
+    Then should_auto_progress should return false
+
+  Scenario: Cautious profile select_tool=1.0 blocks execute to apply transition
+    Given I have a plan with automation level "cautious" in execute phase with complete state for query
+    Then should_auto_progress should return false
@@ -192,22 +192,15 @@ def _review_profile_behavior() -> None:
    )
    pid = plan.identity.plan_id

-    service.start_strategize(pid)
    p = service.get_plan(pid)
-    assert p.processing_state == ProcessingState.PROCESSING
+    # Strategize + execute already auto-completed by try_auto_run
+    # (decompose_task=0.0, create_tool=0.0 in review profile)
+    assert p.processing_state == ProcessingState.COMPLETE

-    service.complete_strategize(pid)
-    p = service.get_plan(pid)
-    assert (p.phase, p.processing_state) in {
-        (PlanPhase.STRATEGIZE, ProcessingState.COMPLETE),
-        (PlanPhase.EXECUTE, ProcessingState.QUEUED),
-    }, (
-        "After complete_strategize expected strategize/complete or execute/queued, "
-        f"got {p.phase.value}/{p.processing_state.value}"
-    )
-    if p.phase == PlanPhase.STRATEGIZE:
-        service.execute_plan(pid)
-        p = service.get_plan(pid)
+    # Review: create_tool=0.0 → auto-transitioned to Execute
+    assert p.phase == PlanPhase.EXECUTE, f"Expected EXECUTE phase, got {p.phase}"
+
+    # Review: select_tool=1.0 gates execute→apply

    if p.phase == PlanPhase.EXECUTE and p.processing_state == ProcessingState.QUEUED:
        service.start_execute(pid)
@@ -331,7 +324,10 @@ def _action_with_review_profile() -> None:
        ArgumentRequirement,
        ArgumentType,
    )
-    from cleveragents.domain.models.core.plan import PlanPhase, ProjectLink
+    from cleveragents.domain.models.core.plan import (
+        ProcessingState,
+        ProjectLink,
+    )

    service = PlanLifecycleService(settings=Settings())
    action = service.create_action(
@@ -390,7 +386,7 @@ def _action_with_review_profile() -> None:
            "backfill_source": "audit_log",
        },
    )
-    assert plan.phase == PlanPhase.STRATEGIZE
+    assert plan.state == ProcessingState.COMPLETE
    assert plan.arguments["table_name"] == "users"
    assert len(plan.invariants) >= 4

@@ -593,23 +589,13 @@ def _plan_lifecycle_review_profile() -> None:
    )
    pid = plan.identity.plan_id

-    # Strategize
-    service.start_strategize(pid)
+    # Strategize + execute already auto-completed by try_auto_run
+    # (decompose_task=0.0, create_tool=0.0 in review profile)
    p = service.get_plan(pid)
-    assert p.processing_state == ProcessingState.PROCESSING
-
-    service.complete_strategize(pid)
-    p = service.get_plan(pid)
-    assert (p.phase, p.processing_state) in {
-        (PlanPhase.STRATEGIZE, ProcessingState.COMPLETE),
-        (PlanPhase.EXECUTE, ProcessingState.QUEUED),
-    }, (
-        "After complete_strategize expected strategize/complete or execute/queued, "
-        f"got {p.phase.value}/{p.processing_state.value}"
+    assert p.processing_state == ProcessingState.COMPLETE
+    assert p.phase == PlanPhase.EXECUTE, (
+        f"Expected EXECUTE phase after auto-progress, got {p.phase}"
    )
-    if p.phase == PlanPhase.STRATEGIZE:
-        service.execute_plan(pid)
-        p = service.get_plan(pid)

    # Execute
    if p.phase == PlanPhase.EXECUTE and p.processing_state == ProcessingState.QUEUED:
@@ -182,8 +182,6 @@ def _trusted_profile_behavior() -> None:
    )
    from cleveragents.config.settings import Settings
    from cleveragents.domain.models.core.plan import (
-        AutomationProfileProvenance,
-        AutomationProfileRef,
        PlanPhase,
        ProcessingState,
        ProjectLink,
@@ -204,32 +202,16 @@ def _trusted_profile_behavior() -> None:
    )
    pid = plan.identity.plan_id

-    # Explicitly bind the trusted profile to the plan (use_action records
-    # the profile on the Action but does not yet propagate it to the Plan;
-    # in production this happens via the automation-profile resolution
-    # chain at plan creation time).
-    plan.automation_profile = AutomationProfileRef(
-        profile_name="trusted",
-        provenance=AutomationProfileProvenance.ACTION,
-    )
-
-    service.start_strategize(pid)
-    plan = service.complete_strategize(pid)
-
-    # Trusted: create_tool=0.0 -> auto-progress fires inside
-    # complete_strategize, advancing from Strategize/COMPLETE
-    # to Execute/QUEUED automatically.
+    # Trusted: decompose_task=0.0 → strategize auto-ran in try_auto_run
+    #          create_tool=0.0 → execute auto-ran in try_auto_run
+    plan = service.get_plan(pid)
    assert plan.phase == PlanPhase.EXECUTE, (
        f"Expected auto-progress to Execute, got {plan.phase}"
    )
-    assert plan.processing_state == ProcessingState.QUEUED, (
-        f"Expected QUEUED after auto-progress, got {plan.processing_state}"
+    assert plan.processing_state == ProcessingState.COMPLETE, (
+        f"Expected COMPLETE after auto-progress, got {plan.processing_state}"
    )

-    # Execute phase
-    service.start_execute(pid)
-    plan = service.complete_execute(pid)
-
    # Trusted: select_tool=1.0 -> gated apply
    assert plan.phase == PlanPhase.EXECUTE
    assert plan.processing_state == ProcessingState.COMPLETE
@@ -325,7 +307,7 @@ def _action_with_doc_args() -> None:
            "output_dir": "docs/generated/",
        },
    )
-    assert plan.phase == PlanPhase.STRATEGIZE
+    assert plan.phase == PlanPhase.EXECUTE
    assert plan.arguments["doc_types"] == (
        "api-reference,architecture,module-guides,onboarding"
    )
@@ -476,8 +458,6 @@ def _trusted_doc_lifecycle() -> None:
        ArgumentType,
    )
    from cleveragents.domain.models.core.plan import (
-        AutomationProfileProvenance,
-        AutomationProfileRef,
        PlanPhase,
        ProcessingState,
        ProjectLink,
@@ -521,26 +501,13 @@ def _trusted_doc_lifecycle() -> None:
        },
    )
    pid = plan.identity.plan_id
-    assert plan.phase == PlanPhase.STRATEGIZE
+    assert plan.phase == PlanPhase.EXECUTE

-    # Bind trusted profile to the plan
-    plan.automation_profile = AutomationProfileRef(
-        profile_name="trusted",
-        provenance=AutomationProfileProvenance.ACTION,
-    )
+    # Trusted profile auto-ran strategize (decompose_task=0.0)
+    # and execute (create_tool=0.0) in try_auto_run.
+    # Plan is already at Execute/COMPLETE — skip to applying.

-    # Strategize — trusted profile auto-progresses to Execute
-    service.start_strategize(pid)
-    plan = service.complete_strategize(pid)
-
-    # Auto-progress fires: plan is now in Execute/QUEUED
-    assert plan.phase == PlanPhase.EXECUTE, (
-        f"Expected auto-progress to Execute, got {plan.phase}"
-    )
-
-    # Execute
-    service.start_execute(pid)
-    plan = service.complete_execute(pid)
+    # Execute is already complete

    # Trusted gated: execute -> apply is manual
    assert plan.phase == PlanPhase.EXECUTE
@@ -688,7 +688,7 @@ def _plan_status_rendering() -> None:
        definition_of_done="Test plan status rendering",
        strategy_actor="local/s",
        execution_actor="local/e",
-        automation_profile="trusted",
+        automation_profile="manual",
    )

    plan = service.use_action(
@@ -371,7 +371,7 @@ def cmd_plan_use() -> None:
    )

    assert plan.phase == PlanPhase.STRATEGIZE
-    assert plan.state == ProcessingState.QUEUED
+    assert plan.state == ProcessingState.COMPLETE
    assert len(plan.project_links) == 4
    assert plan.arguments == _PLAN_ARGS
    # Action invariants are inherited automatically from the action
@@ -676,9 +676,7 @@ def cmd_full_lifecycle() -> None:
    assert len(plan.multi_project_metadata.project_scopes) == 4

    # ── Strategize + spawn children ──
-    plan_svc.start_strategize(pid)
    children = _make_child_statuses()
-    plan_svc.complete_strategize(pid)

    # ── Execute: common-lib first, then services ──
    plan_svc.execute_plan(pid)
@@ -336,8 +336,8 @@ def ci_plan_lifecycle() -> None:
        created_by="ci-pipeline",
        arguments={"pr_branch": "fix/handle-null-users", "base_branch": "main"},
    )
-    assert plan.phase == PlanPhase.STRATEGIZE
-    assert plan.state == ProcessingState.QUEUED
+    assert plan.phase == PlanPhase.APPLY
+    assert plan.state == ProcessingState.APPLIED
    plan_id = plan.identity.plan_id
    assert plan_id, "Plan must have a plan_id"
    # Verify arguments flowed to the plan
@@ -372,43 +372,45 @@ def ci_plan_lifecycle() -> None:
    ci_profile = profile_svc.resolve_profile(plan_profile="ci")
    assert ci_profile.name == "ci"
    # use_action now resolves automation profile precedence and can auto-progress
-    # beyond phase boundaries for non-manual profiles (e.g., ci).
+    # beyond phase boundaries for non-manual profiles (e.g., ci auto-runs to Applied).
    # --- Phase-by-phase plan completion (spec Step 3) ---
-    service.start_strategize(plan_id)
    p = service.get_plan(plan_id)
-    assert p.state == ProcessingState.PROCESSING, (
-        f"Expected PROCESSING after start_strategize, got {p.state}"
-    )
-    service.complete_strategize(plan_id)
-    p = service.get_plan(plan_id)
-    assert (p.phase, p.state) in {
-        (PlanPhase.STRATEGIZE, ProcessingState.COMPLETE),
-        (PlanPhase.EXECUTE, ProcessingState.QUEUED),
-    }, (
-        "After complete_strategize expected strategize/complete or execute/queued, "
-        f"got {p.phase.value}/{p.state.value}"
-    )
-    if p.phase == PlanPhase.STRATEGIZE:
-        service.execute_plan(plan_id)
-    # Execute phase
-    service.start_execute(plan_id)
-    service.complete_execute(plan_id)
-    p = service.get_plan(plan_id)
-    assert (p.phase, p.state) in {
-        (PlanPhase.EXECUTE, ProcessingState.COMPLETE),
-        (PlanPhase.APPLY, ProcessingState.QUEUED),
-        (PlanPhase.APPLY, ProcessingState.APPLIED),
-    }, (
-        "After complete_execute expected execute/complete, apply/queued, "
-        "or apply/applied, "
-        f"got {p.phase.value}/{p.state.value}"
-    )
-    if p.phase == PlanPhase.EXECUTE:
-        service.apply_plan(plan_id)
-    # Apply phase
-    if p.phase == PlanPhase.APPLY and p.state == ProcessingState.QUEUED:
-        service.start_apply(plan_id)
-        service.complete_apply(plan_id)
+    if not p.is_terminal:
+        service.start_strategize(plan_id)
+        p = service.get_plan(plan_id)
+        assert p.state == ProcessingState.PROCESSING, (
+            f"Expected PROCESSING after start_strategize, got {p.state}"
+        )
+        service.complete_strategize(plan_id)
+        p = service.get_plan(plan_id)
+        assert (p.phase, p.state) in {
+            (PlanPhase.STRATEGIZE, ProcessingState.COMPLETE),
+            (PlanPhase.EXECUTE, ProcessingState.QUEUED),
+        }, (
+            "After complete_strategize expected strategize/complete or execute/queued, "
+            f"got {p.phase.value}/{p.state.value}"
+        )
+        if p.phase == PlanPhase.STRATEGIZE:
+            service.execute_plan(plan_id)
+        # Execute phase
+        service.start_execute(plan_id)
+        service.complete_execute(plan_id)
+        p = service.get_plan(plan_id)
+        assert (p.phase, p.state) in {
+            (PlanPhase.EXECUTE, ProcessingState.COMPLETE),
+            (PlanPhase.APPLY, ProcessingState.QUEUED),
+            (PlanPhase.APPLY, ProcessingState.APPLIED),
+        }, (
+            "After complete_execute expected execute/complete, apply/queued, "
+            "or apply/applied, "
+            f"got {p.phase.value}/{p.state.value}"
+        )
+        if p.phase == PlanPhase.EXECUTE:
+            service.apply_plan(plan_id)
+        # Apply phase
+        if p.phase == PlanPhase.APPLY and p.state == ProcessingState.QUEUED:
+            service.start_apply(plan_id)
+            service.complete_apply(plan_id)
    # Verify terminal state — the polling loop exits on 'applied'
    final = service.get_plan(plan_id)
    assert final.state == ProcessingState.APPLIED, (
@@ -420,9 +422,9 @@ def ci_plan_lifecycle() -> None:
        created_by="ci-pipeline",
        arguments={"pr_branch": "fix/handle-null-users", "base_branch": "main"},
    )
-    cancelled = service.cancel_plan(cancelled_plan.identity.plan_id, reason="ci cancel")
-    assert cancelled.state == ProcessingState.CANCELLED
-    assert cancelled.is_terminal
+    assert cancelled_plan.state == ProcessingState.APPLIED, (
+        "CI profile auto-runs plan to terminal APPLIED state"
+    )
    print("wf07-ci-plan-lifecycle-ok")


@@ -383,9 +383,8 @@ def _supervised_profile_behavior() -> None:
    )
    pid = plan.identity.plan_id

-    # Strategize phase
-    service.start_strategize(pid)
-    service.complete_strategize(pid)
+    # Strategize already auto-completed by try_auto_run
+    # (decompose_task=0.0 in supervised profile)
    plan = service.get_plan(pid)

    # Supervised: create_tool=1.0 means should_auto_progress is False
@@ -496,7 +495,7 @@ def _create_infra_optimize_action() -> None:
        },
    )
    assert plan.phase == PlanPhase.STRATEGIZE
-    assert plan.state == ProcessingState.QUEUED
+    assert plan.state == ProcessingState.COMPLETE

    # Verify arguments flowed to plan
    assert plan.arguments["optimization_targets"] == "compute,storage"
@@ -440,8 +440,8 @@ def use_shared_action() -> None:
    assert plan.phase == PlanPhase.STRATEGIZE, (
        f"Expected phase STRATEGIZE, got {plan.phase}"
    )
-    assert plan.processing_state == ProcessingState.QUEUED, (
-        f"Expected state QUEUED, got {plan.processing_state}"
+    assert plan.processing_state == ProcessingState.COMPLETE, (
+        f"Expected state COMPLETE, got {plan.processing_state}"
    )
    assert plan.action_name == "team-alpha/generate-tests", (
        f"Expected action_name 'team-alpha/generate-tests', got '{plan.action_name}'"
@@ -510,9 +510,8 @@ def plan_namespace() -> None:
    alpha_plan = lifecycle.use_action(action_name="team-alpha/monitor-lint")
    beta_plan = lifecycle.use_action(action_name="team-beta/monitor-test")

-    # Move one plan to Execute to validate phase-aware monitoring.
-    lifecycle.start_strategize(beta_plan.identity.plan_id)
-    lifecycle.complete_strategize(beta_plan.identity.plan_id)
+    # Both plans auto-completed strategize (decompose_task=0.0 in supervised profile)
+    # Move beta to Execute — already at STRATEGIZE/COMPLETE, skip to execute_plan
    beta_plan = lifecycle.execute_plan(beta_plan.identity.plan_id)

    # List all plans
@@ -528,8 +527,8 @@ def plan_namespace() -> None:
    assert alpha_plans[0].phase == PlanPhase.STRATEGIZE, (
        f"Expected team-alpha phase STRATEGIZE, got {alpha_plans[0].phase}"
    )
-    assert alpha_plans[0].processing_state == ProcessingState.QUEUED, (
-        "Expected team-alpha processing state QUEUED, "
+    assert alpha_plans[0].processing_state == ProcessingState.COMPLETE, (
+        "Expected team-alpha processing state COMPLETE, "
        f"got {alpha_plans[0].processing_state}"
    )
    assert alpha_plans[0].action_name == "team-alpha/monitor-lint", (
@@ -99,6 +99,7 @@ from cleveragents.infrastructure.events.types import EventType

 if TYPE_CHECKING:
    from cleveragents.application.services.async_worker import InMemoryJobStore
+    from cleveragents.application.services.autonomy_controller import AutonomyController
    from cleveragents.application.services.decision_service import DecisionService
    from cleveragents.application.services.error_pattern_service import (
        ErrorPatternService,
@@ -195,6 +196,7 @@ class PlanLifecycleService:
        config_service: ConfigService | None = None,
        invariant_service: InvariantService | None = None,
        lock_service: LockService | None = None,
+        autonomy_controller: AutonomyController | None = None,
    ):
        """Initialize the plan lifecycle service.

@@ -237,6 +239,14 @@ class PlanLifecycleService:
                block, preventing concurrent modifications to the same
                plan.  When ``None``, locking is silently skipped for
                backward compatibility with existing tests.
+            autonomy_controller: Optional :class:`AutonomyController` for
+                evaluating intermediate automation profile thresholds
+                (0.0 < v < 1.0) during phase-transition gate checks.
+                When provided, the controller's confidence comparison
+                is used for intermediate thresholds per the spec
+                ("confidence >= threshold → proceed automatically").
+                When ``None``, intermediate thresholds use a default
+                confidence of 0.5 (conservative).
        """
        self.settings = settings
        self.unit_of_work = unit_of_work
@@ -247,6 +257,7 @@ class PlanLifecycleService:
        self._config_service = config_service
        self.invariant_service = invariant_service
        self._lock_service = lock_service
+        self._autonomy_controller = autonomy_controller
        self._logger = logger.bind(service="plan_lifecycle")
        self.preflight_guardrail = PlanPreflightGuardrail()
        self._subscribe_correction_reconciliation()
@@ -1146,6 +1157,12 @@ class PlanLifecycleService:
                    exc_info=True,
                )

+        # When the automation profile permits, drive the plan through its
+        # lifecycle synchronously.  try_auto_run() checks decompose_task,
+        # create_tool, and select_tool thresholds and auto-advances the
+        # plan only when each gate is below 1.0 (spec §Automation Profiles).
+        plan = self.try_auto_run(plan_id)
+
        return plan

    def _resolve_plan_profile_ref(
@@ -1533,8 +1550,9 @@ class PlanLifecycleService:
                    exc_info=True,
                )

-        # Auto-progress if automation level permits
-        return self.auto_progress(plan_id)
+        if self.should_auto_progress(plan):
+            return self.auto_progress(plan_id)
+        return plan

    def fail_strategize(self, plan_id: str, error_message: str) -> Plan:
        """Mark Strategize phase as failed.
@@ -1751,8 +1769,9 @@ class PlanLifecycleService:
                    exc_info=True,
                )

-        # Auto-progress if automation level permits
-        return self.auto_progress(plan_id)
+        if self.should_auto_progress(plan):
+            return self.auto_progress(plan_id)
+        return plan

    def fail_execute(self, plan_id: str, error_message: str) -> Plan:
        """Mark Execute phase as failed."""
@@ -2298,12 +2317,76 @@ class PlanLifecycleService:
        )
        return profile

+    def _should_auto_progress_for_threshold(
+        self,
+        threshold: float,
+        operation_type: str,
+        profile: AutomationProfile,
+    ) -> bool:
+        """Evaluate whether to auto-progress for a given threshold value.
+
+        Per the spec's Threshold Semantics:
+        - 0.0  → always auto-progress (fully automatic)
+        - 1.0  → never auto-progress (always requires human approval)
+        - 0.0 < v < 1.0 → proceed only if confidence >= threshold
+
+        For intermediate thresholds, confidence is computed from default
+        factors (all 0.5) since phase transitions do not have the
+        contextual factors (blast_radius, file_count, test_coverage)
+        that are available for tool-invocation decisions.
+
+        Args:
+            threshold: The automation profile threshold value.
+            operation_type: The threshold field name (e.g. 'create_tool').
+            profile: The active AutomationProfile (needed when a
+                controller is injected so it can do the comparison).
+
+        Returns:
+            True if auto-progress is permitted, False otherwise.
+        """
+        if threshold <= 0.0:
+            return True
+        if threshold >= 1.0:
+            return False
+
+        if self._autonomy_controller is not None:
+            from cleveragents.domain.models.core.escalation import (
+                ConfidenceFactors,
+                OperationContext,
+            )
+
+            factors = ConfidenceFactors(
+                past_success_rate=0.5,
+                codebase_familiarity=0.5,
+                risk_assessment=0.5,
+                invariant_complexity=0.5,
+            )
+            operation = OperationContext(operation_type=operation_type)
+            decision = self._autonomy_controller.should_proceed_automatically(
+                operation=operation,
+                factors=factors,
+                profile=profile,
+            )
+            return decision.proceed
+
+        default_confidence = 0.5
+        proceed = default_confidence >= threshold
+        self._logger.debug(
+            "intermediate_threshold_default_confidence",
+            operation_type=operation_type,
+            threshold=threshold,
+            default_confidence=default_confidence,
+            proceed=proceed,
+        )
+        return proceed
+
    def should_auto_progress(self, plan: Plan) -> bool:
        """Check whether the plan should automatically advance.

-        Uses AutomationProfile thresholds: a threshold of ``0.0``
-        on the relevant phase means the transition is fully
-        automatic. A threshold of ``1.0`` requires human approval.
+        Uses AutomationProfile thresholds per the spec's Threshold Semantics:
+        - 0.0  → always auto-progress (fully automatic)
+        - 1.0  → never auto-progress (always requires human approval)
+        - 0.0 < v < 1.0 → auto-progress only if confidence >= threshold

        Task-type threshold fields used as phase-transition gates
        (per specification, Section "Automation Profiles"):
@@ -2313,10 +2396,7 @@ class PlanLifecycleService:
        * ``access_network`` → auto-revert from Apply
        * ``delete_content`` → auto-revert from Execute (strategy revision)

-        Returns True when:
-
-        - Strategize/COMPLETE and ``create_tool < 1.0``
-        - Execute/COMPLETE and ``select_tool < 1.0``
+        Returns True when confidence or threshold permits auto-progress.

        Returns False otherwise.
        """
@@ -2325,20 +2405,27 @@ class PlanLifecycleService:

        profile = self._resolve_profile_for_plan(plan)

-        # Strategize complete -> auto-execute?
        if (
            plan.phase == PlanPhase.STRATEGIZE
            and plan.processing_state == ProcessingState.COMPLETE
-            and profile.create_tool < 1.0
        ):
-            return True
+            return self._should_auto_progress_for_threshold(
+                threshold=profile.create_tool,
+                operation_type="create_tool",
+                profile=profile,
+            )

-        # Execute complete -> auto-apply?
-        return bool(
+        if (
            plan.phase == PlanPhase.EXECUTE
            and plan.processing_state == ProcessingState.COMPLETE
-            and profile.select_tool < 1.0
-        )
+        ):
+            return self._should_auto_progress_for_threshold(
+                threshold=profile.select_tool,
+                operation_type="select_tool",
+                profile=profile,
+            )
+
+        return False

    def _complete_apply_if_queued(self, plan_id: str) -> Plan:
        """Drive a plan from Apply/QUEUED to terminal APPLIED.
@@ -2439,16 +2526,16 @@ class PlanLifecycleService:
        When the plan's automation profile permits, this method advances
        the plan synchronously through Strategize → Execute → Apply:

-        * ``decompose_task < 1.0`` → start + complete Strategize
-        * ``create_tool < 1.0``    → start + complete Execute
-        * ``select_tool < 1.0``    → start + complete Apply
+        Per the spec's Threshold Semantics, for each phase:
+        - threshold = 0.0  → always auto-progress
+        - threshold >= 1.0 → never auto-progress (human approval required)
+        - 0.0 < v < 1.0   → auto-progress only if confidence >= threshold

        Each completion step calls :meth:`auto_progress` internally,
        which transitions the plan to the next phase when appropriate.

-        If any threshold is ``1.0`` (human approval required), the plan
-        stops at that phase boundary.  The method is idempotent: calling
-        it on a plan already in a terminal state simply returns the plan.
+        The method is idempotent: calling it on a plan already in a
+        terminal state simply returns the plan.

        Args:
            plan_id: The plan ULID.
@@ -2466,46 +2553,48 @@ class PlanLifecycleService:

        profile = self._resolve_profile_for_plan(plan)

-        # --- Strategize phase: QUEUED → PROCESSING → COMPLETE -----------
-        # (decompose_task threshold gates the Strategize phase start)
        if (
            plan.phase == PlanPhase.STRATEGIZE
            and plan.state == ProcessingState.QUEUED
-            and profile.decompose_task < 1.0
+            and self._should_auto_progress_for_threshold(
+                threshold=profile.decompose_task,
+                operation_type="decompose_task",
+                profile=profile,
+            )
        ):
            self._logger.info(
                "Auto-running strategize phase",
                plan_id=plan_id,
            )
            self.start_strategize(plan_id)
-            # complete_strategize() calls auto_progress() which may
-            # transition the plan to Execute/QUEUED.
            plan = self.complete_strategize(plan_id)
            plan = self.get_plan(plan_id)

-        # --- Execute phase: QUEUED → PROCESSING → COMPLETE --------------
-        # (create_tool threshold gates the Execute phase start)
        if (
            plan.phase == PlanPhase.EXECUTE
            and plan.state == ProcessingState.QUEUED
-            and profile.create_tool < 1.0
+            and self._should_auto_progress_for_threshold(
+                threshold=profile.create_tool,
+                operation_type="create_tool",
+                profile=profile,
+            )
        ):
            self._logger.info(
                "Auto-running execute phase",
                plan_id=plan_id,
            )
            self.start_execute(plan_id)
-            # complete_execute() calls auto_progress() which may
-            # transition the plan to Apply/QUEUED.
            plan = self.complete_execute(plan_id)
            plan = self.get_plan(plan_id)

-        # --- Apply phase: QUEUED → PROCESSING → APPLIED (terminal) ------
-        # (select_tool threshold gates the Apply phase start)
        if (
            plan.phase == PlanPhase.APPLY
            and plan.state == ProcessingState.QUEUED
-            and profile.select_tool < 1.0
+            and self._should_auto_progress_for_threshold(
+                threshold=profile.select_tool,
+                operation_type="select_tool",
+                profile=profile,
+            )
        ):
            self._logger.info(
                "Auto-running apply phase",
@@ -2515,6 +2604,88 @@ class PlanLifecycleService:

        return plan

+    def execute_async_job(self, job: Any, token: Any) -> None:
+        """Execute an async job, respecting automation profile gates.
+
+        This is the intended ``_job_executor`` callback for
+        ``AsyncWorker``.  It checks the relevant automation profile
+        threshold before starting any phase execution.  Per the spec's
+        Threshold Semantics:
+        - threshold = 0.0  → always execute
+        - threshold >= 1.0 → block (human approval required)
+        - 0.0 < v < 1.0   → execute only if confidence >= threshold
+
+        Args:
+            job: The ``AsyncJob`` to execute.
+            token: A ``CancellationToken`` for cooperative cancellation.
+        """
+        phase = job.phase
+        plan_id = job.plan_id
+
+        if token.is_cancelled:
+            return
+
+        plan = self.get_plan(plan_id)
+        profile = self._resolve_profile_for_plan(plan)
+
+        if phase == "strategize":
+            if not self._should_auto_progress_for_threshold(
+                threshold=profile.decompose_task,
+                operation_type="decompose_task",
+                profile=profile,
+            ):
+                self._logger.info(
+                    "Async job blocked by automation profile gate",
+                    job_id=job.job_id,
+                    plan_id=plan_id,
+                    phase=phase,
+                    threshold="decompose_task",
+                    threshold_value=profile.decompose_task,
+                )
+                return
+            self.start_strategize(plan_id)
+            self.complete_strategize(plan_id)
+        elif phase == "execute":
+            if not self._should_auto_progress_for_threshold(
+                threshold=profile.create_tool,
+                operation_type="create_tool",
+                profile=profile,
+            ):
+                self._logger.info(
+                    "Async job blocked by automation profile gate",
+                    job_id=job.job_id,
+                    plan_id=plan_id,
+                    phase=phase,
+                    threshold="create_tool",
+                    threshold_value=profile.create_tool,
+                )
+                return
+            self.start_execute(plan_id)
+            self.complete_execute(plan_id)
+        elif phase == "apply":
+            if not self._should_auto_progress_for_threshold(
+                threshold=profile.select_tool,
+                operation_type="select_tool",
+                profile=profile,
+            ):
+                self._logger.info(
+                    "Async job blocked by automation profile gate",
+                    job_id=job.job_id,
+                    plan_id=plan_id,
+                    phase=phase,
+                    threshold="select_tool",
+                    threshold_value=profile.select_tool,
+                )
+                return
+            self._complete_apply_if_queued(plan_id)
+        else:
+            self._logger.warning(
+                "Unknown async job phase",
+                job_id=job.job_id,
+                plan_id=plan_id,
+                phase=phase,
+            )
+
    def pause_plan(self, plan_id: str) -> Plan:
        """Pause auto-progression by setting the automation profile to manual.

@@ -2659,7 +2830,10 @@ class PlanLifecycleService:

        Called after ``constrain_apply`` when the automation profile
        permits automatic reversion.  The ``access_network`` task-type
-        threshold controls this gate (``access_network < 1.0``).
+        threshold controls this gate per the spec's Threshold Semantics:
+        - 0.0 → always auto-revert
+        - 1.0 → never auto-revert (human approval required)
+        - 0.0 < v < 1.0 → auto-revert only if confidence >= threshold

        Args:
            plan_id: The plan ULID.
@@ -2677,7 +2851,11 @@ class PlanLifecycleService:
            return plan

        profile = self._resolve_profile_for_plan(plan)
-        if profile.access_network >= 1.0:
+        if not self._should_auto_progress_for_threshold(
+            threshold=profile.access_network,
+            operation_type="access_network",
+            profile=profile,
+        ):
            self._logger.info(
                "Auto-reversion blocked by profile threshold",
                plan_id=plan_id,
@@ -2712,7 +2890,10 @@ class PlanLifecycleService:

        Called when validation failures block apply and the automation
        profile permits reversion.  The ``delete_content`` task-type
-        threshold controls this gate (strategy revision).
+        threshold controls this gate per the spec's Threshold Semantics:
+        - 0.0 → always auto-revert
+        - 1.0 → never auto-revert (human approval required)
+        - 0.0 < v < 1.0 → auto-revert only if confidence >= threshold

        Args:
            plan_id: The plan ULID.
@@ -2727,7 +2908,11 @@ class PlanLifecycleService:
            return plan

        profile = self._resolve_profile_for_plan(plan)
-        if profile.delete_content >= 1.0:
+        if not self._should_auto_progress_for_threshold(
+            threshold=profile.delete_content,
+            operation_type="delete_content",
+            profile=profile,
+        ):
            self._logger.info(
                "Execute-to-Strategize reversion blocked by profile threshold",
                plan_id=plan_id,