test(persistence): add plan/action persistence suites

benchmarks/persistence_suites_bench.py

@@ -0,0 +1,219 @@
+"""ASV benchmarks for persistence test runtime baseline.
+
+Measures action and plan repository operation latencies to track
+performance regressions in persistence operations.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cleveragents.domain.models.core.action import (
+    Action,
+    ActionArgument,
+    ActionState,
+    ArgumentRequirement,
+    ArgumentType,
+)
+from cleveragents.domain.models.core.plan import (
+    AutomationLevel,
+    NamespacedName,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+from cleveragents.domain.models.core.plan import Plan as V3Plan
+from cleveragents.infrastructure.database.models import Base
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    LifecyclePlanRepository,
+)
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_bench_counter = 0
+
+
+def _bench_ulid() -> str:
+    global _bench_counter
+    _bench_counter += 1
+    n = _bench_counter
+    suffix = ""
+    for _ in range(8):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HGZPS10AQDYTR4PS{suffix}"
+
+
+def _make_bench_action(name: str = "local/ps-bench-action") -> Action:
+    """Create a benchmark action with arguments and invariants."""
+    return Action(
+        namespaced_name=NamespacedName.parse(name),
+        description="Persistence suite benchmark action",
+        long_description=None,
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        estimation_actor=None,
+        review_actor=None,
+        arguments=[
+            ActionArgument(
+                name="target",
+                arg_type=ArgumentType.STRING,
+                requirement=ArgumentRequirement.REQUIRED,
+                description="Target path",
+            ),
+        ],
+        invariants=["No breaking changes"],
+        reusable=True,
+        read_only=False,
+        state=ActionState.AVAILABLE,
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        created_by=None,
+        tags=[],
+    )
+
+
+def _make_bench_plan(plan_id: str | None = None) -> V3Plan:
+    now = datetime.now()
+    return V3Plan(
+        identity=PlanIdentity(plan_id=plan_id or _bench_ulid(), attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="ps-bench-plan"),
+        action_name="local/ps-bench-action",
+        description="Persistence suite benchmark plan",
+        phase=PlanPhase("action"),
+        processing_state=ProcessingState("queued"),
+        automation_level=AutomationLevel("manual"),
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="bench",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+
+
+class TimeActionCreate:
+    """Benchmark action creation latency."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        session = sessionmaker(bind=engine)()
+        self.session = session
+        self.repo = ActionRepository(session_factory=lambda: session)
+        self._counter = 0
+
+    def time_create_action(self) -> None:
+        self._counter += 1
+        action = _make_bench_action(f"local/bench-act-{self._counter}")
+        self.repo.create(action)
+        self.session.commit()
+
+
+class TimeActionGet:
+    """Benchmark action retrieval latency."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        session = sessionmaker(bind=engine)()
+        self.session = session
+        self.repo = ActionRepository(session_factory=lambda: session)
+        action = _make_bench_action()
+        self.repo.create(action)
+        session.commit()
+
+    def time_get_by_name(self) -> None:
+        self.repo.get_by_name("local/ps-bench-action")
+
+
+class TimePlanCreatePersistence:
+    """Benchmark plan creation with FK action."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        session = sessionmaker(bind=engine)()
+        self.session = session
+        factory = lambda: session  # noqa: E731
+        action_repo = ActionRepository(session_factory=factory)
+        action_repo.create(_make_bench_action())
+        session.commit()
+        self.repo = LifecyclePlanRepository(session_factory=factory)
+
+    def time_create_plan(self) -> None:
+        plan = _make_bench_plan()
+        self.repo.create(plan)
+        self.session.commit()
+
+
+class TimePlanUpdatePhase:
+    """Benchmark plan phase update latency."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        session = sessionmaker(bind=engine)()
+        self.session = session
+        factory = lambda: session  # noqa: E731
+        action_repo = ActionRepository(session_factory=factory)
+        action_repo.create(_make_bench_action())
+        session.commit()
+        self.repo = LifecyclePlanRepository(session_factory=factory)
+        self.plan_id = _bench_ulid()
+        plan = _make_bench_plan(self.plan_id)
+        self.repo.create(plan)
+        session.commit()
+
+    def time_update_phase(self) -> None:
+        plan = self.repo.get(self.plan_id)
+        if plan is not None:
+            plan.phase = PlanPhase.STRATEGIZE
+            plan.processing_state = ProcessingState.PROCESSING
+            plan.timestamps.updated_at = datetime.now()
+            self.repo.update(plan)
+            self.session.commit()
+
+
+class TimePlanListFiltered:
+    """Benchmark plan listing with filters."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        session = sessionmaker(bind=engine)()
+        self.session = session
+        factory = lambda: session  # noqa: E731
+        action_repo = ActionRepository(session_factory=factory)
+        action_repo.create(_make_bench_action())
+        session.commit()
+        self.repo = LifecyclePlanRepository(session_factory=factory)
+        for _ in range(30):
+            plan = _make_bench_plan()
+            self.repo.create(plan)
+        session.commit()
+
+    def time_list_all(self) -> None:
+        self.repo.list_plans()
+
+    def time_list_by_phase(self) -> None:
+        self.repo.list_plans(phase="action")
+
+    def time_count(self) -> None:
+        self.repo.count()

docs/development/testing.md

