diff --git a/CHANGELOG.md b/CHANGELOG.md index 9cbc64cf5..fdf6fab13 100644 --- a/CHANGELOG.md +++ b/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) diff --git a/alembic/versions/m4_001_concurrency_locks.py b/alembic/versions/m4_001_concurrency_locks.py new file mode 100644 index 000000000..4aa02b958 --- /dev/null +++ b/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") diff --git a/benchmarks/agent_skills_registry_bench.py b/benchmarks/agent_skills_registry_bench.py index c74f09ebc..a43b4b0da 100644 --- a/benchmarks/agent_skills_registry_bench.py +++ b/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)] diff --git a/benchmarks/concurrency_lock_bench.py b/benchmarks/concurrency_lock_bench.py new file mode 100644 index 000000000..e00446fd5 --- /dev/null +++ b/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") diff --git a/benchmarks/error_recovery_bench.py b/benchmarks/error_recovery_bench.py index 2ad0fcd20..b0ac157e6 100644 --- a/benchmarks/error_recovery_bench.py +++ b/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, ) diff --git a/benchmarks/validation_pipeline_bench.py b/benchmarks/validation_pipeline_bench.py index 0e04f0513..948b47011 100644 --- a/benchmarks/validation_pipeline_bench.py +++ b/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 diff --git a/docs/reference/concurrency.md b/docs/reference/concurrency.md new file mode 100644 index 000000000..524f06963 --- /dev/null +++ b/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) | diff --git a/features/concurrency.feature b/features/concurrency.feature new file mode 100644 index 000000000..da3117b23 --- /dev/null +++ b/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 diff --git a/features/steps/concurrency_steps.py b/features/steps/concurrency_steps.py new file mode 100644 index 000000000..2ff245e2e --- /dev/null +++ b/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)}" + ) diff --git a/robot/concurrency_locks.robot b/robot/concurrency_locks.robot new file mode 100644 index 000000000..60726f61f --- /dev/null +++ b/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 diff --git a/robot/helper_concurrency_locks.py b/robot/helper_concurrency_locks.py new file mode 100644 index 000000000..d0eafd981 --- /dev/null +++ b/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() diff --git a/src/cleveragents/application/services/lock_service.py b/src/cleveragents/application/services/lock_service.py new file mode 100644 index 000000000..f4c76214d --- /dev/null +++ b/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 diff --git a/src/cleveragents/cli/commands/system.py b/src/cleveragents/cli/commands/system.py index df01abebd..6e153364e 100644 --- a/src/cleveragents/cli/commands/system.py +++ b/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 diff --git a/src/cleveragents/core/exceptions.py b/src/cleveragents/core/exceptions.py index 81a0b18b8..131260444 100644 --- a/src/cleveragents/core/exceptions.py +++ b/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", diff --git a/src/cleveragents/infrastructure/database/models.py b/src/cleveragents/infrastructure/database/models.py index 6a4193f08..dbc97450c 100644 --- a/src/cleveragents/infrastructure/database/models.py +++ b/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. diff --git a/vulture_whitelist.py b/vulture_whitelist.py index 8d2ff867d..20972872b 100644 --- a/vulture_whitelist.py +++ b/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