CHANGELOG.md

@@ -143,6 +143,14 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- **Plan Concurrency Race Condition** (#7989): Fixed critical race condition in `execute_plan()` and
+  `apply_plan()` where concurrent CLI/worker sessions could simultaneously modify the same plan,
+  corrupting plan state. `LockService` is now wired into the plan lifecycle with plan-level advisory
+  locking. Each invocation generates a unique caller identity (UUID) to prevent re-entrant lock
+  acquisition by concurrent sessions on the same plan. Concurrent attempts now raise `LockConflictError`
+  instead of silently racing. Lock is acquired before phase transition and released in a `finally`
+  block to ensure cleanup even on error.
+
 - **Validation Gate Empty-Run Guard** (#7508): Fixed `ApplyValidationSummary.all_required_passed`
   returning `True` when zero validations were run, silently bypassing the apply gate. The property
   now returns `False` when the validation result set is empty (`is_empty` is `True`), ensuring

CONTRIBUTORS.md

@@ -15,4 +15,5 @@ Below are some of the specific details of various contributions.
 * Jeffrey Phillips Freeman has acted as Lead Developer, daily contributor, and Project Owner.
 * Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements.
 * HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool.
+* HAL 9000 has contributed the plan concurrency race-condition fix (#7989): wired `LockService` into the plan lifecycle, guarding `execute_plan()` and `apply_plan()` with plan-level advisory locks and unique per-invocation owner identities to prevent silent concurrent state corruption.
 * This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.

src/cleveragents/application/container.py

@@ -55,6 +55,7 @@ from cleveragents.application.services.fix_then_revalidate import (
     FixThenRevalidateOrchestrator,
 )
 from cleveragents.application.services.invariant_service import InvariantService
+from cleveragents.application.services.lock_service import LockService
 from cleveragents.application.services.multi_project_service import (
     MultiProjectService,
 )
@@ -267,6 +268,24 @@ def _build_session_factory(database_url: str) -> sessionmaker[Session]:
     return sessionmaker(bind=engine, expire_on_commit=False)
 
 
+def _build_lock_service(
+    database_url: str,
+) -> LockService:
+    """Build a LockService with a session factory from the database URL.
+
+    Registered as a Singleton so that all callers within a process share
+    the same advisory-lock state.  Follows the same ``_build_*`` pattern
+    used by other DB-backed services (see ``_build_session_factory``,
+    ``_build_resource_registry_service``, etc.).
+    """
+    from sqlalchemy import create_engine
+    from sqlalchemy.orm import sessionmaker
+
+    engine = create_engine(database_url, echo=False)
+    factory = sessionmaker(bind=engine, expire_on_commit=False)
+    return LockService(session_factory=factory)
+
+
 def _build_resource_registry_service(
     database_url: str,
 ) -> ResourceRegistryService:
@@ -657,6 +676,12 @@ class Container(containers.DeclarativeContainer):
         InvariantService,
     )
 
+    # Lock Service - Singleton (shared advisory-lock state per process, #7989)
+    lock_service = providers.Singleton(
+        _build_lock_service,
+        database_url=database_url,
+    )
+
     # Plan Lifecycle Service - Factory (v3 four-phase lifecycle)
     plan_lifecycle_service = providers.Factory(
         PlanLifecycleService,
@@ -665,6 +690,7 @@ class Container(containers.DeclarativeContainer):
         decision_service=decision_service,
         event_bus=event_bus,
         invariant_service=invariant_service,
+        lock_service=lock_service,
     )
 
     # Checkpoint Service - database-backed via CheckpointRepository

src/cleveragents/application/services/plan_lifecycle_service.py

@@ -54,6 +54,7 @@ from __future__ import annotations
 from contextlib import suppress
 from datetime import datetime
 from typing import TYPE_CHECKING, Any
+from uuid import uuid4
 
 import structlog
 from ulid import ULID
@@ -101,6 +102,7 @@ if TYPE_CHECKING:
         ErrorPatternService,
     )
     from cleveragents.application.services.invariant_service import InvariantService
+    from cleveragents.application.services.lock_service import LockService
     from cleveragents.config.settings import Settings
     from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
     from cleveragents.infrastructure.events.protocol import EventBus
@@ -190,6 +192,7 @@ class PlanLifecycleService:
         error_pattern_service: ErrorPatternService | None = None,
         config_service: ConfigService | None = None,
         invariant_service: InvariantService | None = None,
+        lock_service: LockService | None = None,
     ):
         """Initialize the plan lifecycle service.
 
@@ -225,6 +228,13 @@ class PlanLifecycleService:
                 Reconciliation Actor is auto-invoked at each phase
                 transition to verify invariants hold.  When ``None``,
                 reconciliation is silently skipped.
+            lock_service: Optional :class:`LockService` for plan-level
+                advisory locking.  When provided, ``execute_plan`` and
+                ``apply_plan`` acquire a plan lock before performing
+                the phase transition and release it in a ``finally``
+                block, preventing concurrent modifications to the same
+                plan.  When ``None``, locking is silently skipped for
+                backward compatibility with existing tests.
         """
         self.settings = settings
         self.unit_of_work = unit_of_work
