fix(plan): upsert action arguments during plan use to avoid UNIQUE constraint violation (#4197)

## Summary

`agents plan use` crashed with `sqlite3.IntegrityError: UNIQUE constraint failed: action_arguments.action_name, action_arguments.name` when the action had arguments already registered via `action create`. The root cause was `ActionRepository.update()` using SQLAlchemy's relationship `.clear()` + `.append()` pattern, which deferred the DELETE and processed the INSERT first — triggering a UNIQUE constraint violation when the same `(action_name, name)` pair was being re-inserted.

## Approach

Replace the `.clear()` + `.append()` pattern with explicit bulk `sa_delete()` + `session.flush()` before re-inserting child rows for both `action_arguments` and `action_invariants`. After the flush, expire the relationship collections with `session.expire(row, ["arguments_rel", "invariants_rel"])` so SQLAlchemy reloads from the now-empty database state before appending replacements. This avoids stale identity map references and guarantees the DELETE is committed before any INSERT.

## Key Changes

### Bug fix (`src/cleveragents/infrastructure/database/repositories.py`)
- `ActionRepository.update()` now uses `sa_delete(ActionArgumentModel)` and `sa_delete(ActionInvariantModel)` with `synchronize_session=False`, followed by `session.flush()`, before re-inserting child rows.
- Targeted `session.expire(row, ["arguments_rel", "invariants_rel"])` replaces the removed `.clear()` calls to force collection reload.

### Schema parity (`src/cleveragents/infrastructure/database/models.py`)
- Added `UniqueConstraint("action_name", "position")` to `ActionInvariantModel`.
- Added `UniqueConstraint("action_name", "name")`, `CheckConstraint` for `arg_type`, and `CheckConstraint` for `requirement` to `ActionArgumentModel`.

### Alembic migration (`alembic/versions/a5_006_action_invariants_unique_constraint.py`)
- New migration adds all four constraints to both `action_invariants` and `action_arguments` tables.
- Includes deduplication guards and data normalization so the upgrade succeeds on existing databases with invalid or duplicate rows.
- Uses `batch_alter_table` for SQLite compatibility.

### Tests
- **Behave** (`features/plan_use_action_args_integrity.feature`): 6 scenarios covering the core bug path, zero-argument regression, multiple arguments, reusable action double-use, non-reusable action archival with invariants, and direct repository update.
- **Robot** (`robot/plan_use_action_args_integrity.robot`): Integration test mirroring the Behave scenarios via a helper script.
- **Shared factory** (`features/mocks/test_uow_factory.py`): Extracted `build_test_uow()` from both test suites into a single shared module to eliminate duplication (DRY).

### Minor
- Updated `src/cleveragents/domain/repositories/__init__.py` docstring from table format to bullet list (conflict resolution from rebase).

Closes #4174

Reviewed-on: #4197
Reviewed-by: HAL 9000 <HAL9000@cleverthis.com>
Co-authored-by: Rui Hu <rui.hu@cleverthis.com>
Co-committed-by: Rui Hu <rui.hu@cleverthis.com>

alembic/versions/a5_006_action_invariants_unique_constraint.py

@@ -0,0 +1,132 @@
+"""Align action child table constraints with SQLAlchemy metadata.
+
+Ensures schema parity between metadata-based table creation and Alembic
+migrations by enforcing the same uniqueness and check constraints that
+exist in the SQLAlchemy model definitions for ``action_invariants`` and
+``action_arguments``.
+
+Revision ID: a5_006_action_invariants_unique_constraint
+Revises: m9_002_plan_resume_fields
+Create Date: 2026-04-07 08:30:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a5_006_action_invariants_unique_constraint"
+down_revision: str | Sequence[str] | None = "m9_002_plan_resume_fields"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def _deduplicate_action_invariants() -> None:
+    """Remove duplicate (action_name, position) rows before adding the constraint."""
+
+    op.execute(
+        sa.text(
+            """
+            DELETE FROM action_invariants
+            WHERE id NOT IN (
+                SELECT MIN(id)
+                FROM action_invariants
+                GROUP BY action_name, position
+            )
+            """
+        )
+    )
+
+
+def _deduplicate_action_arguments() -> None:
+    """Normalize action argument rows prior to adding constraints."""
+
+    # Drop duplicate (action_name, name) rows keeping the lowest id so the
+    # unique constraint can be created safely.
+    op.execute(
+        sa.text(
+            """
+            DELETE FROM action_arguments
+            WHERE id NOT IN (
+                SELECT MIN(id)
+                FROM action_arguments
+                GROUP BY action_name, name
+            )
+            """
+        )
+    )
+
+    # Coerce invalid arg_type values to "string" so the CHECK constraint passes.
+    op.execute(
+        sa.text(
+            """
+            UPDATE action_arguments
+            SET arg_type = 'string'
+            WHERE arg_type NOT IN ('string', 'integer', 'float', 'boolean', 'list')
+               OR arg_type IS NULL
+            """
+        )
+    )
+
+    # Coerce invalid requirement values to "optional" by default.
+    op.execute(
+        sa.text(
+            """
+            UPDATE action_arguments
+            SET requirement = 'optional'
+            WHERE requirement NOT IN ('required', 'optional')
+               OR requirement IS NULL
+            """
+        )
+    )
+
+
+def upgrade() -> None:
+    """Add the uq_action_invariants_position unique constraint."""
+
+    _deduplicate_action_invariants()
+    with op.batch_alter_table("action_invariants") as batch:
+        batch.create_unique_constraint(
+            "uq_action_invariants_position",
+            ["action_name", "position"],
+        )
+
+    _deduplicate_action_arguments()
+    with op.batch_alter_table("action_arguments") as batch:
+        batch.create_unique_constraint(
+            "uq_action_arguments_name",
+            ["action_name", "name"],
+        )
+        batch.create_check_constraint(
+            "ck_action_arguments_type",
+            "arg_type IN ('string', 'integer', 'float', 'boolean', 'list')",
+        )
+        batch.create_check_constraint(
+            "ck_action_arguments_requirement",
+            "requirement IN ('required', 'optional')",
+        )
+
+
+def downgrade() -> None:
+    """Drop the uq_action_invariants_position unique constraint."""
+    with op.batch_alter_table("action_invariants") as batch:
+        batch.drop_constraint(
+            "uq_action_invariants_position",
+            type_="unique",
+        )
+
+    with op.batch_alter_table("action_arguments") as batch:
+        batch.drop_constraint(
+            "ck_action_arguments_requirement",
+            type_="check",
+        )
+        batch.drop_constraint(
+            "ck_action_arguments_type",
+            type_="check",
+        )
+        batch.drop_constraint(
+            "uq_action_arguments_name",
+            type_="unique",
+        )

features/mocks/test_uow_factory.py

@@ -0,0 +1,71 @@
+"""Shared in-memory UnitOfWork factory for test isolation.
+
+Provides ``build_test_uow`` — a factory that constructs an in-memory
+SQLite-backed ``UnitOfWork`` suitable for both Behave step definitions
+and Robot Framework integration test helpers.
+
+Centralising the UoW construction eliminates duplication and ensures
+both test suites exercise the same database setup.  Any changes to
+``UnitOfWork.__init__`` must be reflected here in a single place rather
+than in each consuming test file.
+
+Used by:
+- ``features/steps/plan_use_action_args_integrity_steps.py``
+- ``robot/helper_plan_use_action_args_integrity.py``
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
+
+
+def build_test_uow() -> tuple[UnitOfWork, sessionmaker[Session], Engine]:
+    """Build an in-memory UoW for testing.
+
+    Creates a fresh SQLite ``:memory:`` database with all tables, foreign
+    key enforcement enabled, and a ``UnitOfWork`` instance that bypasses
+    the constructor to avoid running migrations inside the test harness.
+
+    Returns:
+        A three-tuple of ``(uow, session_factory, engine)``.
+    """
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        future=True,
+        connect_args={"check_same_thread": False},
+    )
+
+    @event.listens_for(engine, "connect")
+    def _enable_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,
+    )
+
+    uow = UnitOfWork.__new__(UnitOfWork)
+    # Keep these attributes in sync with UnitOfWork.__init__ — we bypass
+    # the constructor to avoid running migrations inside the test harness.
+    uow.database_url = "sqlite:///:memory:"
+    uow._engine = engine
+    uow._session_factory = sf
+    uow._database_initialized = True
+    uow._prompt_for_migration = None
+    uow._require_confirmation = False
+
+    return uow, sf, engine