@@ -247,6 +247,82 @@ Plan Execute Transitions Correctly
     Should Contain    ${result.stdout}    execute
 ```
 
+## Persistence Test Suites
+
+The persistence layer (SQLAlchemy repositories for plans and actions) has dedicated test suites at both unit and integration levels.
+
+### Behave: Plan Persistence (`features/plan_persistence.feature`)
+
+19 scenarios covering `LifecyclePlanRepository` operations:
+
+- **Create**: Persist plans via repository, retrieve by ULID, verify all fields round-trip.
+- **Phase/state transitions**: Update `phase` and `processing_state`, verify persistence after each transition.
+- **List filters**: Filter plans by phase, processing state, and action name.
+- **Plan tree links**: Parent/child/root hierarchy with `parent_plan_id` and `root_plan_id`.
+- **Cross-restart**: Close the SQLite session, reopen from disk, verify plans survive reconnection (including project links, arguments, and invariants).
+
+Step definitions: `features/steps/plan_persistence_steps.py`
+
+**Fixtures**: Each scenario gets a fresh temp SQLite database via `_setup_db()` / `_teardown_db()` helper functions. The database is file-backed (not `:memory:`) so cross-restart scenarios can close and reopen the connection.
+
+### Behave: Action Persistence (`features/action_persistence.feature`)
+
+14 scenarios covering `ActionRepository` operations:
+
+- **CRUD**: Create, retrieve by `namespaced_name`, list available, filter by namespace, archive, delete.
+- **Argument ordering**: Persist 3 `ActionArgument` entries and verify positional order is preserved.
+- **Invariant ordering**: Persist 3 invariant strings and verify order is preserved.
+- **Terminal state storage**: Create plans from actions in Apply phase with each terminal `ProcessingState` (`applied`, `constrained`, `errored`, `cancelled`) and verify persistence.
+
+Step definitions: `features/steps/action_persistence_steps.py`
+
+### Robot: Plan Persistence E2E (`robot/plan_persistence_e2e.robot`)
+
+5 end-to-end tests exercising the persistence layer through Python helper scripts:
+
+| Test Case | Description |
+|-----------|-------------|
+| Plan Full Lifecycle Persistence | Create action + plan, transition through all phases to `applied` terminal state |
+| Plan Restart Persistence | Create plan, close DB, reopen, verify all fields survive |
+| Plan Concurrent Session Access | Two independent sessions against the same DB file |
+| Action CRUD Persistence E2E | Create, read, update, delete an action with arguments/invariants |
+| Plan Tree Hierarchy Persistence E2E | Parent/child plan hierarchy with `root_plan_id` verification |
+
+Helper script: `robot/helper_plan_persistence_e2e.py`
+
+### ASV Benchmarks (`benchmarks/persistence_suites_bench.py`)
+
+Performance baselines for persistence test operations:
+
+- `TimePlanPersistenceSuites.time_create_and_retrieve_plan` — Plan round-trip latency
+- `TimePlanPersistenceSuites.time_phase_state_transition` — Phase/state update cycle
+- `TimePlanPersistenceSuites.time_list_plans_filtered` — List with filters
+- `TimePlanPersistenceSuites.time_plan_tree_operations` — Parent/child hierarchy
+- `TimeActionPersistenceSuites.time_create_and_retrieve_action` — Action round-trip
+- `TimeActionPersistenceSuites.time_action_arguments_ordering` — Ordered arguments persistence
+- `TimeActionPersistenceSuites.time_cross_restart_persistence` — Close/reopen cycle
+
+Run with: `nox -s benchmark`
+
+### Running Persistence Tests
+
+```bash
+# Unit tests (Behave) — runs all features including persistence
+nox -s unit_tests
+
+# Run only plan persistence scenarios
+nox -s unit_tests -- features/plan_persistence.feature
+
+# Run only action persistence scenarios
+nox -s unit_tests -- features/action_persistence.feature
+
+# Integration tests (Robot) — runs all robot suites including persistence E2E
+nox -s integration_tests
+
+# Run only plan persistence E2E
+python -m robot robot/plan_persistence_e2e.robot
+```
+
 ## CI Pipeline
 
 The Forgejo CI pipeline runs these jobs in order:

features/action_persistence.feature

@@ -0,0 +1,101 @@
+Feature: Action persistence via ActionRepository
+  As a developer
+  I want actions to persist correctly through the ActionRepository
+  So that action definitions are durable across operations
+
+  Background:
+    Given a fresh in-memory action persistence database
+
+  # Create scenarios
+  Scenario: Create an action and retrieve it by name
+    Given a new action with namespaced name "local/persist-test"
+    When I persist the action via the action repository
+    Then I can retrieve the action by name "local/persist-test"
+    And the retrieved action description should be "Persistence test action"
+
+  Scenario: Create an action preserves all core fields
+    Given a fully-populated action "local/full-persist"
+    When I persist the action via the action repository
+    Then I can retrieve the action by name "local/full-persist"
+    And the action strategy_actor should be "local/strategist"
+    And the action execution_actor should be "local/executor"
+    And the persisted action state should be "available"
+    And the action reusable flag should be true
+
+  # List and archive scenarios
+  Scenario: List actions returns all available actions
+    Given 3 persisted actions in available state
+    When I list available actions from the action repository
+    Then I should get 3 actions in the action list
+
+  Scenario: List actions filters by namespace
+    Given a persisted action "local/ns-one" in available state
+    And a persisted action "myorg/ns-two" in available state
+    When I list actions in namespace "local" from the action repository
+    Then I should get 1 action in the action list
+
+  Scenario: Archive action changes state to archived
+    Given a persisted action "local/to-archive" in available state
+    When I archive the action "local/to-archive" via the action repository
+    Then the action "local/to-archive" should have state "archived"
+
+  Scenario: Delete action removes it from the database
+    Given a persisted action "local/to-delete" in available state
+    When I delete the action "local/to-delete" via the action repository
+    Then the action "local/to-delete" should not exist
+
+  # Arguments ordering scenarios
+  Scenario: Action arguments preserve position ordering
+    Given an action "local/ordered-args" with arguments "alpha" at position 0 and "beta" at position 1 and "gamma" at position 2
+    When I persist the action via the action repository
+    And I retrieve the action "local/ordered-args" from the action repository
+    Then the persisted action should have 3 arguments
+    And the persisted argument at position 0 should be "alpha"
+    And the persisted argument at position 1 should be "beta"
+    And the persisted argument at position 2 should be "gamma"
+
+  # Invariants ordering scenarios
+  Scenario: Action invariants preserve position ordering
+    Given an action "local/ordered-invs" with invariants "No deletion" and "Read only access" and "Log all changes"
+    When I persist the action via the action repository
+    And I retrieve the action "local/ordered-invs" from the action repository
+    Then the persisted action should have 3 invariants
+    And the persisted invariant at position 0 should be "No deletion"
+    And the persisted invariant at position 1 should be "Read only access"
+    And the persisted invariant at position 2 should be "Log all changes"
+
+  # Action phase persistence / terminal state storage
+  Scenario: Action persists in available state by default
+    Given a new action with namespaced name "local/default-state"
+    When I persist the action via the action repository
+    Then the action "local/default-state" should have state "available"
+
+  Scenario: Action persists archived state
+    Given a new action "local/archived-action" in archived state
+    When I persist the action via the action repository
+    Then the action "local/archived-action" should have state "archived"
+
+  # Apply terminal state storage for plans created from this action
+  Scenario: Plan from action records applied terminal state
+    Given a persisted action "local/term-action" in available state
+    And a plan from action "local/term-action" in apply phase with state "applied"
+    When I retrieve the terminal plan from the persistence database
+    Then the terminal plan processing_state should be "applied"
+
+  Scenario: Plan from action records constrained terminal state
+    Given a persisted action "local/constr-action" in available state
+    And a plan from action "local/constr-action" in apply phase with state "constrained"
+    When I retrieve the terminal plan from the persistence database
+    Then the terminal plan processing_state should be "constrained"
+
+  Scenario: Plan from action records errored terminal state
+    Given a persisted action "local/err-action" in available state
+    And a plan from action "local/err-action" in apply phase with state "errored"
+    When I retrieve the terminal plan from the persistence database
+    Then the terminal plan processing_state should be "errored"
+
+  Scenario: Plan from action records cancelled terminal state
+    Given a persisted action "local/cancel-action" in available state
+    And a plan from action "local/cancel-action" in apply phase with state "cancelled"
+    When I retrieve the terminal plan from the persistence database
+    Then the terminal plan processing_state should be "cancelled"

features/plan_persistence.feature

@@ -0,0 +1,121 @@
+Feature: Plan persistence via LifecyclePlanRepository
+  As a developer
+  I want plans to persist correctly through the LifecyclePlanRepository
+  So that plan lifecycle state is durable across operations
+
+  Background:
+    Given a fresh in-memory plan persistence database
+    And a prerequisite action "local/persist-action" exists in the database
+
+  # Create scenarios
+  Scenario: Create a plan and retrieve it by ID
+    Given a new lifecycle plan with ID "01HV000000000000000000PP01"
+    When I persist the plan via the plan repository
+    Then I can retrieve the plan by ID "01HV000000000000000000PP01"
+    And the retrieved plan description should be "Persistence test plan"
+
+  Scenario: Create a plan preserves all identity fields
+    Given a new lifecycle plan with parent tree ID "01HV000000000000000000PP02" parent "01HV000000000000000000PP01" root "01HV000000000000000000PP01"
+    When I persist the plan via the plan repository
+    Then I can retrieve the plan by ID "01HV000000000000000000PP02"
+    And the plan parent_plan_id should be "01HV000000000000000000PP01"
+    And the plan root_plan_id should be "01HV000000000000000000PP01"
+
+  # Phase and state transition scenarios
+  Scenario: Plan persists phase transition from action to strategize
+    Given a persisted plan in phase "action" with state "queued"
+    When I update the plan phase to "strategize" with state "queued"
+    Then the plan persistence phase should be "strategize"
+    And the persisted plan processing_state should be "queued"
+
+  Scenario: Plan persists phase transition from strategize to execute
+    Given a persisted plan in phase "strategize" with state "complete"
+    When I update the plan phase to "execute" with state "queued"
+    Then the plan persistence phase should be "execute"
+    And the persisted plan processing_state should be "queued"
+
+  Scenario: Plan persists phase transition from execute to apply
+    Given a persisted plan in phase "execute" with state "complete"
+    When I update the plan phase to "apply" with state "queued"
+    Then the plan persistence phase should be "apply"
+    And the persisted plan processing_state should be "queued"
+
+  Scenario: Plan persists processing state transition to processing
+    Given a persisted plan in phase "strategize" with state "queued"
+    When I update the plan phase to "strategize" with state "processing"
+    Then the persisted plan processing_state should be "processing"
+
+  Scenario: Plan persists terminal applied state
+    Given a persisted plan in phase "apply" with state "queued"
+    When I update the plan phase to "apply" with state "applied"
+    Then the persisted plan processing_state should be "applied"
+
+  Scenario: Plan persists terminal constrained state
+    Given a persisted plan in phase "apply" with state "queued"
+    When I update the plan phase to "apply" with state "constrained"
+    Then the persisted plan processing_state should be "constrained"
+
+  Scenario: Plan persists terminal errored state
+    Given a persisted plan in phase "apply" with state "queued"
+    When I update the plan phase to "apply" with state "errored"
+    Then the persisted plan processing_state should be "errored"
+
+  Scenario: Plan persists terminal cancelled state
+    Given a persisted plan in phase "strategize" with state "queued"
+    When I update the plan phase to "strategize" with state "cancelled"
+    Then the persisted plan processing_state should be "cancelled"
+
+  # List filter scenarios
+  Scenario: List plans filtered by phase
+    Given 3 persisted plans in phase "strategize"
+    And 2 persisted plans in phase "execute"
+    When I list persisted plans filtered by phase "strategize"
+    Then I should get 3 plans in the list
+
+  Scenario: List plans filtered by processing state
+    Given 2 persisted plans with processing state "queued"
+    And 1 persisted plan with processing state "processing"
+    When I list plans filtered by processing_state "queued"
+    Then I should get 2 plans in the list
+
+  Scenario: List plans filtered by action name
+    Given 2 persisted plans linked to action "local/persist-action"
+    When I list plans filtered by action_name "local/persist-action"
+    Then I should get 2 plans in the list
+
+  # Plan tree link scenarios
+  Scenario: Plan tree preserves parent-child relationships
+    Given a parent plan with ID "01HV0000000000000000TREE01"
+    And a child plan with ID "01HV0000000000000000TREE02" under parent "01HV0000000000000000TREE01"
+    When I retrieve the child plan "01HV0000000000000000TREE02"
+    Then the child plan should reference parent "01HV0000000000000000TREE01"
+    And the child plan should reference root "01HV0000000000000000TREE01"
+
+  Scenario: Plan tree preserves three-level hierarchy
+    Given a root plan with ID "01HV0000000000000000ROOT01"
+    And a mid plan with ID "01HV0000000000000000MID001" under root "01HV0000000000000000ROOT01"
+    And a leaf plan with ID "01HV0000000000000000LEAF01" under parent "01HV0000000000000000MID001" and root "01HV0000000000000000ROOT01"
+    When I retrieve the leaf plan "01HV0000000000000000LEAF01"
+    Then the leaf plan parent should be "01HV0000000000000000MID001"
+    And the leaf plan root should be "01HV0000000000000000ROOT01"
+
+  # Cross-restart scenarios
+  Scenario: Plan status persists across database reconnection
+    Given a persisted plan in phase "execute" with state "processing"
+    When I close and reopen the persistence database
+    Then the plan should still be in phase "execute" with state "processing"
+
+  Scenario: Plan with project links persists across reconnection
+    Given a persisted plan with project links "proj-alpha" and "proj-beta"
+    When I close and reopen the persistence database
+    Then the plan should still have 2 project links
+
+  Scenario: Plan with arguments persists across reconnection
+    Given a persisted plan with arguments "target" and "coverage"
+    When I close and reopen the persistence database
+    Then the plan should still have 2 arguments in order
+
+  Scenario: Plan with invariants persists across reconnection
+    Given a persisted plan with invariant "No breaking changes"
+    When I close and reopen the persistence database
+    Then the plan should still have invariant "No breaking changes"

features/steps/action_persistence_steps.py

@@ -0,0 +1,417 @@
+"""Step definitions for action persistence feature.
+
+Tests ActionRepository CRUD, listing, archiving, argument/invariant
+ordering, and terminal state storage for plans created from actions.
+"""
+
+from __future__ import annotations
+
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cleveragents.domain.models.core.action import (
+    Action,
+    ActionArgument,
+    ActionState,
+    ArgumentRequirement,
+    ArgumentType,
+)
+from cleveragents.domain.models.core.plan import (
+    AutomationLevel,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+from cleveragents.infrastructure.database.models import (
+    Base,
+)
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    LifecyclePlanRepository,
+)
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_ap_counter = 0
+
+
+def _ap_ulid() -> str:
+    """Generate a monotonically-increasing ULID for action persistence tests."""
+    global _ap_counter
+    _ap_counter += 1
+    n = _ap_counter
+    suffix = ""
+    for _ in range(8):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HGZACT10QDYTR4AP{suffix}"
+
+
+def _make_action(
+    name: str,
+    state: ActionState = ActionState.AVAILABLE,
+    arguments: list[ActionArgument] | None = None,
+    invariants: list[str] | None = None,
+) -> Action:
+    """Build an Action domain object for testing."""
+    return Action(
+        namespaced_name=NamespacedName.parse(name),
+        description="Persistence test action",
+        long_description="Detailed description",
+        definition_of_done="All tests pass",
+        strategy_actor="local/strategist",
+        execution_actor="local/executor",
+        review_actor="local/reviewer",
+        arguments=arguments or [],
+        invariants=invariants or [],
+        reusable=True,
+        read_only=False,
+        state=state,
+        created_at=datetime(2026, 1, 1, tzinfo=UTC),
+        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+        created_by="test-user",
+        tags=["test"],
+    )
+
+
+def _ap_setup_db(context: Context) -> None:
+    """Create a temp SQLite DB for action persistence tests."""
+    tmp = tempfile.mktemp(suffix=".db")
+    db_url = f"sqlite:///{tmp}"
+    engine = create_engine(db_url, echo=False)
+    Base.metadata.create_all(engine)
+    sm = sessionmaker(bind=engine)
+    session = sm()
+    context._ap_db_path = tmp
+    context._ap_engine = engine
+    context._ap_session = session
+    context._ap_session_factory = lambda: session
+    context._ap_action_repo = ActionRepository(
+        session_factory=context._ap_session_factory,
+    )
+    context._ap_plan_repo = LifecyclePlanRepository(
+        session_factory=context._ap_session_factory,
+    )
+
+
+def _ap_teardown_db(context: Context) -> None:
+    """Clean up temp DB file."""
+    if hasattr(context, "_ap_session"):
+        context._ap_session.close()
+    if hasattr(context, "_ap_engine"):
+        context._ap_engine.dispose()
+    if hasattr(context, "_ap_db_path"):
+        Path(context._ap_db_path).unlink(missing_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a fresh in-memory action persistence database")
+def step_fresh_action_persistence_db(context: Context) -> None:
+    """Set up a clean SQLite database for action persistence tests."""
+    _ap_setup_db(context)
+    if not hasattr(context, "_cleanup_handlers"):
+        context._cleanup_handlers = []
+    context._cleanup_handlers.append(lambda: _ap_teardown_db(context))
+
+
+# ---------------------------------------------------------------------------
+# Create scenarios
+# ---------------------------------------------------------------------------
+
+
+@given('a new action with namespaced name "{name}"')
+def step_new_action(context: Context, name: str) -> None:
+    """Build a new Action domain object."""
+    context._ap_action = _make_action(name)
+
+
+@given('a fully-populated action "{name}"')
+def step_fully_populated_action(context: Context, name: str) -> None:
+    """Build a fully-populated Action domain object."""
+    args = [
+        ActionArgument(
+            name="coverage",
+            arg_type=ArgumentType.INTEGER,
+            requirement=ArgumentRequirement.REQUIRED,
+            description="Target coverage percentage",
+        ),
+    ]
+    context._ap_action = _make_action(
+        name, arguments=args, invariants=["No regressions"]
+    )
+
+
+@when("I persist the action via the action repository")
+def step_persist_action(context: Context) -> None:
+    """Persist the action through the repository."""
+    context._ap_action_repo.create(context._ap_action)
+    context._ap_session.commit()
+
+
+@then('I can retrieve the action by name "{name}"')
+def step_retrieve_action_by_name(context: Context, name: str) -> None:
+    """Retrieve the action and store for further assertions."""
+    action = context._ap_action_repo.get_by_name(name)
+    assert action is not None, f"Action {name} not found"
+    context._ap_retrieved = action
+
+
+@then('the retrieved action description should be "{desc}"')
+def step_action_description(context: Context, desc: str) -> None:
+    """Verify the action's description."""
+    assert context._ap_retrieved.description == desc
+
+
+@then('the action strategy_actor should be "{actor}"')
+def step_action_strategy_actor(context: Context, actor: str) -> None:
+    """Verify the action's strategy_actor."""
+    assert context._ap_retrieved.strategy_actor == actor
+
+
+@then('the action execution_actor should be "{actor}"')
+def step_action_execution_actor(context: Context, actor: str) -> None:
+    """Verify the action's execution_actor."""
+    assert context._ap_retrieved.execution_actor == actor
+
+
+@then('the persisted action state should be "{state}"')
+def step_action_state_value(context: Context, state: str) -> None:
+    """Verify the persisted action's state."""
+    assert context._ap_retrieved.state == ActionState(state)
+
+
+@then("the action reusable flag should be true")
+def step_action_reusable_true(context: Context) -> None:
+    """Verify the action's reusable flag."""
+    assert context._ap_retrieved.reusable is True
+
+
+# ---------------------------------------------------------------------------
+# List and archive scenarios
+# ---------------------------------------------------------------------------
+
+
+@given("{count:d} persisted actions in available state")
+def step_n_persisted_actions(context: Context, count: int) -> None:
+    """Create N actions in available state."""
+    for i in range(count):
+        action = _make_action(f"local/bulk-action-{i}")
+        context._ap_action_repo.create(action)
+    context._ap_session.commit()
+
+
+@given('a persisted action "{name}" in available state')
+def step_persisted_action_available(context: Context, name: str) -> None:
+    """Create and persist an available action."""
+    action = _make_action(name)
+    context._ap_action_repo.create(action)
+    context._ap_session.commit()
+
+
+@when("I list available actions from the action repository")
+def step_list_available_actions(context: Context) -> None:
+    """List all available actions."""
+    context._ap_listed = context._ap_action_repo.list_available()
+
+
+@when('I list actions in namespace "{namespace}" from the action repository')
+def step_list_actions_namespace(context: Context, namespace: str) -> None:
+    """List actions filtered by namespace."""
+    context._ap_listed = context._ap_action_repo.get_by_namespace(namespace)
+
+
+@then("I should get {count:d} actions in the action list")
+def step_action_list_count(context: Context, count: int) -> None:
+    """Verify the number of actions returned."""
+    assert len(context._ap_listed) == count, (
+        f"Expected {count}, got {len(context._ap_listed)}"
+    )
+
+
+@then("I should get {count:d} action in the action list")
+def step_action_list_count_singular(context: Context, count: int) -> None:
+    """Verify the number of actions returned (singular)."""
+    assert len(context._ap_listed) == count
+
+
+@when('I archive the action "{name}" via the action repository')
+def step_archive_action(context: Context, name: str) -> None:
+    """Archive an action by updating its state."""
+    action = context._ap_action_repo.get_by_name(name)
+    assert action is not None
+    action.state = ActionState.ARCHIVED
+    action.updated_at = datetime.now(tz=UTC)
+    context._ap_action_repo.update(action)
+    context._ap_session.commit()
+
+
+@then('the action "{name}" should have state "{state}"')
+def step_action_has_state(context: Context, name: str, state: str) -> None:
+    """Verify the action's state."""
+    action = context._ap_action_repo.get_by_name(name)
+    assert action is not None
+    assert action.state == ActionState(state)
+
+
+@when('I delete the action "{name}" via the action repository')
+def step_delete_action(context: Context, name: str) -> None:
+    """Delete an action."""
+    result = context._ap_action_repo.delete(name)
+    assert result is True
+    context._ap_session.commit()
+
+
+@then('the action "{name}" should not exist')
+def step_action_not_exist(context: Context, name: str) -> None:
+    """Verify the action no longer exists."""
+    action = context._ap_action_repo.get_by_name(name)
+    assert action is None
+
+
+# ---------------------------------------------------------------------------
+# Arguments ordering scenarios
+# ---------------------------------------------------------------------------
+
+
+@given(
+    'an action "{name}" with arguments "{arg1}" at position 0 and "{arg2}" at position 1 and "{arg3}" at position 2'
+)
+def step_action_with_ordered_args(
+    context: Context, name: str, arg1: str, arg2: str, arg3: str
+) -> None:
+    """Create an action with ordered arguments."""
+    args = [
+        ActionArgument(
+            name=arg1,
+            arg_type=ArgumentType.STRING,
+            requirement=ArgumentRequirement.REQUIRED,
+            description=f"Arg {arg1}",
+        ),
+        ActionArgument(
+            name=arg2,
+            arg_type=ArgumentType.INTEGER,
+            requirement=ArgumentRequirement.OPTIONAL,
+            description=f"Arg {arg2}",
+        ),
+        ActionArgument(
+            name=arg3,
+            arg_type=ArgumentType.BOOLEAN,
+            requirement=ArgumentRequirement.OPTIONAL,
+            description=f"Arg {arg3}",
+        ),
+    ]
+    context._ap_action = _make_action(name, arguments=args)
+
+
+@when('I retrieve the action "{name}" from the action repository')
+def step_retrieve_action(context: Context, name: str) -> None:
+    """Retrieve an action from the repo."""
+    context._ap_retrieved = context._ap_action_repo.get_by_name(name)
+    assert context._ap_retrieved is not None
+
+
+@then("the persisted action should have {count:d} arguments")
+def step_action_argument_count(context: Context, count: int) -> None:
+    """Verify the persisted action argument count."""
+    assert len(context._ap_retrieved.arguments) == count
+
+
+@then('the persisted argument at position {pos:d} should be "{name}"')
+def step_argument_at_position(context: Context, pos: int, name: str) -> None:
+    """Verify the argument at the given position."""
+    assert context._ap_retrieved.arguments[pos].name == name
+
+
+# ---------------------------------------------------------------------------
+# Invariants ordering scenarios
+# ---------------------------------------------------------------------------
+
+
+@given('an action "{name}" with invariants "{inv1}" and "{inv2}" and "{inv3}"')
+def step_action_with_ordered_invariants(
+    context: Context, name: str, inv1: str, inv2: str, inv3: str
+) -> None:
+    """Create an action with ordered invariants."""
+    context._ap_action = _make_action(name, invariants=[inv1, inv2, inv3])
+
+
+@then("the persisted action should have {count:d} invariants")
+def step_action_invariant_count(context: Context, count: int) -> None:
+    """Verify the invariant count."""
+    assert len(context._ap_retrieved.invariants) == count
+
+
+@then('the persisted invariant at position {pos:d} should be "{text}"')
+def step_invariant_at_position(context: Context, pos: int, text: str) -> None:
+    """Verify the invariant text at the given position."""
+    assert context._ap_retrieved.invariants[pos] == text
+
+
+# ---------------------------------------------------------------------------
+# Action state persistence
+# ---------------------------------------------------------------------------
+
+
+@given('a new action "{name}" in archived state')
+def step_new_action_archived(context: Context, name: str) -> None:
+    """Build a new action in archived state."""
+    context._ap_action = _make_action(name, state=ActionState.ARCHIVED)
+
+
+# ---------------------------------------------------------------------------
+# Apply terminal state storage for plans from actions
+# ---------------------------------------------------------------------------
+
+
+@given('a plan from action "{action_name}" in apply phase with state "{state}"')
+def step_plan_from_action_terminal(
+    context: Context, action_name: str, state: str
+) -> None:
+    """Create a plan from the given action in apply phase with terminal state."""
+    plan_id = _ap_ulid()
+    now = datetime(2026, 3, 1, tzinfo=UTC)
+    plan = Plan(
+        identity=PlanIdentity(plan_id=plan_id, attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="terminal-plan"),
+        action_name=action_name,
+        description="Terminal state plan",
+        definition_of_done="Done",
+        phase=PlanPhase.APPLY,
+        processing_state=ProcessingState(state),
+        automation_level=AutomationLevel.MANUAL,
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="test",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+    context._ap_plan_repo.create(plan)
+    context._ap_session.commit()
+    context._ap_terminal_plan_id = plan_id
+
+
+@when("I retrieve the terminal plan from the persistence database")
+def step_retrieve_terminal_plan(context: Context) -> None:
+    """Retrieve the terminal plan."""
+    context._ap_terminal_plan = context._ap_plan_repo.get(context._ap_terminal_plan_id)
+    assert context._ap_terminal_plan is not None
+
+
+@then('the terminal plan processing_state should be "{state}"')
+def step_terminal_plan_state(context: Context, state: str) -> None:
+    """Verify the terminal plan's processing state."""
+    assert context._ap_terminal_plan.processing_state == ProcessingState(state)

features/steps/plan_persistence_steps.py

@@ -0,0 +1,551 @@
+"""Step definitions for plan persistence feature.
+
+Tests LifecyclePlanRepository CRUD, phase/state transitions,
+list filters, plan tree links, and cross-restart durability.
+"""
+
+from __future__ import annotations
+
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cleveragents.domain.models.core.action import Action, ActionState
+from cleveragents.domain.models.core.plan import (
+    AutomationLevel,
+    InvariantSource,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanInvariant,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.infrastructure.database.models import (
+    Base,
+)
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    LifecyclePlanRepository,
+)
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_counter = 0
+
+
+def _next_ulid() -> str:
+    """Generate a monotonically-increasing ULID for test isolation."""
+    global _counter
+    _counter += 1
+    n = _counter
+    suffix = ""
+    for _ in range(8):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HGZ6FE0AQDYTR4BX{suffix}"
+
+
+def _make_plan(
+    plan_id: str,
+    action_name: str = "local/persist-action",
+    phase: PlanPhase = PlanPhase.ACTION,
+    processing_state: ProcessingState = ProcessingState.QUEUED,
+    parent_plan_id: str | None = None,
+    root_plan_id: str | None = None,
+    project_links: list[ProjectLink] | None = None,
+    arguments: dict[str, Any] | None = None,
+    arguments_order: list[str] | None = None,
+    invariants: list[PlanInvariant] | None = None,
+) -> Plan:
+    """Build a Plan domain object for testing."""
+    now = datetime(2026, 3, 1, tzinfo=UTC)
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=plan_id,
+            parent_plan_id=parent_plan_id,
+            root_plan_id=root_plan_id,
+            attempt=1,
+        ),
+        namespaced_name=NamespacedName(namespace="local", name="persist-plan"),
+        action_name=action_name,
+        description="Persistence test plan",
+        definition_of_done="All assertions pass",
+        phase=phase,
+        processing_state=processing_state,
+        automation_level=AutomationLevel.MANUAL,
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        project_links=project_links or [],
+        invariants=invariants or [],
+        arguments=arguments or {},
+        arguments_order=arguments_order or [],
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="test-persist",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+
+
+def _setup_db(context: Context) -> None:
+    """Create a temp SQLite DB and attach repos to context."""
+    tmp = tempfile.mktemp(suffix=".db")
+    db_url = f"sqlite:///{tmp}"
+    engine = create_engine(db_url, echo=False)
+    Base.metadata.create_all(engine)
+    sm = sessionmaker(bind=engine)
+    session = sm()
+    context._pp_db_path = tmp
+    context._pp_db_url = db_url
+    context._pp_engine = engine
+    context._pp_session = session
+    context._pp_session_factory = lambda: session
+    context._pp_plan_repo = LifecyclePlanRepository(
+        session_factory=context._pp_session_factory,
+    )
+    context._pp_action_repo = ActionRepository(
+        session_factory=context._pp_session_factory,
+    )
+
+
+def _teardown_db(context: Context) -> None:
+    """Close the session and clean up temp DB file."""
+    if hasattr(context, "_pp_session"):
+        context._pp_session.close()
+    if hasattr(context, "_pp_engine"):
+        context._pp_engine.dispose()
+    if hasattr(context, "_pp_db_path"):
+        Path(context._pp_db_path).unlink(missing_ok=True)
+
+
+def _create_action(context: Context, action_name: str = "local/persist-action") -> None:
+    """Ensure an action exists for FK constraint."""
+    ns = NamespacedName.parse(action_name)
+    action = Action(
+        namespaced_name=ns,
+        description="Prerequisite action for plan persistence tests",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        state=ActionState.AVAILABLE,
+        created_at=datetime(2026, 1, 1, tzinfo=UTC),
+        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+    )
+    context._pp_action_repo.create(action)
+    context._pp_session.commit()
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a fresh in-memory plan persistence database")
+def step_fresh_plan_persistence_db(context: Context) -> None:
+    """Set up a clean SQLite database for plan persistence tests."""
+    _setup_db(context)
+    if not hasattr(context, "_cleanup_handlers"):
+        context._cleanup_handlers = []
+    context._cleanup_handlers.append(lambda: _teardown_db(context))
+
+
+@given('a prerequisite action "{action_name}" exists in the database')
+def step_prerequisite_action_exists(context: Context, action_name: str) -> None:
+    """Create a prerequisite action for FK constraints."""
+    _create_action(context, action_name)
+
+
+# ---------------------------------------------------------------------------
+# Create scenarios
+# ---------------------------------------------------------------------------
+
+
+@given('a new lifecycle plan with ID "{plan_id}"')
+def step_new_lifecycle_plan(context: Context, plan_id: str) -> None:
+    """Build a new Plan domain object."""
+    context._pp_plan = _make_plan(plan_id)
+
+
+@given(
+    'a new lifecycle plan with parent tree ID "{plan_id}" parent "{parent_id}" root "{root_id}"'
+)
+def step_new_lifecycle_plan_with_tree(
+    context: Context, plan_id: str, parent_id: str, root_id: str
+) -> None:
+    """Build a new Plan with parent/root IDs."""
+    # Ensure parent plan exists first
+    parent = _make_plan(parent_id)
+    context._pp_plan_repo.create(parent)
+    context._pp_session.commit()
+    context._pp_plan = _make_plan(
+        plan_id, parent_plan_id=parent_id, root_plan_id=root_id
+    )
+
+
+@when("I persist the plan via the plan repository")
+def step_persist_plan(context: Context) -> None:
+    """Persist the plan through the repository."""
+    context._pp_plan_repo.create(context._pp_plan)
+    context._pp_session.commit()
+
+
+@then('I can retrieve the plan by ID "{plan_id}"')
+def step_retrieve_plan_by_id(context: Context, plan_id: str) -> None:
+    """Retrieve the plan and store for further assertions."""
+    plan = context._pp_plan_repo.get(plan_id)
+    assert plan is not None, f"Plan {plan_id} not found"
+    context._pp_retrieved = plan
+
+
+@then('the retrieved plan description should be "{description}"')
+def step_plan_description(context: Context, description: str) -> None:
+    """Verify the plan's description."""
+    assert context._pp_retrieved.description == description
+
+
+@then('the plan parent_plan_id should be "{parent_id}"')
+def step_plan_parent_id(context: Context, parent_id: str) -> None:
+    """Verify the plan's parent_plan_id."""
+    assert context._pp_retrieved.identity.parent_plan_id == parent_id
+
+
+@then('the plan root_plan_id should be "{root_id}"')
+def step_plan_root_id(context: Context, root_id: str) -> None:
+    """Verify the plan's root_plan_id."""
+    assert context._pp_retrieved.identity.root_plan_id == root_id
+
+
+# ---------------------------------------------------------------------------
+# Phase/state transition scenarios
+# ---------------------------------------------------------------------------
+
+
+@given('a persisted plan in phase "{phase}" with state "{state}"')
+def step_persisted_plan_phase_state(context: Context, phase: str, state: str) -> None:
+    """Create and persist a plan in the given phase/state."""
+    plan_id = _next_ulid()
+    plan = _make_plan(
+        plan_id,
+        phase=PlanPhase(phase),
+        processing_state=ProcessingState(state),
+    )
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+    context._pp_current_plan_id = plan_id
+
+
+@when('I update the plan phase to "{phase}" with state "{state}"')
+def step_update_plan_phase_state(context: Context, phase: str, state: str) -> None:
+    """Update the plan's phase and processing state."""
+    plan = context._pp_plan_repo.get(context._pp_current_plan_id)
+    assert plan is not None
+    plan.phase = PlanPhase(phase)
+    plan.processing_state = ProcessingState(state)
+    plan.timestamps.updated_at = datetime.now(tz=UTC)
+    context._pp_plan_repo.update(plan)
+    context._pp_session.commit()
+
+
+@then('the plan persistence phase should be "{phase}"')
+def step_verify_plan_phase(context: Context, phase: str) -> None:
+    """Verify the persisted plan's phase."""
+    plan = context._pp_plan_repo.get(context._pp_current_plan_id)
+    assert plan is not None
+    assert plan.phase == PlanPhase(phase), f"Expected {phase}, got {plan.phase}"
+
+
+@then('the persisted plan processing_state should be "{state}"')
+def step_verify_plan_state(context: Context, state: str) -> None:
+    """Verify the persisted plan's processing state."""
+    plan = context._pp_plan_repo.get(context._pp_current_plan_id)
+    assert plan is not None
+    assert plan.processing_state == ProcessingState(state), (
+        f"Expected {state}, got {plan.processing_state}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# List filter scenarios
+# ---------------------------------------------------------------------------
+
+
+@given('{count:d} persisted plans in phase "{phase}"')
+def step_n_plans_in_phase(context: Context, count: int, phase: str) -> None:
+    """Create N plans in the given phase."""
+    for _ in range(count):
+        plan_id = _next_ulid()
+        plan = _make_plan(plan_id, phase=PlanPhase(phase))
+        context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@given('{count:d} persisted plans with processing state "{state}"')
+def step_n_plans_in_state(context: Context, count: int, state: str) -> None:
+    """Create N plans with the given processing state."""
+    for _ in range(count):
+        plan_id = _next_ulid()
+        plan = _make_plan(plan_id, processing_state=ProcessingState(state))
+        context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@given('{count:d} persisted plan with processing state "{state}"')
+def step_one_plan_in_state(context: Context, count: int, state: str) -> None:
+    """Create 1 plan with the given processing state (singular)."""
+    for _ in range(count):
+        plan_id = _next_ulid()
+        plan = _make_plan(plan_id, processing_state=ProcessingState(state))
+        context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@given('{count:d} persisted plans linked to action "{action_name}"')
+def step_n_plans_linked_to_action(
+    context: Context, count: int, action_name: str
+) -> None:
+    """Create N plans linked to the given action."""
+    for _ in range(count):
+        plan_id = _next_ulid()
+        plan = _make_plan(plan_id, action_name=action_name)
+        context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@when('I list persisted plans filtered by phase "{phase}"')
+def step_list_plans_by_phase(context: Context, phase: str) -> None:
+    """List plans filtered by phase."""
+    context._pp_listed = context._pp_plan_repo.list_plans(phase=phase)
+
+
+@when('I list plans filtered by processing_state "{state}"')
+def step_list_plans_by_state(context: Context, state: str) -> None:
+    """List plans filtered by processing state."""
+    context._pp_listed = context._pp_plan_repo.list_plans(processing_state=state)
+
+
+@when('I list plans filtered by action_name "{action_name}"')
+def step_list_plans_by_action(context: Context, action_name: str) -> None:
+    """List plans filtered by action name."""
+    context._pp_listed = context._pp_plan_repo.list_plans(action_name=action_name)
+
+
+@then("I should get {count:d} plans in the list")
+def step_verify_plan_count(context: Context, count: int) -> None:
+    """Verify the number of plans returned."""
+    assert len(context._pp_listed) == count, (
+        f"Expected {count}, got {len(context._pp_listed)}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Plan tree link scenarios
+# ---------------------------------------------------------------------------
+
+
+@given('a parent plan with ID "{plan_id}"')
+def step_parent_plan(context: Context, plan_id: str) -> None:
+    """Create and persist a parent plan."""
+    plan = _make_plan(plan_id, phase=PlanPhase.EXECUTE)
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@given('a child plan with ID "{child_id}" under parent "{parent_id}"')
+def step_child_plan(context: Context, child_id: str, parent_id: str) -> None:
+    """Create and persist a child plan referencing the parent."""
+    plan = _make_plan(
+        child_id,
+        parent_plan_id=parent_id,
+        root_plan_id=parent_id,
+        phase=PlanPhase.STRATEGIZE,
+    )
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@when('I retrieve the child plan "{plan_id}"')
+def step_retrieve_child(context: Context, plan_id: str) -> None:
+    """Retrieve a child plan from the repo."""
+    context._pp_child = context._pp_plan_repo.get(plan_id)
+    assert context._pp_child is not None
+
+
+@then('the child plan should reference parent "{parent_id}"')
+def step_child_references_parent(context: Context, parent_id: str) -> None:
+    """Verify the child plan's parent reference."""
+    assert context._pp_child.identity.parent_plan_id == parent_id
+
+
+@then('the child plan should reference root "{root_id}"')
+def step_child_references_root(context: Context, root_id: str) -> None:
+    """Verify the child plan's root reference."""
+    assert context._pp_child.identity.root_plan_id == root_id
+
+
+@given('a root plan with ID "{plan_id}"')
+def step_root_plan(context: Context, plan_id: str) -> None:
+    """Create and persist a root plan."""
+    plan = _make_plan(plan_id, phase=PlanPhase.EXECUTE)
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@given('a mid plan with ID "{mid_id}" under root "{root_id}"')
+def step_mid_plan(context: Context, mid_id: str, root_id: str) -> None:
+    """Create and persist a mid-level plan."""
+    plan = _make_plan(
+        mid_id,
+        parent_plan_id=root_id,
+        root_plan_id=root_id,
+        phase=PlanPhase.STRATEGIZE,
+    )
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@given(
+    'a leaf plan with ID "{leaf_id}" under parent "{parent_id}" and root "{root_id}"'
+)
+def step_leaf_plan(
+    context: Context, leaf_id: str, parent_id: str, root_id: str
+) -> None:
+    """Create and persist a leaf plan."""
+    plan = _make_plan(
+        leaf_id,
+        parent_plan_id=parent_id,
+        root_plan_id=root_id,
+        phase=PlanPhase.STRATEGIZE,
+    )
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+
+
+@when('I retrieve the leaf plan "{plan_id}"')
+def step_retrieve_leaf(context: Context, plan_id: str) -> None:
+    """Retrieve a leaf plan from the repo."""
+    context._pp_leaf = context._pp_plan_repo.get(plan_id)
+    assert context._pp_leaf is not None
+
+
+@then('the leaf plan parent should be "{parent_id}"')
+def step_leaf_parent(context: Context, parent_id: str) -> None:
+    """Verify the leaf plan's parent reference."""
+    assert context._pp_leaf.identity.parent_plan_id == parent_id
+
+
+@then('the leaf plan root should be "{root_id}"')
+def step_leaf_root(context: Context, root_id: str) -> None:
+    """Verify the leaf plan's root reference."""
+    assert context._pp_leaf.identity.root_plan_id == root_id
+
+
+# ---------------------------------------------------------------------------
+# Cross-restart scenarios
+# ---------------------------------------------------------------------------
+
+
+@when("I close and reopen the persistence database")
+def step_reopen_database(context: Context) -> None:
+    """Close the session and engine, then reopen from the same file."""
+    db_path = context._pp_db_path
+    context._pp_session.close()
+    context._pp_engine.dispose()
+
+    engine = create_engine(f"sqlite:///{db_path}", echo=False)
+    sm = sessionmaker(bind=engine)
+    session = sm()
+    context._pp_engine = engine
+    context._pp_session = session
+    context._pp_session_factory = lambda: session
+    context._pp_plan_repo = LifecyclePlanRepository(
+        session_factory=context._pp_session_factory,
+    )
+    context._pp_action_repo = ActionRepository(
+        session_factory=context._pp_session_factory,
+    )
+
+
+@then('the plan should still be in phase "{phase}" with state "{state}"')
+def step_plan_phase_after_restart(context: Context, phase: str, state: str) -> None:
+    """Verify plan phase/state after reconnection."""
+    plan = context._pp_plan_repo.get(context._pp_current_plan_id)
+    assert plan is not None, "Plan not found after reconnect"
+    assert plan.phase == PlanPhase(phase)
+    assert plan.processing_state == ProcessingState(state)
+
+
+@given('a persisted plan with project links "{link1}" and "{link2}"')
+def step_plan_with_project_links(context: Context, link1: str, link2: str) -> None:
+    """Create a plan with project links."""
+    plan_id = _next_ulid()
+    plan = _make_plan(
+        plan_id,
+        project_links=[
+            ProjectLink(project_name=link1),
+            ProjectLink(project_name=link2),
+        ],
+    )
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+    context._pp_current_plan_id = plan_id
+
+
+@then("the plan should still have {count:d} project links")
+def step_plan_project_links_count(context: Context, count: int) -> None:
+    """Verify the plan still has the expected number of project links."""
+    plan = context._pp_plan_repo.get(context._pp_current_plan_id)
+    assert plan is not None
+    assert len(plan.project_links) == count
+
+
+@given('a persisted plan with arguments "{arg1}" and "{arg2}"')
+def step_plan_with_arguments(context: Context, arg1: str, arg2: str) -> None:
+    """Create a plan with arguments."""
+    plan_id = _next_ulid()
+    plan = _make_plan(
+        plan_id,
+        arguments={arg1: "value1", arg2: "value2"},
+        arguments_order=[arg1, arg2],
+    )
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+    context._pp_current_plan_id = plan_id
+
+
+@then("the plan should still have {count:d} arguments in order")
+def step_plan_arguments_count(context: Context, count: int) -> None:
+    """Verify the plan still has the expected number of arguments."""
+    plan = context._pp_plan_repo.get(context._pp_current_plan_id)
+    assert plan is not None
+    assert len(plan.arguments) == count
+    assert len(plan.arguments_order) == count
+
+
+@given('a persisted plan with invariant "{text}"')
+def step_plan_with_invariant(context: Context, text: str) -> None:
+    """Create a plan with an invariant."""
+    plan_id = _next_ulid()
+    plan = _make_plan(
+        plan_id,
+        invariants=[PlanInvariant(text=text, source=InvariantSource.PLAN)],
+    )
+    context._pp_plan_repo.create(plan)
+    context._pp_session.commit()
+    context._pp_current_plan_id = plan_id
+
+
+@then('the plan should still have invariant "{text}"')
+def step_plan_has_invariant(context: Context, text: str) -> None:
+    """Verify the plan still has the invariant."""
+    plan = context._pp_plan_repo.get(context._pp_current_plan_id)
+    assert plan is not None
+    invariant_texts = [inv.text for inv in plan.invariants]
+    assert text in invariant_texts, f"Expected '{text}' in {invariant_texts}"

implementation_plan.md

@@ -1821,18 +1821,18 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
   - [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. Done: Day 7, February 15, 2026.
 
 - [ ] **COMMIT (Owner: Brent | Group: A5.tests | Branch: feature/m1-persistence-tests | Planned: Day 11 | Expected: Day 13) - Commit message: "test(persistence): add plan/action persistence suites"**
-  - [ ] Git [Brent]: `git checkout master`
-  - [ ] Git [Brent]: `git pull origin master`
-  - [ ] Git [Brent]: `git checkout -b feature/m1-persistence-tests`
-  - [ ] Git [Brent]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
-  - [ ] Tests (Behave) [Brent]: Add plan persistence scenarios (create, phase/state transitions, list filters, plan tree links).
-  - [ ] Tests (Behave) [Brent]: Add scenarios for Action phase persistence and Apply terminal state storage (`applied`, `constrained`, `errored`, `cancelled`).
-  - [ ] Tests (Behave) [Brent]: Add action persistence scenarios (create/list/archive, arguments/invariants ordering).
-  - [ ] Tests (Behave) [Brent]: Add cross-restart scenarios verifying plan status persists across process restarts.
-  - [ ] Tests (Robot) [Brent]: Add plan persistence E2E (full lifecycle, restart persistence, concurrent CLI access).
-  - [ ] Docs [Brent]: Update `docs/development/testing.md` with persistence suites, fixtures, and nox commands.
-  - [ ] Tests (ASV) [Brent]: Add `benchmarks/persistence_suites_bench.py` for test runtime baseline.
-  - [ ] Quality [Brent]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
+  - [X] Git [Brent]: `git checkout master`
+  - [X] Git [Brent]: `git pull origin master`
+  - [X] Git [Brent]: `git checkout -b feature/m1-persistence-tests`
+  - [X] Git [Brent]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
+  - [X] Tests (Behave) [Brent]: Add plan persistence scenarios (create, phase/state transitions, list filters, plan tree links).
+  - [X] Tests (Behave) [Brent]: Add scenarios for Action phase persistence and Apply terminal state storage (`applied`, `constrained`, `errored`, `cancelled`).
+  - [X] Tests (Behave) [Brent]: Add action persistence scenarios (create/list/archive, arguments/invariants ordering).
+  - [X] Tests (Behave) [Brent]: Add cross-restart scenarios verifying plan status persists across process restarts.
+  - [X] Tests (Robot) [Brent]: Add plan persistence E2E (full lifecycle, restart persistence, concurrent CLI access).
+  - [X] Docs [Brent]: Update `docs/development/testing.md` with persistence suites, fixtures, and nox commands.
+  - [X] Tests (ASV) [Brent]: Add `benchmarks/persistence_suites_bench.py` for test runtime baseline.
+  - [X] Quality [Brent]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
   - [ ] Git [Brent]: `git add .`
   - [ ] Git [Brent]: `git commit -m "test(persistence): add plan/action persistence suites"`
   - [ ] Forgejo PR [Brent]: Open PR from `feature/m1-persistence-tests` to `master` with description "Add Behave/Robot persistence coverage for action + plan repositories.".

robot/helper_plan_persistence_e2e.py

@@ -0,0 +1,392 @@
+"""Helper utilities for plan persistence E2E Robot integration tests.
+
+Covers:
+- Full lifecycle persistence (all phase transitions)
+- Restart persistence (close + reopen DB)
+- Concurrent session access
+- Action CRUD
+- Plan tree hierarchy
+"""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session
+
+from cleveragents.domain.models.core.action import (
+    Action,
+    ActionArgument,
+    ActionState,
+    ArgumentRequirement,
+    ArgumentType,
+)
+from cleveragents.domain.models.core.plan import (
+    AutomationLevel,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.infrastructure.database.models import (
+    init_database,
+)
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    LifecyclePlanRepository,
+)
+
+
+def _create_temp_db() -> tuple[str, Session]:
+    """Create a temporary SQLite database and return (path, session)."""
+    tmp = tempfile.mktemp(suffix=".db")
+    db_url = f"sqlite:///{tmp}"
+    engine = init_database(db_url)
+    session = Session(bind=engine)
+    return tmp, session
+
+
+def _make_action(name: str = "local/e2e-action") -> Action:
+    """Create a minimal action for FK satisfaction."""
+    return Action(
+        namespaced_name=NamespacedName.parse(name),
+        description="E2E test action",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        state=ActionState.AVAILABLE,
+        created_at=datetime(2026, 1, 1, tzinfo=UTC),
+        updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+    )
+
+
+def _make_plan(
+    plan_id: str,
+    action_name: str = "local/e2e-action",
+    phase: PlanPhase = PlanPhase.ACTION,
+    state: ProcessingState = ProcessingState.QUEUED,
+    parent_plan_id: str | None = None,
+    root_plan_id: str | None = None,
+) -> Plan:
+    """Create a Plan domain object."""
+    now = datetime(2026, 3, 1, tzinfo=UTC)
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=plan_id,
+            parent_plan_id=parent_plan_id,
+            root_plan_id=root_plan_id,
+            attempt=1,
+        ),
+        namespaced_name=NamespacedName(namespace="local", name="e2e-plan"),
+        action_name=action_name,
+        description="E2E test plan",
+        definition_of_done="E2E assertions pass",
+        phase=phase,
+        processing_state=state,
+        automation_level=AutomationLevel.MANUAL,
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        project_links=[ProjectLink(project_name="local/proj-e2e")],
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="e2e-test",
+        tags=["e2e"],
+        reusable=True,
+        read_only=False,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Full lifecycle
+# ---------------------------------------------------------------------------
+
+
+def _full_lifecycle() -> None:
+    """E2E: Create plan, transition through all phases to applied."""
+    tmp, session = _create_temp_db()
+    factory = lambda: session  # noqa: E731
+    try:
+        # Setup action
+        action_repo = ActionRepository(session_factory=factory)
+        action_repo.create(_make_action())
+        session.commit()
+
+        plan_repo = LifecyclePlanRepository(session_factory=factory)
+
+        # Phase 1: Create plan in action phase
+        plan = _make_plan("01HV00000000000000E2E0FC01")
+        plan_repo.create(plan)
+        session.commit()
+
+        retrieved = plan_repo.get("01HV00000000000000E2E0FC01")
+        assert retrieved is not None
+        assert retrieved.phase == PlanPhase.ACTION
+
+        # Phase 2: Transition to strategize
+        retrieved.phase = PlanPhase.STRATEGIZE
+        retrieved.processing_state = ProcessingState.PROCESSING
+        retrieved.timestamps.updated_at = datetime.now(tz=UTC)
+        plan_repo.update(retrieved)
+        session.commit()
+
+        retrieved = plan_repo.get("01HV00000000000000E2E0FC01")
+        assert retrieved is not None
+        assert retrieved.phase == PlanPhase.STRATEGIZE
+        assert retrieved.processing_state == ProcessingState.PROCESSING
+
+        # Phase 3: Complete strategize -> execute
+        retrieved.phase = PlanPhase.EXECUTE
+        retrieved.processing_state = ProcessingState.QUEUED
+        retrieved.timestamps.updated_at = datetime.now(tz=UTC)
+        plan_repo.update(retrieved)
+        session.commit()
+
+        retrieved = plan_repo.get("01HV00000000000000E2E0FC01")
+        assert retrieved is not None
+        assert retrieved.phase == PlanPhase.EXECUTE
+
+        # Phase 4: Execute -> apply -> applied
+        retrieved.phase = PlanPhase.APPLY
+        retrieved.processing_state = ProcessingState.APPLIED
+        retrieved.timestamps.updated_at = datetime.now(tz=UTC)
+        plan_repo.update(retrieved)
+        session.commit()
+
+        retrieved = plan_repo.get("01HV00000000000000E2E0FC01")
+        assert retrieved is not None
+        assert retrieved.phase == PlanPhase.APPLY
+        assert retrieved.processing_state == ProcessingState.APPLIED
+        assert retrieved.is_terminal
+
+        print("full-lifecycle-ok")
+    finally:
+        session.close()
+        Path(tmp).unlink(missing_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Restart persistence
+# ---------------------------------------------------------------------------
+
+
+def _restart_persistence() -> None:
+    """E2E: Verify plan survives DB close and reopen."""
+    tmp, session = _create_temp_db()
+    factory = lambda: session  # noqa: E731
+    try:
+        action_repo = ActionRepository(session_factory=factory)
+        action_repo.create(_make_action())
+        session.commit()
+
+        plan_repo = LifecyclePlanRepository(session_factory=factory)
+        plan = _make_plan(
+            "01HV00000000000000E2ERST01",
+            phase=PlanPhase.EXECUTE,
+            state=ProcessingState.PROCESSING,
+        )
+        plan_repo.create(plan)
+        session.commit()
+
+        # Close session (simulate process restart)
+        session.close()
+
+        # Reopen
+        engine2 = create_engine(f"sqlite:///{tmp}", echo=False)
+        session2 = Session(bind=engine2)
+        factory2 = lambda: session2  # noqa: E731
+        plan_repo2 = LifecyclePlanRepository(session_factory=factory2)
+
+        retrieved = plan_repo2.get("01HV00000000000000E2ERST01")
+        assert retrieved is not None
+        assert retrieved.phase == PlanPhase.EXECUTE
+        assert retrieved.processing_state == ProcessingState.PROCESSING
+        assert retrieved.description == "E2E test plan"
+        assert len(retrieved.project_links) == 1
+
+        session2.close()
+        engine2.dispose()
+
+        print("restart-persistence-ok")
+    finally:
+        Path(tmp).unlink(missing_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Concurrent access
+# ---------------------------------------------------------------------------
+
+
+def _concurrent_access() -> None:
+    """E2E: Two sessions see each other's committed data."""
+    tmp, session1 = _create_temp_db()
+    factory1 = lambda: session1  # noqa: E731
+    try:
+        action_repo = ActionRepository(session_factory=factory1)
+        action_repo.create(_make_action())
+        session1.commit()
+
+        plan_repo1 = LifecyclePlanRepository(session_factory=factory1)
+        plan = _make_plan("01HV00000000000000E2ECNC01")
+        plan_repo1.create(plan)
+        session1.commit()
+
+        # Open second session
+        engine2 = create_engine(f"sqlite:///{tmp}", echo=False)
+        session2 = Session(bind=engine2)
+        factory2 = lambda: session2  # noqa: E731
+        plan_repo2 = LifecyclePlanRepository(session_factory=factory2)
+
+        retrieved = plan_repo2.get("01HV00000000000000E2ECNC01")
+        assert retrieved is not None
+        assert retrieved.description == "E2E test plan"
+
+        session2.close()
+        engine2.dispose()
+
+        print("concurrent-access-ok")
+    finally:
+        session1.close()
+        Path(tmp).unlink(missing_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Action CRUD
+# ---------------------------------------------------------------------------
+
+
+def _action_crud() -> None:
+    """E2E: Create, read, update, delete an action."""
+    tmp, session = _create_temp_db()
+    factory = lambda: session  # noqa: E731
+    try:
+        repo = ActionRepository(session_factory=factory)
+
+        # Create
+        action = Action(
+            namespaced_name=NamespacedName.parse("local/crud-action"),
+            description="CRUD test",
+            definition_of_done="Done",
+            strategy_actor="local/s",
+            execution_actor="local/e",
+            arguments=[
+                ActionArgument(
+                    name="target",
+                    arg_type=ArgumentType.STRING,
+                    requirement=ArgumentRequirement.REQUIRED,
+                    description="Target path",
+                ),
+            ],
+            invariants=["No deletions"],
+            state=ActionState.AVAILABLE,
+            created_at=datetime(2026, 1, 1, tzinfo=UTC),
+            updated_at=datetime(2026, 1, 1, tzinfo=UTC),
+        )
+        repo.create(action)
+        session.commit()
+
+        # Read
+        fetched = repo.get_by_name("local/crud-action")
+        assert fetched is not None
+        assert fetched.description == "CRUD test"
+        assert len(fetched.arguments) == 1
+        assert len(fetched.invariants) == 1
+
+        # Update
+        fetched.description = "Updated CRUD test"
+        fetched.updated_at = datetime.now(tz=UTC)
+        repo.update(fetched)
+        session.commit()
+
+        refetched = repo.get_by_name("local/crud-action")
+        assert refetched is not None
+        assert refetched.description == "Updated CRUD test"
+
+        # Delete
+        deleted = repo.delete("local/crud-action")
+        assert deleted is True
+        session.commit()
+
+        gone = repo.get_by_name("local/crud-action")
+        assert gone is None
+
+        print("action-crud-ok")
+    finally:
+        session.close()
+        Path(tmp).unlink(missing_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Plan tree
+# ---------------------------------------------------------------------------
+
+
+def _plan_tree() -> None:
+    """E2E: Parent-child plan hierarchy persistence."""
+    tmp, session = _create_temp_db()
+    factory = lambda: session  # noqa: E731
+    try:
+        action_repo = ActionRepository(session_factory=factory)
+        action_repo.create(_make_action())
+        session.commit()
+
+        plan_repo = LifecyclePlanRepository(session_factory=factory)
+
+        # Parent
+        parent = _make_plan("01HV00000000000000E2ETRP01")
+        plan_repo.create(parent)
+        session.commit()
+
+        # Child
+        child = _make_plan(
+            "01HV00000000000000E2ETRC01",
+            parent_plan_id="01HV00000000000000E2ETRP01",
+            root_plan_id="01HV00000000000000E2ETRP01",
+        )
+        plan_repo.create(child)
+        session.commit()
+
+        loaded_child = plan_repo.get("01HV00000000000000E2ETRC01")
+        assert loaded_child is not None
+        assert loaded_child.identity.parent_plan_id == "01HV00000000000000E2ETRP01"
+        assert loaded_child.identity.root_plan_id == "01HV00000000000000E2ETRP01"
+        assert loaded_child.is_subplan
+
+        loaded_parent = plan_repo.get("01HV00000000000000E2ETRP01")
+        assert loaded_parent is not None
+        assert not loaded_parent.is_subplan
+
+        print("plan-tree-ok")
+    finally:
+        session.close()
+        Path(tmp).unlink(missing_ok=True)
+
+
+# ---------------------------------------------------------------------------
+# Main dispatcher
+# ---------------------------------------------------------------------------
+
+
+def main() -> None:
+    if len(sys.argv) < 2:
+        raise SystemExit("Expected command argument")
+    command = sys.argv[1]
+    commands = {
+        "full-lifecycle": _full_lifecycle,
+        "restart-persistence": _restart_persistence,
+        "concurrent-access": _concurrent_access,
+        "action-crud": _action_crud,
+        "plan-tree": _plan_tree,
+    }
+    if command not in commands:
+        raise SystemExit(f"Unknown command: {command}")
+    commands[command]()
+
+
+if __name__ == "__main__":
+    main()

robot/plan_persistence_e2e.robot

@@ -0,0 +1,52 @@
+*** Settings ***
+Documentation     E2E integration tests for plan persistence: full lifecycle,
+...               restart persistence, and concurrent CLI access patterns.
+...               Covers subtask 5 of A5.tests.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_plan_persistence_e2e.py
+
+*** Test Cases ***
+Plan Full Lifecycle Persistence
+    [Documentation]    Create action, create plan, transition through all phases,
+    ...                reach applied terminal state, and verify persistence at each step.
+    [Tags]    database    plan    lifecycle    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    full-lifecycle    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    full-lifecycle-ok
+
+Plan Restart Persistence
+    [Documentation]    Create a plan, close the database, reopen it, and verify
+    ...                all plan fields survive the reconnection.
+    [Tags]    database    plan    restart    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    restart-persistence    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    restart-persistence-ok
+
+Plan Concurrent Session Access
+    [Documentation]    Open two independent sessions against the same database
+    ...                and verify that plans created in one session are visible
+    ...                in the other after commit.
+    [Tags]    database    plan    concurrent    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    concurrent-access    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    concurrent-access-ok
+
+Action CRUD Persistence E2E
+    [Documentation]    Create, read, update, and delete an action via repository,
+    ...                verifying persistence at each step.
+    [Tags]    database    action    crud    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    action-crud    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    action-crud-ok
+
+Plan Tree Hierarchy Persistence E2E
+    [Documentation]    Create a parent plan and child plan, verify hierarchy
+    ...                links are persisted correctly including root_plan_id.
+    [Tags]    database    plan    hierarchy    persistence
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    plan-tree    cwd=${WORKSPACE}    timeout=30s
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    plan-tree-ok