@@ -234,6 +244,7 @@ class PlanLifecycleService:
         self.error_pattern_service = error_pattern_service
         self._config_service = config_service
         self.invariant_service = invariant_service
+        self._lock_service = lock_service
         self._logger = logger.bind(service="plan_lifecycle")
         self.preflight_guardrail = PlanPreflightGuardrail()
         self._subscribe_correction_reconciliation()
@@ -1497,6 +1508,11 @@ class PlanLifecycleService:
 
         This is the 'execute' command.
 
+        A plan-level advisory lock is acquired before the transition and
+        released in a ``finally`` block so that concurrent callers on the
+        same ``plan_id`` receive a ``LockConflictError`` rather than
+        silently racing into the same phase transition.
+
         Args:
             plan_id: The plan ULID
 
@@ -1507,67 +1523,91 @@ class PlanLifecycleService:
             NotFoundError: If plan not found
             InvalidPhaseTransitionError: If transition is not valid
             PlanNotReadyError: If plan is not ready for transition
+            LockConflictError: If another session holds the plan lock
         """
-        plan = self.get_plan(plan_id)
+        # Generate a unique owner_id for this invocation to ensure that
+        # concurrent sessions cannot re-entrantly acquire the same lock.
+        # The LockService treats owner_id as the caller identity and allows
+        # re-entrant acquisition for the same owner; using a unique UUID per
+        # invocation ensures that concurrent sessions present different owners
+        # and thus trigger LockConflictError when attempting to acquire the
+        # same plan lock.
+        owner_id: str = str(uuid4())
 
-        # Validate transition
-        if not can_transition(plan.phase, PlanPhase.EXECUTE):
-            raise InvalidPhaseTransitionError(plan.phase, PlanPhase.EXECUTE)
-
-        # Must be in COMPLETE state to transition
-        if plan.state != ProcessingState.COMPLETE:
-            raise PlanNotReadyError(
-                plan_id, plan.phase, plan.state or ProcessingState.QUEUED
+        if self._lock_service is not None:
+            self._lock_service.acquire(
+                owner_id=owner_id,
+                resource_type="plan",
+                resource_id=plan_id,
             )
+        try:
+            plan = self.get_plan(plan_id)
 
-        # Run estimation actor if configured (informational only)
-        self._run_estimation(plan)
+            # Validate transition
+            if not can_transition(plan.phase, PlanPhase.EXECUTE):
+                raise InvalidPhaseTransitionError(plan.phase, PlanPhase.EXECUTE)
 
-        # Layer 4: Consult Error Pattern Database for preventive guidance
-        self._consult_error_patterns(plan)
+            # Must be in COMPLETE state to transition
+            if plan.state != ProcessingState.COMPLETE:
+                raise PlanNotReadyError(
+                    plan_id, plan.phase, plan.state or ProcessingState.QUEUED
+                )
 
-        # Invariant Reconciliation: verify invariants before Execute
-        self._run_invariant_reconciliation(plan)
+            # Run estimation actor if configured (informational only)
+            self._run_estimation(plan)
 
-        # Transition to Execute phase — set processing_state first so that
-        # the phase-state validator sees QUEUED (valid in any phase) when
-        # the phase assignment triggers re-validation.
-        plan.processing_state = ProcessingState.QUEUED
-        plan.phase = PlanPhase.EXECUTE
-        plan.timestamps.updated_at = datetime.now()
+            # Layer 4: Consult Error Pattern Database for preventive guidance
+            self._consult_error_patterns(plan)
 
-        self._commit_plan(plan)
-        self._logger.info(
-            "Plan transitioned to Execute",
-            plan_id=plan_id,
-            phase=plan.phase.value,
-        )
-        if self.event_bus is not None:
-            try:
-                self.event_bus.emit(
-                    DomainEvent(
-                        event_type=EventType.PLAN_PHASE_CHANGED,
-                        plan_id=plan_id,
-                        details={
-                            "phase": plan.phase.value,
-                            "processing_state": plan.processing_state.value
-                            if plan.processing_state
-                            else None,
-                        },
+            # Invariant Reconciliation: verify invariants before Execute
+            self._run_invariant_reconciliation(plan)
+
+            # Transition to Execute phase — set processing_state first so that
+            # the phase-state validator sees QUEUED (valid in any phase) when
+            # the phase assignment triggers re-validation.
+            plan.processing_state = ProcessingState.QUEUED
+            plan.phase = PlanPhase.EXECUTE
+            plan.timestamps.updated_at = datetime.now()
+
+            self._commit_plan(plan)
+            self._logger.info(
+                "Plan transitioned to Execute",
+                plan_id=plan_id,
+                phase=plan.phase.value,
+            )
+            if self.event_bus is not None:
+                try:
+                    self.event_bus.emit(
+                        DomainEvent(
+                            event_type=EventType.PLAN_PHASE_CHANGED,
+                            plan_id=plan_id,
+                            details={
+                                "phase": plan.phase.value,
+                                "processing_state": plan.processing_state.value
+                                if plan.processing_state
+                                else None,
+                            },
+                        )
+                    )
+                except Exception:
+                    self._logger.warning(
+                        "event_bus_emit_failed",
+                        event_type="PLAN_PHASE_CHANGED",
+                        plan_id=plan_id,
+                        exc_info=True,
                     )
-                )
-            except Exception:
-                self._logger.warning(
-                    "event_bus_emit_failed",
-                    event_type="PLAN_PHASE_CHANGED",
-                    plan_id=plan_id,
-                    exc_info=True,
-                )
 
-        # Enqueue async job when async execution is enabled
-        self._maybe_enqueue_async_job(plan_id, "execute")
+            # Enqueue async job when async execution is enabled
+            self._maybe_enqueue_async_job(plan_id, "execute")
 
-        return plan
+            return plan
+        finally:
+            if self._lock_service is not None:
+                self._lock_service.release(
+                    owner_id=owner_id,
+                    resource_type="plan",
+                    resource_id=plan_id,
+                )
 
     def start_execute(self, plan_id: str) -> Plan:
         """Start the Execute phase processing."""
@@ -1681,6 +1721,11 @@ class PlanLifecycleService:
 
         This is the 'apply' command.
 
+        A plan-level advisory lock is acquired before the transition and
+        released in a ``finally`` block so that concurrent callers on the
+        same ``plan_id`` receive a ``LockConflictError`` rather than
+        silently racing into the same phase transition.
+
         Args:
             plan_id: The plan ULID
 
@@ -1691,41 +1736,65 @@ class PlanLifecycleService:
             NotFoundError: If plan not found
             InvalidPhaseTransitionError: If transition is not valid
             PlanNotReadyError: If plan is not ready for transition
+            LockConflictError: If another session holds the plan lock
         """