features/plan_use_action_args_integrity.feature

@@ -0,0 +1,101 @@
+@domain @repository @action_repository
+Feature: Plan use with action arguments does not crash on UNIQUE constraint
+  As a user running 'agents plan use' on an action with arguments
+  I want the plan to be created successfully without a database error
+  So that the full plan lifecycle works after action create + plan use
+
+  Background:
+    Given a persisted plan lifecycle service for action args integrity tests
+
+  # ---------------------------------------------------------------------------
+  # Core bug fix: plan use with arguments that already exist
+  # ---------------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_4174
+  Scenario: plan use preserves arguments registered via action create
+    Given an action "local/args-test" with arguments:
+      | name       | type   | requirement | default |
+      | my_arg     | string | optional    | hello   |
+      | count      | integer| required    |         |
+    When I use the action "local/args-test" to create a plan with arguments:
+      | name  | value |
+      | count | 42    |
+    Then the plan should be created successfully in strategize phase
+    And the plan should have a valid ULID identifier
+
+  # ---------------------------------------------------------------------------
+  # Regression: actions with zero arguments still work
+  # ---------------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_4174
+  Scenario: plan use succeeds when the action has zero arguments
+    Given an action "local/no-args" with no arguments
+    When I use the action "local/no-args" to create a plan without arguments
+    Then the plan should be created successfully in strategize phase
+    And the plan should have a valid ULID identifier
+
+  # ---------------------------------------------------------------------------
+  # Regression: actions with multiple arguments work
+  # ---------------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_4174
+  Scenario: plan use succeeds when the action has multiple arguments
+    Given an action "local/multi-args" with arguments:
+      | name        | type    | requirement | default |
+      | target      | string  | required    |         |
+      | verbose     | boolean | optional    | true    |
+      | threshold   | float   | optional    | 0.8     |
+    When I use the action "local/multi-args" to create a plan with arguments:
+      | name   | value  |
+      | target | src/   |
+    Then the plan should be created successfully in strategize phase
+
+  # ---------------------------------------------------------------------------
+  # Regression: running plan use twice on the same action does not crash
+  # ---------------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_4174
+  Scenario: plan use twice on the same reusable action does not crash
+    Given an action "local/reuse-test" with arguments:
+      | name    | type   | requirement | default |
+      | my_arg  | string | optional    | world   |
+    When I use the action "local/reuse-test" to create a plan without arguments
+    And I use the action "local/reuse-test" to create a second plan without arguments
+    Then both plans should be created successfully
+
+  # ---------------------------------------------------------------------------
+  # Core bug path: archive non-reusable action (triggers repository update)
+  # ---------------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_4174
+  Scenario: plan use archives a non-reusable action with arguments and invariants
+    Given invariants for the next action:
+      | text                              |
+      | Plans must log archival outcomes. |
+    And a non-reusable action "local/non-reuse" with arguments:
+      | name     | type   | requirement | default |
+      | payload  | string | required    |         |
+      | retries  | integer| optional    | 3       |
+    When I use the action "local/non-reuse" to create a plan with arguments:
+      | name    | value |
+      | payload | data  |
+    Then the plan should be created successfully in strategize phase
+    And the action "local/non-reuse" should still have 2 argument(s)
+    And the action "local/non-reuse" should still have 1 invariant(s)
+
+  # ---------------------------------------------------------------------------
+  # ActionRepository.update upserts arguments without IntegrityError
+  # ---------------------------------------------------------------------------
+
+  @tdd_issue @tdd_issue_4174
+  Scenario: Updating an action with existing arguments does not raise IntegrityError
+    Given invariants for the next action:
+      | text                                |
+      | Persisted actions must remain valid. |
+    Given an action "local/update-args" with arguments:
+      | name    | type   | requirement | default |
+      | my_arg  | string | optional    | hello   |
+    When the action "local/update-args" is updated through the repository
+    Then the repository should not raise an integrity error
+    And the action "local/update-args" should still have 1 argument(s)
+    And the action "local/update-args" should still have 1 invariant(s)

