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

@@ -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

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,
                 )