feat(concurrency): add plan and project locks

Implemented plan-level and project-level locking with configurable timeouts.
Added locks table via Alembic migration storing owner_id, resource_type,
resource_id, acquired_at, and expires_at. Locks enforced in
PlanLifecycleService transitions. Support for re-entrant acquisition,
lock renewal, graceful shutdown release, and startup cleanup of expired
locks. Added diagnostics check for stale lock reporting.

ISSUES CLOSED: #327

CHANGELOG.md

@@ -2,6 +2,9 @@
 
 ## Unreleased
 
+- Added plan-level and project-level advisory locking with configurable timeouts, re-entrant
+  acquisition, conflict detection, lock renewal, graceful shutdown release, startup cleanup of
+  expired locks, and diagnostics check for stale lock reporting. (#327)
 - Added core plan apply service with diff review output (plain, rich, JSON, YAML), artifact
   summaries, apply summary persistence, merge-failure handling with sandbox rollback, and
   empty ChangeSet guard. (#155)

alembic/versions/m4_001_concurrency_locks.py

@@ -0,0 +1,71 @@
+"""Add concurrency locks table.
+
+This migration creates the ``locks`` table for plan-level and
+project-level concurrency locks with timeout support.
+
+The table stores the lock owner, resource type/id pair, and acquisition
+and expiry timestamps.  A unique constraint on (resource_type,
+resource_id) ensures at most one active lock per resource.
+
+Revision ID: m4_001_concurrency_locks
+Revises: c0_002_merge_skill_registry
+Create Date: 2026-02-24 12:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "m4_001_concurrency_locks"
+down_revision: str | Sequence[str] | None = "c0_002_merge_skill_registry"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create locks table for concurrency control."""
+    op.create_table(
+        "locks",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column("owner_id", sa.String(255), nullable=False),
+        sa.Column("resource_type", sa.String(50), nullable=False),
+        sa.Column("resource_id", sa.String(255), nullable=False),
+        sa.Column("acquired_at", sa.String(30), nullable=False),
+        sa.Column("expires_at", sa.String(30), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "resource_type",
+            "resource_id",
+            name="uq_locks_resource",
+        ),
+    )
+
+    op.create_index(
+        "ix_locks_owner_id",
+        "locks",
+        ["owner_id"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_locks_expires_at",
+        "locks",
+        ["expires_at"],
+        unique=False,
+    )
+    op.create_index(
+        "ix_locks_resource_type",
+        "locks",
+        ["resource_type"],
+        unique=False,
+    )
+
+
+def downgrade() -> None:
+    """Drop locks table."""
+    op.drop_index("ix_locks_resource_type", table_name="locks")
+    op.drop_index("ix_locks_expires_at", table_name="locks")
+    op.drop_index("ix_locks_owner_id", table_name="locks")
+    op.drop_table("locks")

benchmarks/agent_skills_registry_bench.py

@@ -13,6 +13,7 @@ import shutil
 import sys
 import tempfile
 from pathlib import Path
+from typing import ClassVar
 
 try:
     from cleveragents.skills.discovery import (
@@ -101,8 +102,8 @@ class BuildToolSpecSuite:
 class RegisterSuite:
     """Benchmark tool registration with conflict handling."""
 
-    params: list[int] = [10, 50, 100]
-    param_names: list[str] = ["count"]
+    params: ClassVar[list[int]] = [10, 50, 100]
+    param_names: ClassVar[list[str]] = ["count"]
 
     def setup(self, count: int) -> None:
         self.skills = [_make_discovered(f"tool-{i}") for i in range(count)]

benchmarks/concurrency_lock_bench.py

@@ -0,0 +1,144 @@
+"""ASV benchmarks for concurrency lock overhead.
+
+Measures the cost of lock acquire, release, renew, and cleanup
+operations against an in-memory SQLite database.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.application.services.lock_service import LockService
+from cleveragents.infrastructure.database.models import Base, LockModel
+
+_bench_counter = 9000
+
+
+def _next_id() -> str:
+    global _bench_counter
+    _bench_counter += 1
+    return f"bench-res-{_bench_counter}"
+
+
+def _build_lock_service() -> tuple[LockService, sessionmaker[Session]]:
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        future=True,
+        connect_args={"check_same_thread": False},
+    )
+
+    @event.listens_for(engine, "connect")
+    def _fk(dbapi_conn: Any, _rec: Any) -> None:
+        cursor = dbapi_conn.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    Base.metadata.create_all(engine)
+    sf: sessionmaker[Session] = sessionmaker(
+        bind=engine,
+        expire_on_commit=False,
+        autoflush=False,
+        autocommit=False,
+        class_=Session,
+    )
+    return LockService(session_factory=sf), sf
+
+
+class TimeAcquireLock:
+    """Benchmark acquiring a new lock."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.svc, self._sf = _build_lock_service()
+
+    def time_acquire(self) -> None:
+        rid = _next_id()
+        self.svc.acquire("owner-bench", "plan", rid)
+
+
+class TimeReentrantAcquire:
+    """Benchmark re-entrant lock acquisition."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.svc, self._sf = _build_lock_service()
+        self.svc.acquire("owner-bench", "plan", "reentrant-res")
+
+    def time_reentrant_acquire(self) -> None:
+        self.svc.acquire("owner-bench", "plan", "reentrant-res")
+
+
+class TimeReleaseLock:
+    """Benchmark releasing a lock."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.svc, self._sf = _build_lock_service()
+        self._rid = _next_id()
+        self.svc.acquire("owner-bench", "plan", self._rid)
+
+    def time_release(self) -> None:
+        self.svc.release("owner-bench", "plan", self._rid)
+        self.svc.acquire("owner-bench", "plan", self._rid)
+
+
+class TimeRenewLock:
+    """Benchmark renewing a lock."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.svc, self._sf = _build_lock_service()
+        self.svc.acquire("owner-bench", "plan", "renew-res")
+
+    def time_renew(self) -> None:
+        self.svc.renew("owner-bench", "plan", "renew-res")
+
+
+class TimeCleanupExpired:
+    """Benchmark expired lock cleanup."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.svc, self._sf = _build_lock_service()
+        now = datetime.now(tz=UTC)
+        expired = (now - timedelta(seconds=60)).isoformat()
+        acquired = (now - timedelta(seconds=120)).isoformat()
+        session = self._sf()
+        for i in range(50):
+            session.add(
+                LockModel(
+                    owner_id=f"expired-{i}",
+                    resource_type="plan",
+                    resource_id=f"cleanup-{i}",
+                    acquired_at=acquired,
+                    expires_at=expired,
+                )
+            )
+        session.commit()
+        session.close()
+
+    def time_cleanup(self) -> None:
+        self.svc.cleanup_expired()
+
+
+class TimeIsLocked:
+    """Benchmark is_locked check."""
+
+    timeout = 30
+
+    def setup(self) -> None:
+        self.svc, self._sf = _build_lock_service()
+        self.svc.acquire("owner-bench", "plan", "locked-res")
+
+    def time_is_locked(self) -> None:
+        self.svc.is_locked("plan", "locked-res")

benchmarks/error_recovery_bench.py

@@ -11,7 +11,6 @@ from __future__ import annotations
 
 import sys
 from pathlib import Path
-from typing import Any
 from unittest.mock import MagicMock
 
 try:
@@ -20,10 +19,6 @@ try:
     )
     from cleveragents.domain.models.core.error_recovery import (
         ErrorCategory,
-        ErrorHistory,
-        ErrorRecord,
-        ErrorRecoveryPolicy,
-        RecoveryHint,
         classify_error,
         get_recovery_hints,
     )
@@ -39,10 +34,6 @@ except ModuleNotFoundError:
     )
     from cleveragents.domain.models.core.error_recovery import (
         ErrorCategory,
-        ErrorHistory,
-        ErrorRecord,
-        ErrorRecoveryPolicy,
-        RecoveryHint,
         classify_error,
         get_recovery_hints,
     )

benchmarks/validation_pipeline_bench.py

@@ -12,18 +12,23 @@ import sys
 from pathlib import Path
 from typing import Any
 
-_SRC = str(Path(__file__).resolve().parents[1] / "src")
-if _SRC not in sys.path:
-    sys.path.insert(0, _SRC)
-
-from cleveragents.application.services.validation_pipeline import (
-    ValidationCommand,
-    ValidationPipeline,
-    ValidationResult,
-    ValidationSummary,
-)
-from cleveragents.domain.models.core.tool import ValidationMode
-
+try:
+    from cleveragents.application.services.validation_pipeline import (
+        ValidationCommand,
+        ValidationPipeline,
+        ValidationResult,
+        ValidationSummary,
+    )
+    from cleveragents.domain.models.core.tool import ValidationMode
+except ModuleNotFoundError:
+    sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+    from cleveragents.application.services.validation_pipeline import (
+        ValidationCommand,
+        ValidationPipeline,
+        ValidationResult,
+        ValidationSummary,
+    )
+    from cleveragents.domain.models.core.tool import ValidationMode
 
 # ---------------------------------------------------------------------------
 # Helpers

docs/reference/concurrency.md

@@ -0,0 +1,101 @@
+# Concurrency Locks
+
+CleverAgents uses advisory locking to prevent concurrent modifications
+to plans and projects.  The lock mechanism is backed by a SQLite table
+and enforced at the application layer before lifecycle transitions.
+
+## Overview
+
+The `LockService` provides mutual-exclusion primitives for two resource
+types: **plan** and **project**.  Each resource can have at most one
+active lock at any time.  Locks are identified by the tuple
+`(resource_type, resource_id)` and scoped to an **owner** (typically
+the session or process identity).
+
+## Lock Lifecycle
+
+```text
+acquire  -->  active  -->  release / expire
+               |
+               +--> renew (extend TTL)
+```
+
+1. **Acquire**: Creates a new lock row or re-acquires for the same owner.
+2. **Renew**: Extends the TTL of an active lock.
+3. **Release**: Explicitly removes the lock.
+4. **Expire**: The lock becomes stale after `expires_at` passes.
+
+## Re-entrant Acquisition
+
+If the same owner calls `acquire()` on a resource it already holds,
+the TTL is silently extended.  No error is raised.
+
+## Conflict Handling
+
+If a **different** owner attempts to acquire a lock that is still active
+(not expired), a `LockConflictError` is raised with details about the
+current holder.  Expired locks from other owners are transparently
+replaced.
+
+## TTL Defaults and Renewal Strategy
+
+| Constant                  | Default | Description                    |
+|---------------------------|---------|--------------------------------|
+| `DEFAULT_LOCK_TTL_SECS`  | 300     | Default lock lifetime (5 min)  |
+| `MAX_LOCK_TTL_SECS`      | 3600    | Maximum allowed TTL (1 hr)     |
+| `MIN_LOCK_TTL_SECS`      | 5       | Minimum allowed TTL (5 sec)    |
+
+For long-running phases (e.g. Execute), the caller should periodically
+call `renew()` with the desired TTL before the current lock expires.
+A recommended renewal interval is `ttl / 2` (e.g. every 150 seconds
+for the default 300-second TTL).
+
+## Database Schema
+
+The `locks` table (migration `m4_001_concurrency_locks`):
+
+| Column          | Type         | Description                   |
+|-----------------|-------------|-------------------------------|
+| `id`            | INTEGER PK  | Auto-increment primary key    |
+| `owner_id`      | VARCHAR(255) | Lock owner identity           |
+| `resource_type` | VARCHAR(50)  | `plan` or `project`           |
+| `resource_id`   | VARCHAR(255) | Resource identifier           |
+| `acquired_at`   | VARCHAR(30)  | ISO-8601 acquisition time     |
+| `expires_at`    | VARCHAR(30)  | ISO-8601 expiry time          |
+
+A unique constraint on `(resource_type, resource_id)` enforces at most
+one lock per resource.
+
+## Startup Cleanup
+
+On application startup, `cleanup_expired()` is called to purge any
+locks whose `expires_at` has passed.  This handles the case where a
+previous process terminated without releasing its locks.
+
+## Graceful Shutdown
+
+During graceful shutdown, `release_all_for_owner(owner_id)` removes all
+locks held by the current process so that resources become immediately
+available to other instances.
+
+## Diagnostics
+
+The `agents diagnostics` command includes a **Stale locks** check that
+reports the number of expired-but-not-yet-cleaned locks.  A non-zero
+count indicates that the cleanup routine has not run recently and may
+warrant investigation.
+
+## Integration Points
+
+- **PlanLifecycleService**: Transitions acquire a plan-level lock before
+  mutating phase/state and release it after persistence.
+- **Startup**: The application container calls `cleanup_expired()` during
+  initialisation.
+
+## Error Types
+
+| Exception           | Condition                                      |
+|---------------------|------------------------------------------------|
+| `LockConflictError` | Another owner holds an active lock             |
+| `LockExpiredError`  | Attempted to renew an already-expired lock     |
+| `ValidationError`   | Invalid parameters (empty strings, bad TTL)    |

features/concurrency.feature

@@ -0,0 +1,139 @@
+Feature: Concurrency Locks
+  As a developer
+  I want plan-level and project-level advisory locks
+  So that concurrent modifications to shared resources are prevented
+
+  Background:
+    Given I have a lock service backed by an in-memory database
+
+  # --- Acquisition ---
+
+  Scenario: Acquire a plan lock
+    When I acquire a lock on plan "plan-001" as owner "session-a"
+    Then the lock should be acquired successfully
+
+  Scenario: Acquire a project lock
+    When I acquire a lock on project "local/my-proj" as owner "session-a"
+    Then the lock should be acquired successfully
+
+  # --- Re-entrant acquisition ---
+
+  Scenario: Same owner re-acquires a plan lock
+    Given owner "session-a" holds a lock on plan "plan-001"
+    When I acquire a lock on plan "plan-001" as owner "session-a"
+    Then the lock should be acquired successfully
+
+  # --- Conflict detection ---
+
+  Scenario: Different owner is rejected with LockConflictError
+    Given owner "session-a" holds a lock on plan "plan-001"
+    When I try to acquire a lock on plan "plan-001" as owner "session-b"
+    Then a lock conflict error should be raised for owner "session-a"
+
+  # --- Release ---
+
+  Scenario: Release a held lock
+    Given owner "session-a" holds a lock on plan "plan-001"
+    When I release the lock on plan "plan-001" as owner "session-a"
+    Then the lock should be released successfully
+
+  Scenario: Release returns false for non-existent lock
+    When I release the lock on plan "plan-999" as owner "nobody"
+    Then the release result should be false
+
+  # --- Renewal ---
+
+  Scenario: Renew a held lock
+    Given owner "session-a" holds a lock on plan "plan-001"
+    When I renew the lock on plan "plan-001" as owner "session-a"
+    Then the lock renewal should succeed
+
+  Scenario: Renew an expired lock raises LockExpiredError
+    Given owner "session-a" held an expired lock on plan "plan-001"
+    When I try to renew the expired lock on plan "plan-001" as owner "session-a"
+    Then a lock expired error should be raised
+
+  # --- Expiry and cleanup ---
+
+  Scenario: Expired lock is replaced by a new owner
+    Given owner "session-a" held an expired lock on plan "plan-001"
+    When I acquire a lock on plan "plan-001" as owner "session-b"
+    Then the lock should be acquired successfully
+
+  Scenario: Cleanup expired locks on startup
+    Given there are 3 expired locks in the database
+    When I run cleanup expired locks
+    Then the expired lock count should be 3
+
+  # --- Graceful shutdown ---
+
+  Scenario: Release all locks for an owner on shutdown
+    Given owner "session-a" holds locks on plans "p1" and "p2" and project "proj-1"
+    When I release all locks for owner "session-a"
+    Then the released count should be 3
+
+  # --- Diagnostics ---
+
+  Scenario: Count stale locks for diagnostics
+    Given there are 2 expired locks in the database
+    When I count stale locks
+    Then the stale lock count should be 2
+
+  # --- is_locked check ---
+
+  Scenario: is_locked returns true for active lock
+    Given owner "session-a" holds a lock on plan "plan-001"
+    When I check if plan "plan-001" is locked
+    Then the resource should be locked
+
+  Scenario: is_locked returns false when no lock exists
+    When I check if plan "plan-999" is locked
+    Then the resource should not be locked
+
+  # --- Validation ---
+
+  Scenario: Reject empty owner_id
+    When I try to acquire a lock with empty owner_id
+    Then a lock validation error should be raised
+
+  Scenario: Reject invalid resource type
+    When I try to acquire a lock with invalid resource type "widget"
+    Then a lock validation error should be raised
+
+  Scenario: Reject empty resource_id
+    When I try to acquire a lock with empty resource_id
+    Then a lock validation error should be raised
+
+  Scenario: Reject TTL below minimum
+    When I try to acquire a lock with TTL 2
+    Then a lock validation error should be raised
+
+  Scenario: Reject TTL above maximum
+    When I try to acquire a lock with TTL 7200
+    Then a lock validation error should be raised
+
+  # --- Additional validation ---
+
+  Scenario: Reject None session_factory
+    When I try to create a lock service with None session factory
+    Then a lock validation error should be raised
+
+  Scenario: Reject empty resource_type on acquire
+    When I try to acquire a lock with empty resource type
+    Then a lock validation error should be raised
+
+  Scenario: Reject empty owner_id on release
+    When I try to release a lock with empty owner_id
+    Then a lock validation error should be raised
+
+  Scenario: Reject empty owner_id on renew
+    When I try to renew a lock with empty owner_id
+    Then a lock validation error should be raised
+
+  Scenario: Reject empty owner_id on release_all_for_owner
+    When I try to release all locks with empty owner_id
+    Then a lock validation error should be raised
+
+  Scenario: Renew returns false for nonexistent lock
+    When I try to renew a lock that does not exist
+    Then the renew result should be false

features/steps/concurrency_steps.py

@@ -0,0 +1,452 @@
+"""Step definitions for concurrency lock scenarios."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.application.services.lock_service import LockService
+from cleveragents.core.exceptions import (
+    LockConflictError,
+    LockExpiredError,
+    ValidationError,
+)
+from cleveragents.infrastructure.database.models import Base, LockModel
+
+
+def _build_lock_service(context: Context) -> LockService:
+    """Build a LockService with an in-memory SQLite backend."""
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        future=True,
+        connect_args={"check_same_thread": False},
+    )
+
+    @event.listens_for(engine, "connect")
+    def _fk(dbapi_conn: Any, _rec: Any) -> None:
+        cursor = dbapi_conn.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(
+        bind=engine,
+        expire_on_commit=False,
+        autoflush=False,
+        autocommit=False,
+        class_=Session,
+    )
+    context.lock_engine = engine
+    context.lock_session_factory = factory
+    return LockService(session_factory=factory)
+
+
+# --- Background ---
+
+
+@given("I have a lock service backed by an in-memory database")
+def step_create_lock_service(context: Context) -> None:
+    """Create a LockService backed by an in-memory SQLite database."""
+    context.lock_service = _build_lock_service(context)
+    context.error = None
+    context.result = None
+
+
+# --- Acquisition ---
+
+
+@when('I acquire a lock on plan "{plan_id}" as owner "{owner}"')
+def step_acquire_plan_lock(context: Context, plan_id: str, owner: str) -> None:
+    """Acquire a plan-level lock."""
+    context.result = context.lock_service.acquire(
+        owner_id=owner,
+        resource_type="plan",
+        resource_id=plan_id,
+    )
+
+
+@when('I acquire a lock on project "{project_id}" as owner "{owner}"')
+def step_acquire_project_lock(context: Context, project_id: str, owner: str) -> None:
+    """Acquire a project-level lock."""
+    context.result = context.lock_service.acquire(
+        owner_id=owner,
+        resource_type="project",
+        resource_id=project_id,
+    )
+
+
+@then("the lock should be acquired successfully")
+def step_lock_acquired(context: Context) -> None:
+    """Assert the lock was acquired."""
+    assert context.result is True, f"Expected True, got {context.result}"
+
+
+# --- Re-entrant / fixtures ---
+
+
+@given('owner "{owner}" holds a lock on plan "{plan_id}"')
+def step_owner_holds_plan_lock(context: Context, owner: str, plan_id: str) -> None:
+    """Ensure the specified owner holds a plan lock."""
+    context.lock_service.acquire(
+        owner_id=owner,
+        resource_type="plan",
+        resource_id=plan_id,
+    )
+
+
+@given('owner "{owner}" holds locks on plans "{p1}" and "{p2}" and project "{proj}"')
+def step_owner_holds_multiple(
+    context: Context, owner: str, p1: str, p2: str, proj: str
+) -> None:
+    """Ensure the owner holds multiple locks."""
+    svc: LockService = context.lock_service
+    svc.acquire(owner_id=owner, resource_type="plan", resource_id=p1)
+    svc.acquire(owner_id=owner, resource_type="plan", resource_id=p2)
+    svc.acquire(owner_id=owner, resource_type="project", resource_id=proj)
+
+
+# --- Conflict ---
+
+
+@when('I try to acquire a lock on plan "{plan_id}" as owner "{owner}"')
+def step_try_acquire_plan_lock(context: Context, plan_id: str, owner: str) -> None:
+    """Try to acquire a plan lock, capturing any error."""
+    try:
+        context.result = context.lock_service.acquire(
+            owner_id=owner,
+            resource_type="plan",
+            resource_id=plan_id,
+        )
+    except LockConflictError as exc:
+        context.error = exc
+
+
+@then('a lock conflict error should be raised for owner "{expected_owner}"')
+def step_lock_conflict_error(context: Context, expected_owner: str) -> None:
+    """Assert a LockConflictError was raised."""
+    assert isinstance(context.error, LockConflictError), (
+        f"Expected LockConflictError, got {type(context.error)}"
+    )
+    assert context.error.owner_id == expected_owner
+
+
+# --- Release ---
+
+
+@when('I release the lock on plan "{plan_id}" as owner "{owner}"')
+def step_release_plan_lock(context: Context, plan_id: str, owner: str) -> None:
+    """Release a plan lock."""
+    context.result = context.lock_service.release(
+        owner_id=owner,
+        resource_type="plan",
+        resource_id=plan_id,
+    )
+
+
+@then("the lock should be released successfully")
+def step_lock_released(context: Context) -> None:
+    """Assert the lock was released."""
+    assert context.result is True, f"Expected True, got {context.result}"
+
+
+@then("the release result should be false")
+def step_release_false(context: Context) -> None:
+    """Assert the release returned False."""
+    assert context.result is False, f"Expected False, got {context.result}"
+
+
+# --- Renewal ---
+
+
+@when('I renew the lock on plan "{plan_id}" as owner "{owner}"')
+def step_renew_lock(context: Context, plan_id: str, owner: str) -> None:
+    """Renew a plan lock."""
+    context.result = context.lock_service.renew(
+        owner_id=owner,
+        resource_type="plan",
+        resource_id=plan_id,
+    )
+
+
+@then("the lock renewal should succeed")
+def step_renewal_success(context: Context) -> None:
+    """Assert the renewal succeeded."""
+    assert context.result is True, f"Expected True, got {context.result}"
+
+
+# --- Expired lock fixtures ---
+
+
+@given('owner "{owner}" held an expired lock on plan "{plan_id}"')
+def step_create_expired_lock(context: Context, owner: str, plan_id: str) -> None:
+    """Insert an already-expired lock directly into the database."""
+    now = datetime.now(tz=UTC)
+    expired_at = (now - timedelta(seconds=60)).isoformat()
+    acquired_at = (now - timedelta(seconds=120)).isoformat()
+
+    session: Session = context.lock_session_factory()
+    lock = LockModel(
+        owner_id=owner,
+        resource_type="plan",
+        resource_id=plan_id,
+        acquired_at=acquired_at,
+        expires_at=expired_at,
+    )
+    session.add(lock)
+    session.commit()
+    session.close()
+
+
+@when('I try to renew the expired lock on plan "{plan_id}" as owner "{owner}"')
+def step_try_renew_expired(context: Context, plan_id: str, owner: str) -> None:
+    """Attempt to renew an expired lock."""
+    try:
+        context.result = context.lock_service.renew(
+            owner_id=owner,
+            resource_type="plan",
+            resource_id=plan_id,
+        )
+    except LockExpiredError as exc:
+        context.error = exc
+
+
+@then("a lock expired error should be raised")
+def step_lock_expired_error(context: Context) -> None:
+    """Assert a LockExpiredError was raised."""
+    assert isinstance(context.error, LockExpiredError), (
+        f"Expected LockExpiredError, got {type(context.error)}"
+    )
+
+
+# --- Cleanup ---
+
+
+@given("there are {count:d} expired locks in the database")
+def step_insert_expired_locks(context: Context, count: int) -> None:
+    """Insert multiple expired locks into the database."""
+    now = datetime.now(tz=UTC)
+    expired_at = (now - timedelta(seconds=60)).isoformat()
+    acquired_at = (now - timedelta(seconds=120)).isoformat()
+
+    session: Session = context.lock_session_factory()
+    for i in range(count):
+        session.add(
+            LockModel(
+                owner_id=f"expired-owner-{i}",
+                resource_type="plan",
+                resource_id=f"expired-plan-{i}",
+                acquired_at=acquired_at,
+                expires_at=expired_at,
+            )
+        )
+    session.commit()
+    session.close()
+
+
+@when("I run cleanup expired locks")
+def step_cleanup_expired(context: Context) -> None:
+    """Run cleanup of expired locks."""
+    context.result = context.lock_service.cleanup_expired()
+
+
+@then("the expired lock count should be {count:d}")
+def step_expired_count(context: Context, count: int) -> None:
+    """Assert the cleanup purged the expected count."""
+    assert context.result == count, f"Expected {count}, got {context.result}"
+
+
+# --- Graceful shutdown ---
+
+
+@when('I release all locks for owner "{owner}"')
+def step_release_all(context: Context, owner: str) -> None:
+    """Release all locks for an owner."""
+    context.result = context.lock_service.release_all_for_owner(owner_id=owner)
+
+
+@then("the released count should be {count:d}")
+def step_released_count(context: Context, count: int) -> None:
+    """Assert the release-all count."""
+    assert context.result == count, f"Expected {count}, got {context.result}"
+
+
+# --- Diagnostics ---
+
+
+@when("I count stale locks")
+def step_count_stale(context: Context) -> None:
+    """Count stale locks."""
+    context.result = context.lock_service.count_stale_locks()
+
+
+@then("the stale lock count should be {count:d}")
+def step_stale_count(context: Context, count: int) -> None:
+    """Assert the stale lock count."""
+    assert context.result == count, f"Expected {count}, got {context.result}"
+
+
+# --- is_locked ---
+
+
+@when('I check if plan "{plan_id}" is locked')
+def step_check_is_locked(context: Context, plan_id: str) -> None:
+    """Check if a plan resource is locked."""
+    context.result = context.lock_service.is_locked(
+        resource_type="plan",
+        resource_id=plan_id,
+    )
+
+
+@then("the resource should be locked")
+def step_resource_locked(context: Context) -> None:
+    """Assert the resource is locked."""
+    assert context.result is True, f"Expected True, got {context.result}"
+
+
+@then("the resource should not be locked")
+def step_resource_not_locked(context: Context) -> None:
+    """Assert the resource is not locked."""
+    assert context.result is False, f"Expected False, got {context.result}"
+
+
+# --- Validation errors ---
+
+
+@when("I try to acquire a lock with empty owner_id")
+def step_try_empty_owner(context: Context) -> None:
+    """Try to acquire with empty owner_id."""
+    try:
+        context.lock_service.acquire(
+            owner_id="",
+            resource_type="plan",
+            resource_id="plan-001",
+        )
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when('I try to acquire a lock with invalid resource type "{rtype}"')
+def step_try_invalid_resource_type(context: Context, rtype: str) -> None:
+    """Try to acquire with an invalid resource type."""
+    try:
+        context.lock_service.acquire(
+            owner_id="session-a",
+            resource_type=rtype,
+            resource_id="res-001",
+        )
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to acquire a lock with empty resource_id")
+def step_try_empty_resource_id(context: Context) -> None:
+    """Try to acquire with empty resource_id."""
+    try:
+        context.lock_service.acquire(
+            owner_id="session-a",
+            resource_type="plan",
+            resource_id="",
+        )
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to acquire a lock with TTL {ttl:d}")
+def step_try_bad_ttl(context: Context, ttl: int) -> None:
+    """Try to acquire with out-of-range TTL."""
+    try:
+        context.lock_service.acquire(
+            owner_id="session-a",
+            resource_type="plan",
+            resource_id="plan-001",
+            ttl_seconds=ttl,
+        )
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to create a lock service with None session factory")
+def step_try_none_factory(context: Context) -> None:
+    """Try to create a LockService with None."""
+    try:
+        LockService(session_factory=None)  # type: ignore[arg-type]
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to acquire a lock with empty resource type")
+def step_try_empty_resource_type(context: Context) -> None:
+    """Try to acquire with empty resource_type."""
+    try:
+        context.lock_service.acquire(
+            owner_id="session-a",
+            resource_type="",
+            resource_id="plan-001",
+        )
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to release a lock with empty owner_id")
+def step_try_release_empty_owner(context: Context) -> None:
+    """Try to release with empty owner_id."""
+    try:
+        context.lock_service.release(
+            owner_id="",
+            resource_type="plan",
+            resource_id="plan-001",
+        )
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to renew a lock with empty owner_id")
+def step_try_renew_empty_owner(context: Context) -> None:
+    """Try to renew with empty owner_id."""
+    try:
+        context.lock_service.renew(
+            owner_id="",
+            resource_type="plan",
+            resource_id="plan-001",
+        )
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to release all locks with empty owner_id")
+def step_try_release_all_empty_owner(context: Context) -> None:
+    """Try to release_all_for_owner with empty owner_id."""
+    try:
+        context.lock_service.release_all_for_owner(owner_id="")
+    except ValidationError as exc:
+        context.error = exc
+
+
+@when("I try to renew a lock that does not exist")
+def step_try_renew_nonexistent(context: Context) -> None:
+    """Try to renew a lock that doesn't exist."""
+    context.result = context.lock_service.renew(
+        owner_id="no-one",
+        resource_type="plan",
+        resource_id="no-such-lock",
+    )
+
+
+@then("the renew result should be false")
+def step_renew_false(context: Context) -> None:
+    """Assert renew returned False."""
+    assert context.result is False, f"Expected False, got {context.result}"
+
+
+@then("a lock validation error should be raised")
+def step_lock_validation_error(context: Context) -> None:
+    """Assert a ValidationError was raised for lock operations."""
+    assert isinstance(context.error, ValidationError), (
+        f"Expected ValidationError, got {type(context.error)}"
+    )

robot/concurrency_locks.robot

@@ -0,0 +1,17 @@
+*** Settings ***
+Documentation    Smoke tests for concurrency lock service
+Library          Process
+Library          OperatingSystem
+Library          String
+
+*** Variables ***
+${PYTHON}    python
+
+*** Test Cases ***
+Lock Acquire Release Smoke Test
+    [Documentation]    Acquire and release a concurrency lock via helper
+    ${result}=    Run Process    ${PYTHON}    ${CURDIR}/helper_concurrency_locks.py
+    ...    stderr=STDOUT    timeout=30s
+    Log    ${result.stdout}
+    Should Contain    ${result.stdout}    PASS: concurrency_locks smoke test
+    Should Be Equal As Integers    ${result.rc}    0

robot/helper_concurrency_locks.py

@@ -0,0 +1,135 @@
+"""Helper script for concurrency_locks.robot smoke test."""
+
+from __future__ import annotations
+
+import sys
+from datetime import UTC, datetime, timedelta
+from typing import Any
+
+sys.path.insert(0, "src")
+sys.path.insert(0, ".")
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.application.services.lock_service import LockService
+from cleveragents.core.exceptions import LockConflictError, LockExpiredError
+from cleveragents.infrastructure.database.models import Base, LockModel
+
+
+def main() -> None:
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        future=True,
+        connect_args={"check_same_thread": False},
+    )
+
+    @event.listens_for(engine, "connect")
+    def _fk(dbapi_conn: Any, _rec: Any) -> None:
+        cursor = dbapi_conn.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    Base.metadata.create_all(engine)
+    sf: sessionmaker[Session] = sessionmaker(
+        bind=engine,
+        expire_on_commit=False,
+        autoflush=False,
+        autocommit=False,
+        class_=Session,
+    )
+
+    svc = LockService(session_factory=sf)
+
+    # Test 1: basic acquire
+    assert svc.acquire("owner-1", "plan", "p1") is True
+    print("PASS: acquire")
+
+    # Test 2: re-entrant
+    assert svc.acquire("owner-1", "plan", "p1") is True
+    print("PASS: re-entrant acquire")
+
+    # Test 3: conflict
+    try:
+        svc.acquire("owner-2", "plan", "p1")
+        print("FAIL: expected LockConflictError")
+        sys.exit(1)
+    except LockConflictError:
+        print("PASS: conflict detection")
+
+    # Test 4: release
+    assert svc.release("owner-1", "plan", "p1") is True
+    print("PASS: release")
+
+    # Test 5: acquire after release
+    assert svc.acquire("owner-2", "plan", "p1") is True
+    print("PASS: acquire after release")
+
+    # Test 6: renew
+    assert svc.renew("owner-2", "plan", "p1") is True
+    print("PASS: renew")
+
+    # Test 7: release_all_for_owner
+    svc.acquire("owner-3", "plan", "p10")
+    svc.acquire("owner-3", "project", "proj-1")
+    count = svc.release_all_for_owner("owner-3")
+    assert count == 2, f"Expected 2, got {count}"
+    print("PASS: release_all_for_owner")
+
+    # Test 8: cleanup_expired
+    now = datetime.now(tz=UTC)
+    expired = (now - timedelta(seconds=60)).isoformat()
+    acquired = (now - timedelta(seconds=120)).isoformat()
+    session = sf()
+    session.add(
+        LockModel(
+            owner_id="old",
+            resource_type="plan",
+            resource_id="stale-1",
+            acquired_at=acquired,
+            expires_at=expired,
+        )
+    )
+    session.commit()
+    session.close()
+
+    cleaned = svc.cleanup_expired()
+    assert cleaned >= 1, f"Expected >=1 cleaned, got {cleaned}"
+    print("PASS: cleanup_expired")
+
+    # Test 9: count_stale_locks returns 0 after cleanup
+    assert svc.count_stale_locks() == 0
+    print("PASS: count_stale_locks")
+
+    # Test 10: is_locked
+    svc.acquire("owner-4", "plan", "check-me")
+    assert svc.is_locked("plan", "check-me") is True
+    assert svc.is_locked("plan", "not-locked") is False
+    print("PASS: is_locked")
+
+    # Test 11: renew expired raises LockExpiredError
+    session2 = sf()
+    session2.add(
+        LockModel(
+            owner_id="owner-5",
+            resource_type="plan",
+            resource_id="expired-renew",
+            acquired_at=acquired,
+            expires_at=expired,
+        )
+    )
+    session2.commit()
+    session2.close()
+    try:
+        svc.renew("owner-5", "plan", "expired-renew")
+        print("FAIL: expected LockExpiredError")
+        sys.exit(1)
+    except LockExpiredError:
+        print("PASS: renew expired raises error")
+
+    print("\nPASS: concurrency_locks smoke test")
+
+
+if __name__ == "__main__":
+    main()

src/cleveragents/application/services/lock_service.py

@@ -0,0 +1,448 @@
+"""Concurrency lock service for plan and project resources.
+
+The ``LockService`` provides advisory locking with configurable TTL
+for plan-level and project-level mutual exclusion.  Locks are
+persisted in the ``locks`` table and enforced before state transitions.
+
+## Features
+
+- **Re-entrant**: same owner can re-acquire a held lock (extends TTL).
+- **Conflict detection**: different owner receives ``LockConflictError``.
+- **Automatic expiry**: expired locks are transparently cleaned up.
+- **Renewal**: owners can extend TTL for long-running phases.
+- **Graceful shutdown**: ``release_all_for_owner`` cleans up on exit.
+- **Startup cleanup**: ``cleanup_expired`` purges stale rows.
+- **Diagnostics**: ``count_stale_locks`` for health checks.
+
+## Lock TTL Defaults
+
+| Constant                  | Value | Description                   |
+|---------------------------|-------|-------------------------------|
+| ``DEFAULT_LOCK_TTL_SECS`` | 300   | Default lock lifetime (5 min) |
+| ``MAX_LOCK_TTL_SECS``     | 3600  | Maximum allowed TTL (1 hr)    |
+| ``MIN_LOCK_TTL_SECS``     | 5     | Minimum allowed TTL (5 sec)   |
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from datetime import UTC, datetime, timedelta
+
+import structlog
+from sqlalchemy import delete, func, select
+from sqlalchemy.orm import Session
+
+from cleveragents.core.exceptions import (
+    LockConflictError,
+    LockExpiredError,
+    ValidationError,
+)
+from cleveragents.infrastructure.database.models import LockModel
+
+logger = structlog.get_logger(__name__)
+
+DEFAULT_LOCK_TTL_SECS: int = 300
+MAX_LOCK_TTL_SECS: int = 3600
+MIN_LOCK_TTL_SECS: int = 5
+
+VALID_RESOURCE_TYPES: frozenset[str] = frozenset({"plan", "project"})
+
+
+class LockService:
+    """Advisory lock manager for plan and project resources.
+
+    All database access goes through the injected *session_factory*.
+
+    Args:
+        session_factory: Callable returning a ``Session`` instance.
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        if session_factory is None:
+            raise ValidationError("session_factory is required")
+        self._session_factory = session_factory
+        self._logger = logger.bind(service="lock_service")
+
+    # ------------------------------------------------------------------
+    # Validation helpers
+    # ------------------------------------------------------------------
+
+    @staticmethod
+    def _validate_resource_type(resource_type: str) -> None:
+        """Validate that *resource_type* is a known value.
+
+        Args:
+            resource_type: Must be ``'plan'`` or ``'project'``.
+
+        Raises:
+            ValidationError: If the resource type is not recognised.
+        """
+        if not resource_type:
+            raise ValidationError("resource_type must not be empty")
+        if resource_type not in VALID_RESOURCE_TYPES:
+            raise ValidationError(
+                f"resource_type must be one of {sorted(VALID_RESOURCE_TYPES)}, "
+                f"got '{resource_type}'"
+            )
+
+    @staticmethod
+    def _validate_non_empty(value: str, name: str) -> None:
+        """Validate that *value* is a non-empty string.
+
+        Args:
+            value: The string to check.
+            name: Human-readable parameter name for the error message.
+
+        Raises:
+            ValidationError: If *value* is empty.
+        """
+        if not value:
+            raise ValidationError(f"{name} must not be empty")
+
+    @staticmethod
+    def _validate_ttl(ttl_seconds: int) -> None:
+        """Validate that *ttl_seconds* falls within bounds.
+
+        Args:
+            ttl_seconds: TTL to validate.
+
+        Raises:
+            ValidationError: If out of range.
+        """
+        if ttl_seconds < MIN_LOCK_TTL_SECS:
+            raise ValidationError(
+                f"ttl_seconds must be >= {MIN_LOCK_TTL_SECS}, got {ttl_seconds}"
+            )
+        if ttl_seconds > MAX_LOCK_TTL_SECS:
+            raise ValidationError(
+                f"ttl_seconds must be <= {MAX_LOCK_TTL_SECS}, got {ttl_seconds}"
+            )
+
+    # ------------------------------------------------------------------
+    # Core operations
+    # ------------------------------------------------------------------
+
+    def acquire(
+        self,
+        owner_id: str,
+        resource_type: str,
+        resource_id: str,
+        ttl_seconds: int = DEFAULT_LOCK_TTL_SECS,
+    ) -> bool:
+        """Acquire an advisory lock on a resource.
+
+        If the same *owner_id* already holds the lock, the TTL is
+        extended (re-entrant acquisition).  If a **different** owner
+        holds a non-expired lock, a ``LockConflictError`` is raised.
+
+        Expired locks held by other owners are transparently replaced.
+
+        Args:
+            owner_id: Unique identifier for the lock owner.
+            resource_type: ``'plan'`` or ``'project'``.
+            resource_id: Identifier of the resource to lock.
+            ttl_seconds: Time-to-live in seconds (default 300).
+
+        Returns:
+            ``True`` if the lock was acquired or renewed.
+
+        Raises:
+            ValidationError: If any parameter is invalid.
+            LockConflictError: If a different owner holds the lock.
+        """
+        self._validate_non_empty(owner_id, "owner_id")
+        self._validate_resource_type(resource_type)
+        self._validate_non_empty(resource_id, "resource_id")
+        self._validate_ttl(ttl_seconds)
+
+        now = datetime.now(tz=UTC)
+        expires = now + timedelta(seconds=ttl_seconds)
+        now_iso = now.isoformat()
+        expires_iso = expires.isoformat()
+
+        session = self._session_factory()
+        try:
+            stmt = select(LockModel).where(
+                LockModel.resource_type == resource_type,
+                LockModel.resource_id == resource_id,
+            )
+            existing: LockModel | None = session.execute(stmt).scalar_one_or_none()
+
+            if existing is not None:
+                existing_owner = str(existing.owner_id)
+                existing_expires = str(existing.expires_at)
+
+                if existing_owner == owner_id:
+                    # Re-entrant: extend TTL
+                    existing.expires_at = expires_iso
+                    session.commit()
+                    self._logger.info(
+                        "Lock renewed (re-entrant)",
+                        owner_id=owner_id,
+                        resource_type=resource_type,
+                        resource_id=resource_id,
+                    )
+                    return True
+
+                # Different owner — check expiry
+                if existing_expires >= now_iso:
+                    raise LockConflictError(
+                        resource_type=resource_type,
+                        resource_id=resource_id,
+                        owner_id=existing_owner,
+                    )
+
+                # Expired: replace
+                existing.owner_id = owner_id
+                existing.acquired_at = now_iso
+                existing.expires_at = expires_iso
+                session.commit()
+                self._logger.info(
+                    "Expired lock replaced",
+                    owner_id=owner_id,
+                    resource_type=resource_type,
+                    resource_id=resource_id,
+                )
+                return True
+
+            # No existing lock — create new
+            lock = LockModel(
+                owner_id=owner_id,
+                resource_type=resource_type,
+                resource_id=resource_id,
+                acquired_at=now_iso,
+                expires_at=expires_iso,
+            )
+            session.add(lock)
+            session.commit()
+            self._logger.info(
+                "Lock acquired",
+                owner_id=owner_id,
+                resource_type=resource_type,
+                resource_id=resource_id,
+            )
+            return True
+        except LockConflictError:
+            session.rollback()
+            raise
+        except Exception:
+            session.rollback()
+            raise
+
+    def release(
+        self,
+        owner_id: str,
+        resource_type: str,
+        resource_id: str,
+    ) -> bool:
+        """Release a lock held by *owner_id*.
+
+        Only the owning identity may release.  Returns ``False`` when
+        no matching lock exists.
+
+        Args:
+            owner_id: The identity that holds the lock.
+            resource_type: ``'plan'`` or ``'project'``.
+            resource_id: Resource identifier.
+
+        Returns:
+            ``True`` if a lock was released, ``False`` otherwise.
+
+        Raises:
+            ValidationError: If parameters are invalid.
+        """
+        self._validate_non_empty(owner_id, "owner_id")
+        self._validate_resource_type(resource_type)
+        self._validate_non_empty(resource_id, "resource_id")
+
+        session = self._session_factory()
+        try:
+            stmt = delete(LockModel).where(
+                LockModel.owner_id == owner_id,
+                LockModel.resource_type == resource_type,
+                LockModel.resource_id == resource_id,
+            )
+            result = session.execute(stmt)
+            session.commit()
+            deleted: bool = int(getattr(result, "rowcount", 0)) > 0
+            if deleted:
+                self._logger.info(
+                    "Lock released",
+                    owner_id=owner_id,
+                    resource_type=resource_type,
+                    resource_id=resource_id,
+                )
+            return deleted
+        except Exception:
+            session.rollback()
+            raise
+
+    def renew(
+        self,
+        owner_id: str,
+        resource_type: str,
+        resource_id: str,
+        ttl_seconds: int = DEFAULT_LOCK_TTL_SECS,
+    ) -> bool:
+        """Renew (extend) a lock's TTL.
+
+        The lock must be held by *owner_id* and must not have expired.
+
+        Args:
+            owner_id: The owning identity.
+            resource_type: ``'plan'`` or ``'project'``.
+            resource_id: Resource identifier.
+            ttl_seconds: New TTL from now.
+
+        Returns:
+            ``True`` if the lock was renewed.
+
+        Raises:
+            ValidationError: If parameters are invalid.
+            LockExpiredError: If the lock has already expired.
+        """
+        self._validate_non_empty(owner_id, "owner_id")
+        self._validate_resource_type(resource_type)
+        self._validate_non_empty(resource_id, "resource_id")
+        self._validate_ttl(ttl_seconds)
+
+        now = datetime.now(tz=UTC)
+        now_iso = now.isoformat()
+        expires_iso = (now + timedelta(seconds=ttl_seconds)).isoformat()
+
+        session = self._session_factory()
+        try:
+            stmt = select(LockModel).where(
+                LockModel.owner_id == owner_id,
+                LockModel.resource_type == resource_type,
+                LockModel.resource_id == resource_id,
+            )
+            lock: LockModel | None = session.execute(stmt).scalar_one_or_none()
+            if lock is None:
+                return False
+
+            if str(lock.expires_at) < now_iso:
+                raise LockExpiredError(
+                    resource_type=resource_type,
+                    resource_id=resource_id,
+                )
+
+            lock.expires_at = expires_iso
+            session.commit()
+            self._logger.info(
+                "Lock renewed",
+                owner_id=owner_id,
+                resource_type=resource_type,
+                resource_id=resource_id,
+            )
+            return True
+        except LockExpiredError:
+            session.rollback()
+            raise
+        except Exception:
+            session.rollback()
+            raise
+
+    def release_all_for_owner(self, owner_id: str) -> int:
+        """Release every lock held by *owner_id* (graceful shutdown).
+
+        Args:
+            owner_id: Identity whose locks should be released.
+
+        Returns:
+            Number of locks released.
+
+        Raises:
+            ValidationError: If *owner_id* is empty.
+        """
+        self._validate_non_empty(owner_id, "owner_id")
+
+        session = self._session_factory()
+        try:
+            stmt = delete(LockModel).where(LockModel.owner_id == owner_id)
+            result = session.execute(stmt)
+            session.commit()
+            count: int = int(getattr(result, "rowcount", 0))
+            if count > 0:
+                self._logger.info(
+                    "Released all locks for owner",
+                    owner_id=owner_id,
+                    count=count,
+                )
+            return count
+        except Exception:
+            session.rollback()
+            raise
+
+    def cleanup_expired(self) -> int:
+        """Delete all expired locks (startup routine).
+
+        Returns:
+            Number of expired locks purged.
+        """
+        now_iso = datetime.now(tz=UTC).isoformat()
+
+        session = self._session_factory()
+        try:
+            stmt = delete(LockModel).where(LockModel.expires_at < now_iso)
+            result = session.execute(stmt)
+            session.commit()
+            count: int = int(getattr(result, "rowcount", 0))
+            if count > 0:
+                self._logger.info("Expired locks cleaned up", count=count)
+            return count
+        except Exception:
+            session.rollback()
+            raise
+
+    def count_stale_locks(self) -> int:
+        """Count locks that have expired but not been cleaned up.
+
+        Intended for the ``agents diagnostics`` health check.
+
+        Returns:
+            Number of stale (expired) locks.
+        """
+        now_iso = datetime.now(tz=UTC).isoformat()
+
+        session = self._session_factory()
+        stmt = (
+            select(func.count())
+            .select_from(LockModel)
+            .where(
+                LockModel.expires_at < now_iso,
+            )
+        )
+        result: int = session.execute(stmt).scalar_one()
+        return result
+
+    def is_locked(self, resource_type: str, resource_id: str) -> bool:
+        """Check whether a resource has an active (non-expired) lock.
+
+        Args:
+            resource_type: ``'plan'`` or ``'project'``.
+            resource_id: Resource identifier.
+
+        Returns:
+            ``True`` if an active lock exists.
+
+        Raises:
+            ValidationError: If parameters are invalid.
+        """
+        self._validate_resource_type(resource_type)
+        self._validate_non_empty(resource_id, "resource_id")
+
+        now_iso = datetime.now(tz=UTC).isoformat()
+
+        session = self._session_factory()
+        stmt = (
+            select(func.count())
+            .select_from(LockModel)
+            .where(
+                LockModel.resource_type == resource_type,
+                LockModel.resource_id == resource_id,
+                LockModel.expires_at >= now_iso,
+            )
+        )
+        result: int = session.execute(stmt).scalar_one()
+        return result > 0

src/cleveragents/cli/commands/system.py

@@ -18,7 +18,13 @@ from enum import StrEnum
 from pathlib import Path
 from typing import Any
 
+from sqlalchemy import create_engine
+from sqlalchemy import inspect as sa_inspect
+from sqlalchemy.orm import Session, sessionmaker
+
 from cleveragents import __version__
+from cleveragents.application.container import get_database_url
+from cleveragents.application.services.lock_service import LockService
 
 # ---------------------------------------------------------------------------
 # Diagnostic check status
@@ -339,6 +345,46 @@ def _check_file_permissions() -> dict[str, Any]:
     }
 
 