-        plan = self.get_plan(plan_id)
+        # Generate a unique owner_id for this invocation to ensure that
+        # concurrent sessions cannot re-entrantly acquire the same lock.
+        # The LockService treats owner_id as the caller identity and allows
+        # re-entrant acquisition for the same owner; using a unique UUID per
+        # invocation ensures that concurrent sessions present different owners
+        # and thus trigger LockConflictError when attempting to acquire the
+        # same plan lock.
+        owner_id: str = str(uuid4())
 
-        # Validate transition
-        if not can_transition(plan.phase, PlanPhase.APPLY):
-            raise InvalidPhaseTransitionError(plan.phase, PlanPhase.APPLY)
+        if self._lock_service is not None:
+            self._lock_service.acquire(
+                owner_id=owner_id,
+                resource_type="plan",
+                resource_id=plan_id,
+            )
+        try:
+            plan = self.get_plan(plan_id)
 
-        # Must be in COMPLETE state to transition
-        if plan.state != ProcessingState.COMPLETE:
-            raise PlanNotReadyError(
-                plan_id, plan.phase, plan.state or ProcessingState.QUEUED
+            # Validate transition
+            if not can_transition(plan.phase, PlanPhase.APPLY):
+                raise InvalidPhaseTransitionError(plan.phase, PlanPhase.APPLY)
+
+            # Must be in COMPLETE state to transition
+            if plan.state != ProcessingState.COMPLETE:
+                raise PlanNotReadyError(
+                    plan_id, plan.phase, plan.state or ProcessingState.QUEUED
+                )
+
+            # Invariant Reconciliation: verify invariants before Apply
+            self._run_invariant_reconciliation(plan)
+
+            # Transition to Apply phase — set processing_state first so that
+            # the phase-state validator sees QUEUED (valid in any phase) when
+            # the phase assignment triggers re-validation.
+            plan.processing_state = ProcessingState.QUEUED
+            plan.phase = PlanPhase.APPLY
+            plan.timestamps.apply_started_at = datetime.now()
+            plan.timestamps.updated_at = datetime.now()
+
+            self._commit_plan(plan)
+            self._logger.info(
+                "Plan transitioned to Apply",
+                plan_id=plan_id,
+                phase=plan.phase.value,
             )
 
-        # Invariant Reconciliation: verify invariants before Apply
-        self._run_invariant_reconciliation(plan)
+            # Enqueue async job when async execution is enabled
+            self._maybe_enqueue_async_job(plan_id, "apply")
 
-        # Transition to Apply phase — set processing_state first so that
-        # the phase-state validator sees QUEUED (valid in any phase) when
-        # the phase assignment triggers re-validation.
-        plan.processing_state = ProcessingState.QUEUED
-        plan.phase = PlanPhase.APPLY
-        plan.timestamps.apply_started_at = datetime.now()
-        plan.timestamps.updated_at = datetime.now()
-
-        self._commit_plan(plan)
-        self._logger.info(
-            "Plan transitioned to Apply",
-            plan_id=plan_id,
-            phase=plan.phase.value,
-        )
-
-        # Enqueue async job when async execution is enabled
-        self._maybe_enqueue_async_job(plan_id, "apply")
-
-        return plan
+            return plan
+        finally:
+            if self._lock_service is not None:
+                self._lock_service.release(
+                    owner_id=owner_id,
+                    resource_type="plan",
+                    resource_id=plan_id,
+                )
 
     def start_apply(self, plan_id: str) -> Plan:
         """Start the Apply phase processing."""