diff --git a/CHANGELOG.md b/CHANGELOG.md index 7907767ad..0cf791e12 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/CONTRIBUTORS.md b/CONTRIBUTORS.md index 62a76a4bc..31d4c882f 100644 --- a/CONTRIBUTORS.md +++ b/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. diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 69f6b7ed4..e86956146 100644 --- a/src/cleveragents/application/container.py +++ b/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 diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 846c110e5..b3b28aa2d 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/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."""