+def _check_stale_locks() -> dict[str, Any]:
+    """Check for stale (expired) concurrency locks."""
+    try:
+        db_url = get_database_url()
+        engine = create_engine(db_url, echo=False)
+
+        # Only check if the locks table exists
+        inspector = sa_inspect(engine)
+        if "locks" not in inspector.get_table_names():
+            return {
+                "name": "Stale locks",
+                "status": CheckStatus.OK,
+                "details": "locks table not yet created",
+            }
+
+        factory: sessionmaker[Session] = sessionmaker(
+            bind=engine, expire_on_commit=False
+        )
+        svc = LockService(session_factory=factory)
+        count = svc.count_stale_locks()
+        if count == 0:
+            return {
+                "name": "Stale locks",
+                "status": CheckStatus.OK,
+                "details": "0 stale locks",
+            }
+        return {
+            "name": "Stale locks",
+            "status": CheckStatus.WARN,
+            "details": f"{count} stale lock(s) found",
+            "recommendation": "Run lock cleanup or restart the service",
+        }
+    except Exception:
+        return {
+            "name": "Stale locks",
+            "status": CheckStatus.WARN,
+            "details": "unable to check",
+        }
+
+
 def build_diagnostics_data() -> dict[str, Any]:
     """Run all diagnostic checks and return structured results."""
     start = time.monotonic()