features/steps/plan_use_action_args_integrity_steps.py

@@ -0,0 +1,320 @@
+"""Step definitions for plan use action arguments integrity tests.
+
+Covers the fix for issue #4174 — ``plan use`` crashing with
+``sqlite3.IntegrityError`` on the ``action_arguments`` UNIQUE
+constraint when action arguments already exist from ``action create``.
+"""
+
+from __future__ import annotations
+
+import re
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.core.action import (
+    ActionArgument,
+    ArgumentRequirement,
+    ArgumentType,
+)
+from cleveragents.domain.models.core.plan import NamespacedName, Plan, PlanPhase
+from cleveragents.infrastructure.database.models import (
+    ActionArgumentModel,
+    ActionInvariantModel,
+)
+from features.mocks.test_uow_factory import build_test_uow
+
+# Crockford base32 ULID pattern (26 uppercase characters)
+_ULID_RE = re.compile(r"^[0-9A-HJKMNP-TV-Z]{26}$")
+
+
+@given("a persisted plan lifecycle service for action args integrity tests")
+def step_create_service(context: Context) -> None:
+    """Create a persisted PlanLifecycleService backed by in-memory SQLite."""
+    uow, sf, engine = build_test_uow()
+    settings = Settings()
+    service = PlanLifecycleService(settings=settings, unit_of_work=uow)
+    context.pai_service = service
+    context.pai_uow = uow
+    context.pai_sf = sf
+    context.pai_engine = engine
+    context.pai_error = None
+    context.pai_pending_invariants: list[str] = []
+    context.add_cleanup(engine.dispose)
+    plans: list[Plan] = []
+    context.pai_plans = plans
+
+
+def _parse_args_table(context: Context) -> list[ActionArgument]:
+    """Parse a Behave table of argument definitions."""
+    arguments: list[ActionArgument] = []
+    table = context.table
+    if table is None:
+        return arguments
+    for row in table:
+        default_raw = row["default"].strip() if "default" in row.headings else ""
+        default_value: str | int | float | bool | None = None
+        if default_raw:
+            arg_type_str = row["type"].strip().lower()
+            if arg_type_str == "integer":
+                default_value = int(default_raw)
+            elif arg_type_str == "float":
+                default_value = float(default_raw)
+            elif arg_type_str == "boolean":
+                default_value = default_raw.lower() in ("true", "1", "yes")
+            else:
+                default_value = default_raw
+
+        arguments.append(
+            ActionArgument(
+                name=row["name"].strip(),
+                arg_type=ArgumentType(row["type"].strip().lower()),
+                requirement=ArgumentRequirement(row["requirement"].strip().lower()),
+                description=f"Test arg {row['name'].strip()}",
+                default_value=default_value,
+            )
+        )
+    return arguments
+
+
+def _consume_pending_invariants(context: Context) -> list[str]:
+    invariants: list[str] = getattr(context, "pai_pending_invariants", [])
+    context.pai_pending_invariants = []
+    return invariants
+
+
+@given("invariants for the next action:")
+def step_store_invariants(context: Context) -> None:
+    """Store invariants to be applied when the next action is created."""
+    table = context.table
+    if table is None:
+        context.pai_pending_invariants = []
+        return
+    context.pai_pending_invariants = [row["text"].strip() for row in table]
+
+
+@given('an action "{name}" with arguments:')
+def step_create_action_with_args(context: Context, name: str) -> None:
+    """Create a persisted action with the specified arguments."""
+    svc: PlanLifecycleService = context.pai_service
+    arguments = _parse_args_table(context)
+    invariants = _consume_pending_invariants(context)
+    svc.create_action(
+        name=name,
+        description=f"Test action {name}",
+        definition_of_done="All tests pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        arguments=arguments,
+        invariants=invariants,
+    )
+
+
+@given('an action "{name}" with no arguments')
+def step_create_action_no_args(context: Context, name: str) -> None:
+    """Create a persisted action with zero arguments."""
+    svc: PlanLifecycleService = context.pai_service
+    invariants = _consume_pending_invariants(context)
+    svc.create_action(
+        name=name,
+        description=f"Test action {name}",
+        definition_of_done="All tests pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        arguments=[],
+        invariants=invariants,
+    )
+
+
+@given('a non-reusable action "{name}" with arguments:')
+def step_create_non_reusable_action(context: Context, name: str) -> None:
+    """Create a persisted non-reusable action with the specified arguments."""
+    svc: PlanLifecycleService = context.pai_service
+    arguments = _parse_args_table(context)
+    invariants = _consume_pending_invariants(context)
+    svc.create_action(
+        name=name,
+        description=f"Test action {name}",
+        definition_of_done="All tests pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        arguments=arguments,
+        invariants=invariants,
+        reusable=False,
+    )
+
+
+@when('I use the action "{name}" to create a plan with arguments:')
+def step_use_action_with_args(context: Context, name: str) -> None:
+    """Use an action to create a plan, passing argument values."""
+    svc: PlanLifecycleService = context.pai_service
+    arguments: dict[str, str | int | float | bool] = {}
+    table = context.table
+    if table is None:
+        raise ValueError("Step requires a table of argument values")
+    for row in table:
+        val_str = row["value"].strip()
+        # Attempt numeric/bool coercion
+        try:
+            arguments[row["name"].strip()] = int(val_str)
+        except ValueError:
+            try:
+                arguments[row["name"].strip()] = float(val_str)
+            except ValueError:
+                if val_str.lower() in ("true", "false"):
+                    arguments[row["name"].strip()] = val_str.lower() == "true"
+                else:
+                    arguments[row["name"].strip()] = val_str
+
+    try:
+        plan = svc.use_action(action_name=name, arguments=arguments)
+        context.pai_plans.append(plan)
+        context.pai_error = None
+    except Exception as exc:
+        context.pai_error = exc
+
+
+@when('I use the action "{name}" to create a plan without arguments')
+def step_use_action_no_args(context: Context, name: str) -> None:
+    """Use an action to create a plan without explicit arguments."""
+    svc: PlanLifecycleService = context.pai_service
+    try:
+        plan = svc.use_action(action_name=name)
+        context.pai_plans.append(plan)
+        context.pai_error = None
+    except Exception as exc:
+        context.pai_error = exc
+
+
+@when('I use the action "{name}" to create a second plan without arguments')
+def step_use_action_second_plan(context: Context, name: str) -> None:
+    """Use the same action a second time to create another plan."""
+    svc: PlanLifecycleService = context.pai_service
+    try:
+        plan = svc.use_action(action_name=name)
+        context.pai_plans.append(plan)
+        context.pai_error = None
+    except Exception as exc:
+        context.pai_error = exc
+
+
+@when('the action "{name}" is updated through the repository')
+def step_update_action_via_repo(context: Context, name: str) -> None:
+    """Update an action via the repository to trigger the upsert code path."""
+    svc: PlanLifecycleService = context.pai_service
+    try:
+        action = svc.get_action(name)
+        action.description = "Updated description"
+        # Persist the update through the UoW (exercises ActionRepository.update)
+        uow = context.pai_uow
+        with uow.transaction() as ctx:
+            ctx.actions.update(action)
+        context.pai_error = None
+    except Exception as exc:
+        context.pai_error = exc
+
+
+@then("the plan should be created successfully in strategize phase")
+def step_plan_created_in_strategize(context: Context) -> None:
+    """Verify the most recent plan was created in Strategize phase."""
+    assert context.pai_error is None, f"Expected no error, got: {context.pai_error}"
+    assert len(context.pai_plans) >= 1, "No plan was created"
+    plan = context.pai_plans[-1]
+    assert plan.phase == PlanPhase.STRATEGIZE, (
+        f"Expected STRATEGIZE phase, got {plan.phase}"
+    )
+
+
+@then("the plan should have a valid ULID identifier")
+def step_plan_has_valid_ulid(context: Context) -> None:
+    """Verify the plan has a valid 26-char Crockford base32 ULID."""
+    plan = context.pai_plans[-1]
+    plan_id = plan.identity.plan_id
+    assert _ULID_RE.match(plan_id), f"Plan ID '{plan_id}' is not a valid ULID"
+
+
+@then("both plans should be created successfully")
+def step_both_plans_created(context: Context) -> None:
+    """Verify exactly two plans were created without errors."""
+    assert context.pai_error is None, f"Expected no error, got: {context.pai_error}"
+    assert len(context.pai_plans) >= 2, (
+        f"Expected at least 2 plans, got {len(context.pai_plans)}"
+    )
+    # Both should be in Strategize phase
+    for plan in context.pai_plans[-2:]:
+        assert plan.phase == PlanPhase.STRATEGIZE, (
+            f"Expected STRATEGIZE phase, got {plan.phase}"
+        )
+
+
+@then("the repository should not raise an integrity error")
+def step_no_integrity_error(context: Context) -> None:
+    """Verify the update did not raise an IntegrityError."""
+    assert context.pai_error is None, f"Expected no error, got: {context.pai_error}"
+
+
+@then('the action "{name}" should still have {count:d} argument(s)')
+def step_action_has_n_arguments(context: Context, name: str, count: int) -> None:
+    """Verify the action has the expected number of arguments after update."""
+    svc: PlanLifecycleService = context.pai_service
+    session_factory: sessionmaker[Session] = context.pai_sf
+    normalised = str(NamespacedName.parse(name))
+
+    # The PlanLifecycleService caches actions between calls.  We clear the
+    # cache entry to ensure the subsequent read reflects freshly persisted
+    # rows instead of the now-stale in-memory snapshot.
+    svc._actions.pop(normalised, None)
+
+    with session_factory() as session:
+        db_count = int(
+            session.scalar(
+                select(func.count())
+                .select_from(ActionArgumentModel)
+                .where(
+                    ActionArgumentModel.action_name == normalised,
+                )
+            )
+            or 0
+        )
+
+    assert db_count == count, f"Expected {count} argument(s) persisted, got {db_count}"
+
+    action = svc.get_action(name)
+    assert len(action.arguments) == count, (
+        f"Expected {count} argument(s), got {len(action.arguments)}"
+    )
+
+
+@then('the action "{name}" should still have {count:d} invariant(s)')
+def step_action_has_n_invariants(context: Context, name: str, count: int) -> None:
+    """Verify the action has the expected number of invariants after update."""
+    svc: PlanLifecycleService = context.pai_service
+    session_factory: sessionmaker[Session] = context.pai_sf
+    normalised = str(NamespacedName.parse(name))
+
+    svc._actions.pop(normalised, None)
+
+    with session_factory() as session:
+        db_count = int(
+            session.scalar(
+                select(func.count())
+                .select_from(ActionInvariantModel)
+                .where(
+                    ActionInvariantModel.action_name == normalised,
+                )
+            )
+            or 0
+        )
+
+    assert db_count == count, f"Expected {count} invariant(s) persisted, got {db_count}"
+
+    action = svc.get_action(name)
+    assert len(action.invariants) == count, (
+        f"Expected {count} invariant(s), got {len(action.invariants)}"
+    )

