From b83f575cb5878bd4dc823a1c40c03087a08a58c1 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 13 Apr 2026 02:11:39 +0000 Subject: [PATCH 1/3] =?UTF-8?q?fix(concurrency):=20wire=20LockService=20in?= =?UTF-8?q?to=20plan=20lifecycle=20=E2=80=94=20guard=20execute=5Fplan=20an?= =?UTF-8?q?d=20apply=5Fplan?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit LockService was implemented but never integrated into the plan execution path, leaving execute_plan() and apply_plan() unprotected against concurrent calls on the same plan_id (race condition, issue #7989). Changes: - container.py: add _build_lock_service() factory and register LockService as a Singleton provider; inject it into PlanLifecycleService via the DI container. - plan_lifecycle_service.py: accept optional lock_service parameter in __init__; in execute_plan() and apply_plan() acquire a plan-level advisory lock before the critical section and release it in a finally block so the lock is always freed even when exceptions occur. When lock_service is None (existing tests without DI wiring) the behaviour is unchanged — locking is silently skipped for backward compatibility. Closes #7989 --- src/cleveragents/application/container.py | 26 +++ .../services/plan_lifecycle_service.py | 210 +++++++++++------- 2 files changed, 156 insertions(+), 80 deletions(-) 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..e66739d12 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -101,6 +101,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 +191,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 +227,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 +243,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 +1507,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 +1522,82 @@ 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) - - # 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=plan_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=plan_id, + resource_type="plan", + resource_id=plan_id, + ) def start_execute(self, plan_id: str) -> Plan: """Start the Execute phase processing.""" @@ -1681,6 +1711,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 +1726,56 @@ 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) + if self._lock_service is not None: + self._lock_service.acquire( + owner_id=plan_id, + resource_type="plan", + resource_id=plan_id, + ) + try: + plan = self.get_plan(plan_id) - # Validate transition - if not can_transition(plan.phase, PlanPhase.APPLY): - raise InvalidPhaseTransitionError(plan.phase, PlanPhase.APPLY) + # 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 + # 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=plan_id, + resource_type="plan", + resource_id=plan_id, + ) def start_apply(self, plan_id: str) -> Plan: """Start the Apply phase processing.""" -- 2.52.0 From b1f7b51a8022007865736b114691ad813d12eaba Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 13 Apr 2026 17:37:35 +0000 Subject: [PATCH 2/3] fix(concurrency): fix lock owner identity to prevent re-entrant acquisition The original implementation used plan_id as the owner_id when acquiring the advisory lock. Because LockService treats owner_id as the caller identity and allows re-entrant acquisition for the same owner, concurrent sessions attempting to lock the same plan would all present the same owner_id and thus silently renew the lock instead of raising LockConflictError. This fix generates a unique UUID for each invocation as the owner_id, ensuring that concurrent sessions present different owners and thus trigger LockConflictError when attempting to acquire the same plan lock. The lock is still acquired before the phase transition and released in a finally block to ensure cleanup even on error. ISSUES CLOSED: #8067 --- CHANGELOG.md | 8 ++++++ .../services/plan_lifecycle_service.py | 27 ++++++++++++++++--- 2 files changed, 31 insertions(+), 4 deletions(-) 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/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index e66739d12..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 @@ -1524,9 +1525,18 @@ class PlanLifecycleService: PlanNotReadyError: If plan is not ready for transition LockConflictError: If another session holds the plan lock """ + # 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()) + if self._lock_service is not None: self._lock_service.acquire( - owner_id=plan_id, + owner_id=owner_id, resource_type="plan", resource_id=plan_id, ) @@ -1594,7 +1604,7 @@ class PlanLifecycleService: finally: if self._lock_service is not None: self._lock_service.release( - owner_id=plan_id, + owner_id=owner_id, resource_type="plan", resource_id=plan_id, ) @@ -1728,9 +1738,18 @@ class PlanLifecycleService: PlanNotReadyError: If plan is not ready for transition LockConflictError: If another session holds the plan lock """ + # 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()) + if self._lock_service is not None: self._lock_service.acquire( - owner_id=plan_id, + owner_id=owner_id, resource_type="plan", resource_id=plan_id, ) @@ -1772,7 +1791,7 @@ class PlanLifecycleService: finally: if self._lock_service is not None: self._lock_service.release( - owner_id=plan_id, + owner_id=owner_id, resource_type="plan", resource_id=plan_id, ) -- 2.52.0 From e757ca9db0b898498ec0d8be8a73a22e1b448b70 Mon Sep 17 00:00:00 2001 From: CleverThis Date: Mon, 13 Apr 2026 19:16:56 +0000 Subject: [PATCH 3/3] docs(contributors): add HAL 9000 concurrency-fix contribution detail MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add a Details entry for HAL 9000 describing the plan lifecycle concurrency race-condition fix (#7989) — wiring LockService into execute_plan/apply_plan with unique per-invocation owner identities. ISSUES CLOSED: #7989 --- CONTRIBUTORS.md | 1 + 1 file changed, 1 insertion(+) 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. -- 2.52.0