@@ -351,6 +397,7 @@ def build_diagnostics_data() -> dict[str, Any]:
     checks.append(_check_disk_space())
     checks.append(_check_file_permissions())
     checks.append(_check_git())
+    checks.append(_check_stale_locks())
 
     elapsed = time.monotonic() - start
 

src/cleveragents/core/exceptions.py

@@ -82,6 +82,57 @@ class ResourceConflictError(DomainError):
     pass
 
 
+class LockConflictError(BusinessRuleViolation):
+    """Raised when a lock cannot be acquired due to a conflicting owner.
+
+    Attributes:
+        resource_type: The type of resource that is locked.
+        resource_id: The identifier of the locked resource.
+        owner_id: The owner that currently holds the lock.
+    """
+
+    def __init__(
+        self,
+        resource_type: str,
+        resource_id: str,
+        owner_id: str,
+    ) -> None:
+        """Initialize lock conflict error.
+
+        Args:
+            resource_type: The type of resource (e.g. 'plan', 'project').
+            resource_id: The identifier of the locked resource.
+            owner_id: The owner that currently holds the lock.
+        """
+        super().__init__(
+            f"Cannot acquire lock on {resource_type}/{resource_id}: "
+            f"held by owner '{owner_id}'"
+        )
+        self.resource_type = resource_type
+        self.resource_id = resource_id
+        self.owner_id = owner_id
+
+
+class LockExpiredError(DomainError):
+    """Raised when a lock operation targets an expired lock.
+
+    Attributes:
+        resource_type: The type of resource.
+        resource_id: The identifier of the resource.
+    """
+
+    def __init__(self, resource_type: str, resource_id: str) -> None:
+        """Initialize lock expired error.
+
+        Args:
+            resource_type: The type of resource.
+            resource_id: The identifier of the resource.
+        """
+        super().__init__(f"Lock on {resource_type}/{resource_id} has expired")
+        self.resource_type = resource_type
+        self.resource_id = resource_id
+
+
 # Infrastructure Exceptions
 class InfrastructureError(CleverAgentsError):
     """Base for infrastructure errors."""
