diff --git a/CHANGELOG.md b/CHANGELOG.md
index 2bc24e812..ebed3f6f6 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -150,6 +150,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 0e48a6668..6448d0507 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
@@ -1527,9 +1528,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,
             )
@@ -1597,7 +1607,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,
                 )
@@ -1731,9 +1741,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,
             )
@@ -1775,7 +1794,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,
                 )