robot/actor_registry_persistence.robot

@@ -8,20 +8,19 @@ ${PYTHON}              python
 
 *** Test Cases ***
 Actor Registry Lists Actors With YAML Metadata
-    ${result}=    Run Process    ${PYTHON}    robot/helper_actor_registry_persistence.py    list_yaml_metadata    stdout=PIPE    stderr=PIPE
+    ${result}=    Run Process    ${PYTHON}    robot/helper_actor_registry_persistence.py    list_yaml_metadata
+    ...    stderr=STDOUT    timeout=120s    on_timeout=kill
     Should Be Equal As Integers    ${result.rc}    0
-    ${payload}=    Evaluate    __import__('json').loads('''${result.stdout.strip()}''')
-    Should Be Equal    ${payload['name']}    local/test
-    Should Be Equal    ${payload['schema_version']}    1.0
-    Should Be True     ${payload['has_yaml']}
-    Should Be True     ${payload['has_meta']}
+    Should Contain    ${result.stdout}    yaml-metadata-ok
 
 Actor Registry Schema Version Defaults To 1.0
-    ${result}=    Run Process    ${PYTHON}    robot/helper_actor_registry_persistence.py    schema_version_default    stdout=PIPE    stderr=PIPE
+    ${result}=    Run Process    ${PYTHON}    robot/helper_actor_registry_persistence.py    schema_version_default
+    ...    stderr=STDOUT    timeout=120s    on_timeout=kill
     Should Be Equal As Integers    ${result.rc}    0