@@ -221,6 +272,8 @@ __all__ = [
     "ExternalServiceError",
     "FileSystemError",
     "InfrastructureError",
+    "LockConflictError",
+    "LockExpiredError",
     "MigrationNotApprovedError",
     "MissingConfigurationError",
     "ModelNotAvailableError",

src/cleveragents/infrastructure/database/models.py

@@ -48,7 +48,14 @@ from sqlalchemy import (
     UniqueConstraint,
     create_engine,
 )
-from sqlalchemy.orm import declarative_base, relationship, sessionmaker, synonym
+from sqlalchemy.orm import (
+    Mapped,
+    declarative_base,
+    mapped_column,
+    relationship,
+    sessionmaker,
+    synonym,
+)
 
 from cleveragents.domain.models.core import (
     ContextType,
@@ -2395,6 +2402,42 @@ class SkillItemModel(Base):  # type: ignore[misc]
     )
 
 
+# ---------------------------------------------------------------------------
+# Concurrency Lock Models (Stage M4 - migration m4_001_concurrency_locks)
+# ---------------------------------------------------------------------------
+
+
+class LockModel(Base):  # type: ignore[misc]
+    """Database model for concurrency locks.
+
+    Provides plan-level and project-level mutual exclusion via
+    owner-scoped advisory locks with configurable TTL.  Only one
+    active lock per ``(resource_type, resource_id)`` is allowed.
+
+    Table: ``locks``
+    """
+
+    __tablename__ = "locks"
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+    owner_id: Mapped[str] = mapped_column(String(255), nullable=False)
+    resource_type: Mapped[str] = mapped_column(String(50), nullable=False)
+    resource_id: Mapped[str] = mapped_column(String(255), nullable=False)
+    acquired_at: Mapped[str] = mapped_column(String(30), nullable=False)
+    expires_at: Mapped[str] = mapped_column(String(30), nullable=False)
+
+    __table_args__ = (
+        UniqueConstraint(
+            "resource_type",
+            "resource_id",
+            name="uq_locks_resource",
+        ),
+        Index("ix_locks_owner_id", "owner_id"),
+        Index("ix_locks_expires_at", "expires_at"),
+        Index("ix_locks_resource_type", "resource_type"),
+    )
+
+
 # Database initialization functions
 def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any:
     """Initialize the database.

vulture_whitelist.py

@@ -288,3 +288,9 @@ _normalise_executor_output  # noqa: B018, F821
 group_by_resource  # noqa: B018, F821
 run_for_plan  # noqa: B018, F821
 all_required_passed  # noqa: B018, F821
+
+# Concurrency lock service — public API (M4)
+LockModel  # noqa: B018, F821
+LockConflictError  # noqa: B018, F821
+LockExpiredError  # noqa: B018, F821
+_check_stale_locks  # noqa: B018, F821