cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

fix(concurrency): wire LockService into plan lifecycle — guard execute_plan and apply_plan

Browse Source 
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
This commit is contained in:
HAL 9000
2026-04-13 02:11:39 +00:00
parent 38bcd41338
commit b83f575cb5
2 changed files with 156 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/cleveragents/application/container.py
+26
View File
@@ -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
+130 -80
View File
@@ -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."""