-    Should Be Equal    ${result.stdout.strip()}    1.0
+    Should Contain    ${result.stdout}    schema-version-default-ok
 
 Actor Namespaced Name Validation
-    ${result}=    Run Process    ${PYTHON}    robot/helper_actor_registry_persistence.py    namespaced_name    stdout=PIPE    stderr=PIPE
+    ${result}=    Run Process    ${PYTHON}    robot/helper_actor_registry_persistence.py    namespaced_name
+    ...    stderr=STDOUT    timeout=120s    on_timeout=kill
     Should Be Equal As Integers    ${result.rc}    0
-    Should Be Equal    ${result.stdout.strip()}    local/valid-name
+    Should Contain    ${result.stdout}    namespaced-name-ok

robot/helper_actor_registry_persistence.py

@@ -16,7 +16,6 @@ Where *test_name* is one of:
 
 from __future__ import annotations
 
-import json
 import sys
 
 
@@ -32,16 +31,13 @@ def _test_list_yaml_metadata() -> None:
         schema_version="1.0",
         compiled_metadata={"nodes": 2},
     )
-    print(
-        json.dumps(
-            {
-                "name": actor.name,
-                "schema_version": actor.schema_version,
-                "has_yaml": actor.yaml_text is not None,
-                "has_meta": actor.compiled_metadata is not None,
-            }
-        )
+    assert actor.name == "local/test", f"Expected 'local/test', got {actor.name!r}"
+    assert actor.schema_version == "1.0", (
+        f"Expected schema_version '1.0', got {actor.schema_version!r}"
     )
+    assert actor.yaml_text is not None, "yaml_text should not be None"
+    assert actor.compiled_metadata is not None, "compiled_metadata should not be None"
+    print("yaml-metadata-ok")
 
 
 def _test_schema_version_default() -> None:
@@ -53,7 +49,10 @@ def _test_schema_version_default() -> None:
         model="gpt-4",
         config_hash=Actor.compute_hash({}),
     )
-    print(actor.schema_version)
+    assert actor.schema_version == "1.0", (
+        f"Expected schema_version '1.0', got {actor.schema_version!r}"
+    )
+    print("schema-version-default-ok")
 
 
 def _test_namespaced_name() -> None:
@@ -65,7 +64,10 @@ def _test_namespaced_name() -> None:
         model="gpt-4",
         config_hash=Actor.compute_hash({}),
     )
-    print(actor.name)
+    assert actor.name == "local/valid-name", (
+        f"Expected name 'local/valid-name', got {actor.name!r}"
+    )
+    print("namespaced-name-ok")
 
 
 _TESTS = {

robot/helper_plan_use_action_args_integrity.py

@@ -0,0 +1,296 @@
+"""Helper script for plan_use_action_args_integrity.robot integration test.
+
+Validates that ``plan use`` does not crash with an IntegrityError when the
+action has arguments that were already inserted by ``action create``.
+
+Covers issue #4174.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure local source tree is importable
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Ensure the features directory is importable for shared mocks
+_FEATURES = str(Path(__file__).resolve().parents[1] / "features")
+if _FEATURES not in sys.path:
+    sys.path.insert(0, _FEATURES)
+
+from mocks.test_uow_factory import build_test_uow  # noqa: E402
+from sqlalchemy import func, select  # noqa: E402
+
+from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings  # noqa: E402
+from cleveragents.domain.models.core.action import (  # noqa: E402
+    ActionArgument,
+    ArgumentRequirement,
+    ArgumentType,
+)
+from cleveragents.domain.models.core.plan import NamespacedName  # noqa: E402
+from cleveragents.infrastructure.database.models import (  # noqa: E402
+    ActionArgumentModel,
+    ActionInvariantModel,
+)
+
+
+def main() -> None:
+    """Run all plan use action arguments integrity tests."""
+    uow, sf, engine = build_test_uow()
+    settings = Settings()
+    svc = PlanLifecycleService(settings=settings, unit_of_work=uow)
+
+    try:
+        # --- Test 1: plan use with arguments succeeds ---
+        test_args = [
+            ActionArgument(
+                name="my_arg",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="A test argument",
+                default_value="hello",
+            ),
+            ActionArgument(
+                name="count",
+                arg_type=ArgumentType.INTEGER,
+                requirement=ArgumentRequirement.REQUIRED,
+                description="A required integer argument",
+            ),
+        ]
+        svc.create_action(
+            name="local/int-args-test",
+            description="Action with arguments",
+            definition_of_done="Tests pass",
+            strategy_actor="openai/gpt-4",
+            execution_actor="openai/gpt-4",
+            arguments=test_args,
+        )
+        plan1 = svc.use_action(
+            action_name="local/int-args-test",
+            arguments={"count": 42},
+        )
+        assert plan1.phase.value == "strategize", (
+            f"FAIL: expected strategize, got {plan1.phase.value}"
+        )
+        assert len(plan1.identity.plan_id) == 26, "FAIL: plan_id not a ULID"
+        print("PASS: plan use with arguments succeeds")
+
+        # --- Test 2: plan use with zero arguments succeeds ---
+        svc.create_action(
+            name="local/int-no-args",
+            description="Action without arguments",
+            definition_of_done="Tests pass",
+            strategy_actor="openai/gpt-4",
+            execution_actor="openai/gpt-4",
+            arguments=[],
+        )
+        plan2 = svc.use_action(action_name="local/int-no-args")
+        assert plan2.phase.value == "strategize", (
+            f"FAIL: expected strategize, got {plan2.phase.value}"
+        )
+        print("PASS: plan use with zero arguments succeeds")
+
+        # --- Test 3: plan use with multiple arguments succeeds ---
+        multi_args = [
+            ActionArgument(
+                name="target",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.REQUIRED,
+                description="Target path",
+            ),
+            ActionArgument(
+                name="verbose",
+                arg_type=ArgumentType.BOOLEAN,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="Verbose output",
+                default_value=True,
+            ),
+            ActionArgument(
+                name="threshold",
+                arg_type=ArgumentType.FLOAT,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="Quality threshold",
+                default_value=0.8,
+            ),
+        ]
+        svc.create_action(
+            name="local/int-multi-args",
+            description="Action with multiple arguments",
+            definition_of_done="Tests pass",
+            strategy_actor="openai/gpt-4",
+            execution_actor="openai/gpt-4",
+            arguments=multi_args,
+        )
+        plan3 = svc.use_action(
+            action_name="local/int-multi-args",
+            arguments={"target": "src/"},
+        )
+        assert plan3.phase.value == "strategize", (
+            f"FAIL: expected strategize, got {plan3.phase.value}"
+        )
+        print("PASS: plan use with multiple arguments succeeds")
+
+        # --- Test 4: plan use twice on same reusable action succeeds ---
+        single_arg = [
+            ActionArgument(
+                name="my_arg",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="A test argument",
+                default_value="world",
+            ),
+        ]
+        svc.create_action(
+            name="local/int-reuse-test",
+            description="Reusable action",
+            definition_of_done="Tests pass",
+            strategy_actor="openai/gpt-4",
+            execution_actor="openai/gpt-4",
+            arguments=single_arg,
+            reusable=True,
+        )
+        plan4a = svc.use_action(action_name="local/int-reuse-test")
+        plan4b = svc.use_action(action_name="local/int-reuse-test")
+        assert plan4a.phase.value == "strategize", "FAIL: first plan wrong phase"
+        assert plan4b.phase.value == "strategize", "FAIL: second plan wrong phase"
+        assert plan4a.identity.plan_id != plan4b.identity.plan_id, (
+            "FAIL: two plans have the same ID"
+        )
+        print("PASS: plan use twice on same action succeeds")
+
+        # --- Test 5: plan use archives non-reusable action with invariants ---
+        non_reusable_args = [
+            ActionArgument(
+                name="payload",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.REQUIRED,
+                description="Payload to process",
+            ),
+            ActionArgument(
+                name="retries",
+                arg_type=ArgumentType.INTEGER,
+                requirement=ArgumentRequirement.OPTIONAL,
+                description="Retry attempts",
+                default_value=3,
+            ),
+        ]
+        invariants = ["Plans must log archival outcomes."]
+        svc.create_action(
+            name="local/int-non-reuse",
+            description="Non-reusable action",
+            definition_of_done="Tests pass",
+            strategy_actor="openai/gpt-4",
+            execution_actor="openai/gpt-4",
+            arguments=non_reusable_args,
+            invariants=invariants,
+            reusable=False,
+        )
+        plan5 = svc.use_action(
+            action_name="local/int-non-reuse",
+            arguments={"payload": "data"},
+        )
+        assert plan5.phase.value == "strategize", (
+            f"FAIL: expected strategize, got {plan5.phase.value}"
+        )
+
+        normalised_non_reuse = str(NamespacedName.parse("local/int-non-reuse"))
+        # Clear service cache to force a fresh read of persisted rows.
+        svc._actions.pop(normalised_non_reuse, None)
+
+        with sf() as session:
+            arg_count = int(
+                session.scalar(
+                    select(func.count())
+                    .select_from(ActionArgumentModel)
+                    .where(ActionArgumentModel.action_name == normalised_non_reuse)
+                )
+                or 0
+            )
+            inv_count = int(
+                session.scalar(
+                    select(func.count())
+                    .select_from(ActionInvariantModel)
+                    .where(ActionInvariantModel.action_name == normalised_non_reuse)
+                )
+                or 0
+            )
+
+        assert arg_count == 2, f"FAIL: expected 2 persisted arguments, got {arg_count}"
+        assert inv_count == 1, f"FAIL: expected 1 persisted invariant, got {inv_count}"
+        action_non_reuse = svc.get_action("local/int-non-reuse")
+        assert len(action_non_reuse.arguments) == 2, (
+            f"FAIL: expected 2 arguments, got {len(action_non_reuse.arguments)}"
+        )
+        assert len(action_non_reuse.invariants) == 1, (
+            f"FAIL: expected 1 invariant, got {len(action_non_reuse.invariants)}"
+        )
+        print("PASS: plan use archives non-reusable action with invariants")
+
+        # --- Test 6: action update with existing arguments succeeds ---
+        svc.create_action(
+            name="local/int-update-test",
+            description="Action to update",
+            definition_of_done="Tests pass",
+            strategy_actor="openai/gpt-4",
+            execution_actor="openai/gpt-4",
+            arguments=single_arg,
+            invariants=["Persisted actions must remain valid."],
+        )
+
+        action = svc.get_action("local/int-update-test")
+        action.description = "Updated description"
+        with uow.transaction() as ctx:
+            ctx.actions.update(action)
+
+        normalised = str(NamespacedName.parse("local/int-update-test"))
+        svc._actions.pop(normalised, None)
+
+        with sf() as session:
+            db_count = int(
+                session.scalar(
+                    select(func.count())
+                    .select_from(ActionArgumentModel)
+                    .where(
+                        ActionArgumentModel.action_name == normalised,
+                    )
+                )
+                or 0
+            )
+            inv_count = int(
+                session.scalar(
+                    select(func.count())
+                    .select_from(ActionInvariantModel)
+                    .where(
+                        ActionInvariantModel.action_name == normalised,
+                    )
+                )
+                or 0
+            )
+
+        updated = svc.get_action("local/int-update-test")
+        assert updated.description == "Updated description", (
+            "FAIL: description not updated"
+        )
+        assert db_count == 1, f"FAIL: expected 1 persisted argument, got {db_count}"
+        assert len(updated.arguments) == 1, (
+            f"FAIL: expected 1 argument, got {len(updated.arguments)}"
+        )
+        assert inv_count == 1, f"FAIL: expected 1 persisted invariant, got {inv_count}"
+        assert len(updated.invariants) == 1, (
+            f"FAIL: expected 1 invariant, got {len(updated.invariants)}"
+        )
+        print("PASS: action update with existing arguments and invariants succeeds")
+
+        print("PASS: plan_use_action_args_integrity smoke test")
+    finally:
+        engine.dispose()
+
+
+if __name__ == "__main__":
+    main()

robot/plan_use_action_args_integrity.robot

@@ -0,0 +1,19 @@
+*** Settings ***
+Documentation    Integration test for plan use action arguments integrity (#4174)
+...              Verifies that plan use does not crash with IntegrityError
+...              when the action has arguments already inserted by action create.
+Library          Process
+Library          OperatingSystem
+Library          String
+
+*** Variables ***
+${PYTHON}    python
+
+*** Test Cases ***
+Plan Use With Action Arguments Does Not Crash
+    [Documentation]    Create action with arguments, then plan use should succeed
+    ${result}=    Run Process    ${PYTHON}    ${CURDIR}/helper_plan_use_action_args_integrity.py
+    ...    stderr=STDOUT    timeout=120s    on_timeout=kill
+    Log    ${result.stdout}
+    Should Contain    ${result.stdout}    PASS: plan_use_action_args_integrity smoke test
+    Should Be Equal As Integers    ${result.rc}    0

src/cleveragents/domain/repositories/__init__.py

@@ -10,12 +10,14 @@ allows persistence backends to be swapped without touching application code.
 
 ## Protocols
 
-| Protocol | Domain Entity | Infrastructure Adapter |
-|---|---|---|
-| ``LifecyclePlanRepositoryProtocol`` | ``Plan`` | ``LifecyclePlanRepository`` |
-| ``ActionRepositoryProtocol`` | ``Action`` | ``ActionRepository`` |
-| ``DecisionRepositoryProtocol`` | ``Decision`` | ``DecisionRepository`` |
-| ``ProjectRepositoryProtocol`` | ``NamespacedProject`` | ``NamespacedProjectRepo`` |
+- ``LifecyclePlanRepositoryProtocol`` → ``Plan``
+  via ``LifecyclePlanRepository``
+- ``ActionRepositoryProtocol`` → ``Action``
+  via ``ActionRepository``
+- ``DecisionRepositoryProtocol`` → ``Decision``
+  via ``DecisionRepository``
+- ``ProjectRepositoryProtocol`` → ``NamespacedProject``
+  via ``NamespacedProjectRepository``
 
 ## Usage
 

src/cleveragents/infrastructure/database/models.py

@@ -512,6 +512,13 @@ class ActionInvariantModel(Base):  # type: ignore[misc]
 
     __allow_unmapped__ = True
     __tablename__ = "action_invariants"
+    __table_args__ = (
+        UniqueConstraint(
+            "action_name",
+            "position",
+            name="uq_action_invariants_position",
+        ),
+    )
 
     id = Column(Integer, primary_key=True, autoincrement=True)
     action_name = Column(
@@ -538,6 +545,17 @@ class ActionArgumentModel(Base):  # type: ignore[misc]
 
     __allow_unmapped__ = True
     __tablename__ = "action_arguments"
+    __table_args__ = (
+        UniqueConstraint("action_name", "name", name="uq_action_arguments_name"),
+        CheckConstraint(
+            "arg_type IN ('string', 'integer', 'float', 'boolean', 'list')",
+            name="ck_action_arguments_type",
+        ),
+        CheckConstraint(
+            "requirement IN ('required', 'optional')",
+            name="ck_action_arguments_requirement",
+        ),
+    )
 
     id = Column(Integer, primary_key=True, autoincrement=True)
     action_name = Column(

src/cleveragents/infrastructure/database/repositories.py

@@ -69,6 +69,7 @@ if TYPE_CHECKING:
     from cleveragents.domain.models.core.checkpoint import Checkpoint
 
 import structlog
+from sqlalchemy import delete as sa_delete
 from sqlalchemy import func as sa_func
 from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
 from sqlalchemy.exc import IntegrityError, OperationalError
@@ -100,6 +101,8 @@ from cleveragents.domain.repositories.project_repository import (
     ProjectRepositoryProtocol,
 )
 from cleveragents.infrastructure.database.models import (
+    ActionArgumentModel,
+    ActionInvariantModel,
     ActorModel,
     AutomationProfileModel,
     ChangeModel,
@@ -1110,13 +1113,34 @@ class ActionRepository(ActionRepositoryProtocol):
             row.tags_json = _json.dumps(action.tags)  # type: ignore[assignment]
             row.updated_at = datetime.now().isoformat()  # type: ignore[assignment]
 
-            # Update child arguments (replace all)
-            from cleveragents.infrastructure.database.models import (
-                ActionArgumentModel,
-                ActionInvariantModel,
-            )
+            # Update child arguments (replace all).
+            #
+            # We use explicit bulk DELETE + flush before re-inserting
+            # to avoid a UNIQUE constraint violation on
+            # (action_name, name).  SQLAlchemy's relationship .clear()
+            # defers the DELETE and may process the INSERT first,
+            # which triggers an IntegrityError when the same
+            # (action_name, name) pair is being re-inserted.
+
+            session.execute(
+                sa_delete(ActionArgumentModel)
+                .where(ActionArgumentModel.action_name == action_name_str)
+                .execution_options(synchronize_session=False)
+            )
+            session.execute(
+                sa_delete(ActionInvariantModel)
+                .where(ActionInvariantModel.action_name == action_name_str)
+                .execution_options(synchronize_session=False)
+            )
+            session.flush()
+
+            # Expire the relationship collections so SQLAlchemy reloads from
+            # the database (now empty) before we append freshly constructed
+            # replacements.  This avoids reusing stale
+            # ``ActionArgumentModel``/``ActionInvariantModel`` instances that
+            # remain in the identity map after the bulk delete.
+            session.expire(row, ["arguments_rel", "invariants_rel"])
 
-            row.arguments_rel.clear()  # type: ignore[union-attr]
             for idx, arg in enumerate(action.arguments):
                 default_json: str | None = None
                 if arg.default_value is not None:
@@ -1139,8 +1163,7 @@ class ActionRepository(ActionRepositoryProtocol):
                     )
                 )
 
-            # Update child invariants (replace all)
-            row.invariants_rel.clear()  # type: ignore[union-attr]
+            # Re-insert child invariants
             now_iso = datetime.now().isoformat()
             for idx, inv_text in enumerate(getattr(action, "invariants", []) or []):
                 row.invariants_rel.append(  # type: ignore[union-attr]