alembic/versions/a5_005_rebaseline_plan_phases.py

@@ -0,0 +1,385 @@
+"""Rebaseline plan phase/state enums for Action + Apply terminal states.
+
+This migration updates the ``v3_plans`` table CHECK constraints to align with
+the four-phase lifecycle: Action -> Strategize -> Execute -> Apply.
+
+Changes:
+- ``ck_v3_plans_phase``: add ``'action'``, remove ``'applied'``
+- ``ck_v3_plans_state``: add ``'applied'`` and ``'constrained'``
+- Default ``phase`` changes from ``'strategize'`` to ``'action'``
+
+SQLite does not support ``ALTER TABLE ... DROP CONSTRAINT``, so we rebuild
+the ``v3_plans`` table using the standard temp-table-copy-rename approach.
+Child tables (``plan_projects``, ``plan_arguments``, ``plan_invariants``)
+have their FKs re-pointed during the rebuild.
+
+Revision ID: a5_005_rebaseline_plan_phases
+Revises: b1_001_resource_registry
+Create Date: 2026-02-15 12:00:00
+
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import Any
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "a5_005_rebaseline_plan_phases"
+down_revision: str | Sequence[str] | None = "b1_001_resource_registry"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+# ── Column lists (shared by upgrade/downgrade) ────────────────────────────
+
+_V3_PLANS_COLUMNS = [
+    sa.Column("plan_id", sa.String(26), nullable=False),
+    sa.Column("parent_plan_id", sa.String(26), nullable=True),
+    sa.Column("root_plan_id", sa.String(26), nullable=True),
+    sa.Column("action_name", sa.String(255), nullable=False),
+    sa.Column("namespaced_name", sa.String(255), nullable=False),
+    sa.Column("namespace", sa.String(100), nullable=False),
+    # phase and processing_state are added dynamically
+    sa.Column("attempt", sa.Integer(), nullable=False, server_default=sa.text("1")),
+    sa.Column("description", sa.Text(), nullable=False),
+    sa.Column("definition_of_done", sa.Text(), nullable=True),
+    sa.Column("strategy_actor", sa.String(255), nullable=True),
+    sa.Column("execution_actor", sa.String(255), nullable=True),
+    sa.Column("review_actor", sa.String(255), nullable=True),
+    sa.Column("apply_actor", sa.String(255), nullable=True),
+    sa.Column("estimation_actor", sa.String(255), nullable=True),
+    sa.Column("invariant_actor", sa.String(255), nullable=True),
+    sa.Column("automation_profile", sa.String(255), nullable=True),
+    sa.Column(
+        "automation_level", sa.String(30), nullable=False, server_default="manual"
+    ),
+    sa.Column("reusable", sa.Boolean(), nullable=False, server_default=sa.text("1")),
+    sa.Column("read_only", sa.Boolean(), nullable=False, server_default=sa.text("0")),
+    sa.Column("inputs_schema_json", sa.Text(), nullable=True),
+    sa.Column("changeset_id", sa.String(26), nullable=True),
+    sa.Column("sandbox_refs_json", sa.Text(), nullable=True),
+    sa.Column("validation_summary_json", sa.Text(), nullable=True),
+    sa.Column("decision_root_id", sa.String(26), nullable=True),
+    sa.Column("error_message", sa.Text(), nullable=True),
+    sa.Column("error_details_json", sa.Text(), nullable=True),
+    sa.Column("cost_estimate_usd", sa.Float(), nullable=True),
+    sa.Column("cost_actual_usd", sa.Float(), nullable=True),
+    sa.Column(
+        "token_count_input",
+        sa.Integer(),
+        nullable=True,
+        server_default=sa.text("0"),
+    ),
+    sa.Column(
+        "token_count_output",
+        sa.Integer(),
+        nullable=True,
+        server_default=sa.text("0"),
+    ),
+    sa.Column("created_by", sa.String(255), nullable=True),
+    sa.Column("tags_json", sa.Text(), nullable=False, server_default="[]"),
+    sa.Column("created_at", sa.String(30), nullable=False),
+    sa.Column("updated_at", sa.String(30), nullable=False),
+    sa.Column("completed_at", sa.String(30), nullable=True),
+    sa.Column("strategize_started_at", sa.String(30), nullable=True),
+    sa.Column("strategize_completed_at", sa.String(30), nullable=True),
+    sa.Column("execute_started_at", sa.String(30), nullable=True),
+    sa.Column("execute_completed_at", sa.String(30), nullable=True),
+    sa.Column("apply_started_at", sa.String(30), nullable=True),
+    sa.Column("applied_at", sa.String(30), nullable=True),
+]
+
+# Column names (excluding phase/processing_state which differ between versions)
+_ALL_DATA_COLUMNS = (
+    "plan_id, parent_plan_id, root_plan_id, action_name, "
+    "namespaced_name, namespace, phase, processing_state, attempt, "
+    "description, definition_of_done, "
+    "strategy_actor, execution_actor, review_actor, apply_actor, "
+    "estimation_actor, invariant_actor, "
+    "automation_profile, automation_level, "
+    "reusable, read_only, inputs_schema_json, "
+    "changeset_id, sandbox_refs_json, validation_summary_json, decision_root_id, "
+    "error_message, error_details_json, "
+    "cost_estimate_usd, cost_actual_usd, "
+    "token_count_input, token_count_output, "
+    "created_by, tags_json, "
+    "created_at, updated_at, completed_at, "
+    "strategize_started_at, strategize_completed_at, "
+    "execute_started_at, execute_completed_at, "
+    "apply_started_at, applied_at"
+)
+
+# ── New (upgrade) constraints ──────────────────────────────────────────────
+_NEW_PHASE_CK = "phase IN ('action', 'strategize', 'execute', 'apply')"
+_NEW_STATE_CK = (
+    "processing_state IN "
+    "('queued', 'processing', 'errored', 'complete', 'cancelled', "
+    "'applied', 'constrained')"
+)
+_NEW_PHASE_DEFAULT = "action"
+
+# ── Old (downgrade) constraints ────────────────────────────────────────────
+_OLD_PHASE_CK = "phase IN ('strategize', 'execute', 'apply', 'applied')"
+_OLD_STATE_CK = (
+    "processing_state IN ('queued', 'processing', 'errored', 'complete', 'cancelled')"
+)
+_OLD_PHASE_DEFAULT = "strategize"
+
+
+def _rebuild_v3_plans(
+    phase_check: str,
+    state_check: str,
+    phase_default: str,
+) -> None:
+    """Rebuild v3_plans with new CHECK constraints (SQLite-safe).
+
+    Steps:
+    1. Create ``_v3_plans_new`` with updated constraints.
+    2. Copy all data from ``v3_plans``.
+    3. Drop child-table FKs (SQLite: drop+recreate child tables).
+    4. Drop old ``v3_plans``.
+    5. Rename ``_v3_plans_new`` → ``v3_plans``.
+    6. Recreate child tables pointing at the renamed table.
+    7. Recreate indexes.
+    """
+    conn = op.get_bind()
+    dialect = conn.dialect.name
+
+    if dialect == "sqlite":
+        # Turn off FK enforcement during rebuild
+        conn.execute(sa.text("PRAGMA foreign_keys = OFF"))
+
+    # ── 1. Create temp table ──────────────────────────────────────────
+    columns: list[Any] = []
+    for col in _V3_PLANS_COLUMNS:
+        columns.append(col.copy())
+    # Insert phase and processing_state at the right position (after namespace)
+    phase_col: Any = sa.Column(
+        "phase", sa.String(20), nullable=False, server_default=phase_default
+    )
+    state_col: Any = sa.Column(
+        "processing_state", sa.String(20), nullable=False, server_default="queued"
+    )
+    # Insert after namespace (index 5 in the list)
+    columns.insert(6, phase_col)
+    columns.insert(7, state_col)
+
+    op.create_table(
+        "_v3_plans_new",
+        *columns,
+        sa.PrimaryKeyConstraint("plan_id"),
+        sa.ForeignKeyConstraint(
+            ["parent_plan_id"],
+            ["_v3_plans_new.plan_id"],
+            name="fk_v3_plans_parent",
+            ondelete="SET NULL",
+        ),
+        sa.ForeignKeyConstraint(
+            ["root_plan_id"],
+            ["_v3_plans_new.plan_id"],
+            name="fk_v3_plans_root",
+            ondelete="SET NULL",
+        ),
+        sa.ForeignKeyConstraint(
+            ["action_name"],
+            ["actions.namespaced_name"],
+            name="fk_v3_plans_action",
+            ondelete="RESTRICT",
+        ),
+        sa.CheckConstraint(phase_check, name="ck_v3_plans_phase"),
+        sa.CheckConstraint(state_check, name="ck_v3_plans_state"),
+        sa.CheckConstraint(
+            "automation_level IN ('manual', 'review_before_apply', 'full_automation')",
+            name="ck_v3_plans_automation",
+        ),
+    )
+
+    # ── 2. Copy data ─────────────────────────────────────────────────
+    conn.execute(
+        sa.text(
+            f"INSERT INTO _v3_plans_new ({_ALL_DATA_COLUMNS}) "
+            f"SELECT {_ALL_DATA_COLUMNS} FROM v3_plans"
+        )
+    )
+
+    # ── 3. Drop child tables (they have FKs to v3_plans) ─────────────
+    # Save child data first
+    conn.execute(
+        sa.text("CREATE TABLE _plan_projects_bak AS SELECT * FROM plan_projects")
+    )
+    conn.execute(
+        sa.text("CREATE TABLE _plan_arguments_bak AS SELECT * FROM plan_arguments")
+    )
+    conn.execute(
+        sa.text("CREATE TABLE _plan_invariants_bak AS SELECT * FROM plan_invariants")
+    )
+
+    op.drop_table("plan_invariants")
+    op.drop_table("plan_arguments")
+    op.drop_table("plan_projects")
+
+    # ── 4. Drop indexes and old v3_plans ──────────────────────────────
+    for idx_name in [
+        "ix_v3_plans_phase_state",
+        "ix_v3_plans_namespace",
+        "ix_v3_plans_action",
+        "ix_v3_plans_created",
+        "ix_v3_plans_root",
+        "ix_v3_plans_parent",
+        "ix_v3_plans_state",
+        "ix_v3_plans_phase",
+    ]:
+        op.drop_index(idx_name, table_name="v3_plans")
+    op.drop_table("v3_plans")
+
+    # ── 5. Rename temp → v3_plans ─────────────────────────────────────
+    op.rename_table("_v3_plans_new", "v3_plans")
+
+    # ── 6. Recreate child tables ──────────────────────────────────────
+    op.create_table(
+        "plan_projects",
+        sa.Column(
+            "plan_id",
+            sa.String(26),
+            sa.ForeignKey(
+                "v3_plans.plan_id",
+                ondelete="CASCADE",
+                name="fk_plan_projects_plan",
+            ),
+            nullable=False,
+        ),
+        sa.Column("project_name", sa.String(255), nullable=False),
+        sa.Column("alias", sa.String(100), nullable=True),
+        sa.Column(
+            "read_only", sa.Boolean(), nullable=False, server_default=sa.text("0")
+        ),
+        sa.Column("created_at", sa.String(30), nullable=False),
+        sa.PrimaryKeyConstraint("plan_id", "project_name"),
+    )
+    op.create_index(
+        "ix_plan_projects_project", "plan_projects", ["project_name"], unique=False
+    )
+
+    op.create_table(
+        "plan_arguments",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column(
+            "plan_id",
+            sa.String(26),
+            sa.ForeignKey(
+                "v3_plans.plan_id",
+                ondelete="CASCADE",
+                name="fk_plan_arguments_plan",
+            ),
+            nullable=False,
+        ),
+        sa.Column("name", sa.String(100), nullable=False),
+        sa.Column("value_json", sa.Text(), nullable=True),
+        sa.Column("value_type", sa.String(20), nullable=False, server_default="string"),
+        sa.Column("position", sa.Integer(), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("plan_id", "name", name="uq_plan_arguments_name"),
+    )
+    op.create_index(
+        "ix_plan_arguments_plan", "plan_arguments", ["plan_id"], unique=False
+    )
+    op.create_index(
+        "ix_plan_arguments_position",
+        "plan_arguments",
+        ["plan_id", "position"],
+        unique=False,
+    )
+
+    op.create_table(
+        "plan_invariants",
+        sa.Column("id", sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column(
+            "plan_id",
+            sa.String(26),
+            sa.ForeignKey(
+                "v3_plans.plan_id",
+                ondelete="CASCADE",
+                name="fk_plan_invariants_plan",
+            ),
+            nullable=False,
+        ),
+        sa.Column("invariant_text", sa.Text(), nullable=False),
+        sa.Column(
+            "source_scope",
+            sa.String(20),
+            nullable=False,
+            server_default="plan",
+        ),
+        sa.Column("source_name", sa.String(255), nullable=True),
+        sa.Column("position", sa.Integer(), nullable=False),
+        sa.Column("created_at", sa.String(30), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint(
+            "plan_id", "invariant_text", name="uq_plan_invariants_text"
+        ),
+        sa.CheckConstraint(
+            "source_scope IN ('global', 'project', 'action', 'plan')",
+            name="ck_plan_invariants_scope",
+        ),
+    )
+    op.create_index(
+        "ix_plan_invariants_plan", "plan_invariants", ["plan_id"], unique=False
+    )
+    op.create_index(
+        "ix_plan_invariants_position",
+        "plan_invariants",
+        ["plan_id", "position"],
+        unique=False,
+    )
+
+    # ── 7. Restore child data ─────────────────────────────────────────
+    conn.execute(sa.text("INSERT INTO plan_projects SELECT * FROM _plan_projects_bak"))
+    conn.execute(
+        sa.text("INSERT INTO plan_arguments SELECT * FROM _plan_arguments_bak")
+    )
+    conn.execute(
+        sa.text("INSERT INTO plan_invariants SELECT * FROM _plan_invariants_bak")
+    )
+
+    conn.execute(sa.text("DROP TABLE _plan_projects_bak"))
+    conn.execute(sa.text("DROP TABLE _plan_arguments_bak"))
+    conn.execute(sa.text("DROP TABLE _plan_invariants_bak"))
+
+    # ── 8. Recreate v3_plans indexes ──────────────────────────────────
+    op.create_index("ix_v3_plans_phase", "v3_plans", ["phase"], unique=False)
+    op.create_index("ix_v3_plans_state", "v3_plans", ["processing_state"], unique=False)
+    op.create_index("ix_v3_plans_parent", "v3_plans", ["parent_plan_id"], unique=False)
+    op.create_index("ix_v3_plans_root", "v3_plans", ["root_plan_id"], unique=False)
+    op.create_index("ix_v3_plans_created", "v3_plans", ["created_at"], unique=False)
+    op.create_index("ix_v3_plans_action", "v3_plans", ["action_name"], unique=False)
+    op.create_index("ix_v3_plans_namespace", "v3_plans", ["namespace"], unique=False)
+    op.create_index(
+        "ix_v3_plans_phase_state",
+        "v3_plans",
+        ["phase", "processing_state"],
+        unique=False,
+    )
+
+    if dialect == "sqlite":
+        conn.execute(sa.text("PRAGMA foreign_keys = ON"))
+
+
+def upgrade() -> None:
+    """Rebaseline phase/state constraints for Action phase + Apply terminals."""
+    _rebuild_v3_plans(
+        phase_check=_NEW_PHASE_CK,
+        state_check=_NEW_STATE_CK,
+        phase_default=_NEW_PHASE_DEFAULT,
+    )
+
+
+def downgrade() -> None:
+    """Revert to pre-rebaseline phase/state constraints."""
+    _rebuild_v3_plans(
+        phase_check=_OLD_PHASE_CK,
+        state_check=_OLD_STATE_CK,
+        phase_default=_OLD_PHASE_DEFAULT,
+    )

benchmarks/plan_phase_migration_bench.py

@@ -0,0 +1,153 @@
+"""ASV benchmarks for plan phase rebaseline migration (a5_005).
+
+Measures:
+- Schema creation time with rebaselined constraints
+- Plan insert throughput with Action phase
+- Plan insert throughput with Apply terminal states
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from cleveragents.infrastructure.database.models import (
+    Base,
+    LifecycleActionModel,
+    LifecyclePlanModel,
+)
+
+
+def _now_iso() -> str:
+    return datetime.now(tz=UTC).isoformat()
+
+
+def _make_ulid(n: int) -> str:
+    return str(n).zfill(26)
+
+
+class SchemaCreationWithRebaseline:
+    """Benchmark schema creation including rebaselined v3_plans constraints."""
+
+    def time_create_all_tables(self) -> None:
+        engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(engine)
+        engine.dispose()
+
+
+class PlanInsertActionPhase:
+    """Benchmark inserting plans with phase='action' (new phase)."""
+
+    def setup(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.session_factory = sessionmaker(bind=self.engine)
+
+        # Create parent action
+        session = self.session_factory()
+        action = LifecycleActionModel()
+        action.namespaced_name = "bench/phase-action"
+        action.namespace = "bench"
+        action.name = "phase-action"
+        action.description = "Benchmark action"
+        action.definition_of_done = "Done"
+        action.strategy_actor = "bench/s"
+        action.execution_actor = "bench/e"
+        action.state = "available"
+        action.tags_json = "[]"
+        action.created_at = _now_iso()
+        action.updated_at = _now_iso()
+        session.add(action)
+        session.commit()
+        session.close()
+
+    def teardown(self) -> None:
+        self.engine.dispose()
+
+    def time_insert_100_action_phase_plans(self) -> None:
+        session = self.session_factory()
+        now = _now_iso()
+        for i in range(100):
+            plan = LifecyclePlanModel()
+            plan.plan_id = _make_ulid(i)
+            plan.action_name = "bench/phase-action"
+            plan.namespaced_name = f"bench/plan-{i}"
+            plan.namespace = "bench"
+            plan.phase = "action"
+            plan.processing_state = "queued"
+            plan.description = f"Benchmark plan {i}"
+            plan.tags_json = "[]"
+            plan.created_at = now
+            plan.updated_at = now
+            session.add(plan)
+        session.commit()
+        session.close()
+
+
+class PlanInsertApplyTerminalStates:
+    """Benchmark inserting plans with Apply terminal states."""
+
+    def setup(self) -> None:
+        self.engine = create_engine("sqlite:///:memory:")
+        Base.metadata.create_all(self.engine)
+        self.session_factory = sessionmaker(bind=self.engine)
+
+        session = self.session_factory()
+        action = LifecycleActionModel()
+        action.namespaced_name = "bench/terminal-action"
+        action.namespace = "bench"
+        action.name = "terminal-action"
+        action.description = "Benchmark terminal action"
+        action.definition_of_done = "Done"
+        action.strategy_actor = "bench/s"
+        action.execution_actor = "bench/e"
+        action.state = "available"
+        action.tags_json = "[]"
+        action.created_at = _now_iso()
+        action.updated_at = _now_iso()
+        session.add(action)
+        session.commit()
+        session.close()
+
+    def teardown(self) -> None:
+        self.engine.dispose()
+
+    def time_insert_50_applied_plans(self) -> None:
+        session = self.session_factory()
+        now = _now_iso()
+        for i in range(50):
+            plan = LifecyclePlanModel()
+            plan.plan_id = _make_ulid(1000 + i)
+            plan.action_name = "bench/terminal-action"
+            plan.namespaced_name = f"bench/applied-{i}"
+            plan.namespace = "bench"
+            plan.phase = "apply"
+            plan.processing_state = "applied"
+            plan.description = f"Applied plan {i}"
+            plan.tags_json = "[]"
+            plan.created_at = now
+            plan.updated_at = now
+            session.add(plan)
+        session.commit()
+        session.close()
+
+    def time_insert_50_constrained_plans(self) -> None:
+        session = self.session_factory()
+        now = _now_iso()
+        for i in range(50):
+            plan = LifecyclePlanModel()
+            plan.plan_id = _make_ulid(2000 + i)
+            plan.action_name = "bench/terminal-action"
+            plan.namespaced_name = f"bench/constrained-{i}"
+            plan.namespace = "bench"
+            plan.phase = "apply"
+            plan.processing_state = "constrained"
+            plan.description = f"Constrained plan {i}"
+            plan.tags_json = "[]"
+            plan.created_at = now
+            plan.updated_at = now
+            session.add(plan)
+        session.commit()
+        session.close()

benchmarks/plan_repository_bench.py

@@ -0,0 +1,191 @@
+"""ASV benchmarks for LifecyclePlanRepository.
+
+Measures create, get, list, and count performance.
+"""
+
+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, ActionState
+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"01HGZ6FE0AQDYTR4BX{suffix}"
+
+
+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="bench-plan"),
+        action_name="local/bench-action",
+        description="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 TimePlanCreate:
+    """Benchmark plan creation."""
+
+    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 = Action(
+            namespaced_name=NamespacedName(namespace="local", name="bench-action"),
+            description="Benchmark action",
+            long_description=None,
+            definition_of_done="Done",
+            strategy_actor="local/s",
+            execution_actor="local/e",
+            estimation_actor=None,
+            review_actor=None,
+            arguments=[],
+            reusable=True,
+            read_only=False,
+            state=ActionState("available"),
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            created_by=None,
+            tags=[],
+        )
+        action_repo.create(action)
+        session.commit()
+        self.repo = LifecyclePlanRepository(session_factory=factory)
+
+    def time_create_single(self) -> None:
+        plan = _make_bench_plan()
+        self.repo.create(plan)
+        self.session.commit()
+
+
+class TimePlanGet:
+    """Benchmark plan retrieval."""
+
+    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 = Action(
+            namespaced_name=NamespacedName(namespace="local", name="bench-action"),
+            description="Benchmark action",
+            long_description=None,
+            definition_of_done="Done",
+            strategy_actor="local/s",
+            execution_actor="local/e",
+            estimation_actor=None,
+            review_actor=None,
+            arguments=[],
+            reusable=True,
+            read_only=False,
+            state=ActionState("available"),
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            created_by=None,
+            tags=[],
+        )
+        action_repo.create(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_get_by_id(self) -> None:
+        self.repo.get(self.plan_id)
+
+
+class TimePlanList:
+    """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 = Action(
+            namespaced_name=NamespacedName(namespace="local", name="bench-action"),
+            description="Benchmark action",
+            long_description=None,
+            definition_of_done="Done",
+            strategy_actor="local/s",
+            execution_actor="local/e",
+            estimation_actor=None,
+            review_actor=None,
+            arguments=[],
+            reusable=True,
+            read_only=False,
+            state=ActionState("available"),
+            created_at=datetime.now(),
+            updated_at=datetime.now(),
+            created_by=None,
+            tags=[],
+        )
+        action_repo.create(action)
+        session.commit()
+        self.repo = LifecyclePlanRepository(session_factory=factory)
+        for _ in range(50):
+            plan = _make_bench_plan()
+            self.repo.create(plan)
+        session.commit()
+
+    def time_list_all(self) -> None:
+        self.repo.list_plans()
+
+    def time_list_with_phase_filter(self) -> None:
+        self.repo.list_plans(phase="action")
+
+    def time_count_all(self) -> None:
+        self.repo.count()

benchmarks/uow_lifecycle_bench.py

@@ -0,0 +1,198 @@
+"""ASV benchmarks for UnitOfWork lifecycle repository wiring.
+
+Measures overhead of creating UnitOfWorkContext and accessing
+actions / lifecycle_plans repositories.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.domain.models.core.action import Action, ActionState
+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.unit_of_work import UnitOfWorkContext
+
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_bench_counter = 5000
+
+
+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"01HGZ6FE0AQDYTR4BX{suffix}"
+
+
+def _make_bench_action(name: str = "local/bench-uow-action") -> Action:
+    parts = name.split("/", 1)
+    namespace = parts[0] if len(parts) == 2 else "local"
+    short_name = parts[1] if len(parts) == 2 else parts[0]
+    now = datetime.now()
+    return Action(
+        namespaced_name=NamespacedName(namespace=namespace, name=short_name),
+        description=f"Bench action {short_name}",
+        long_description=None,
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        estimation_actor=None,
+        review_actor=None,
+        arguments=[],
+        reusable=True,
+        read_only=False,
+        state=ActionState("available"),
+        created_at=now,
+        updated_at=now,
+        created_by=None,
+        tags=[],
+    )
+
+
+def _make_bench_plan(
+    plan_id: str | None = None,
+    action_name: str = "local/bench-uow-action",
+) -> V3Plan:
+    now = datetime.now()
+    return V3Plan(
+        identity=PlanIdentity(plan_id=plan_id or _bench_ulid(), attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="bench-uow-plan"),
+        action_name=action_name,
+        description="Bench plan",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        phase=PlanPhase("action"),
+        processing_state=ProcessingState("queued"),
+        automation_level=AutomationLevel("manual"),
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        project_links=[],
+        arguments={},
+        arguments_order=[],
+        invariants=[],
+        reusable=True,
+        read_only=False,
+        created_by=None,
+        tags=[],
+    )
+
+
+class TimeUowContextCreation:
+    """Benchmark UnitOfWorkContext instantiation."""
+
+    timeout = 30.0
+
+    def setup(self) -> None:
+        self.engine = create_engine(
+            "sqlite:///:memory:",
+            echo=False,
+            future=True,
+            connect_args={"check_same_thread": False},
+        )
+
+        @event.listens_for(self.engine, "connect")
+        def _fk(dbapi_conn, _rec):  # type: ignore[no-untyped-def]
+            cursor = dbapi_conn.cursor()
+            cursor.execute("PRAGMA foreign_keys=ON")
+            cursor.close()
+
+        Base.metadata.create_all(self.engine)
+        self.sf: sessionmaker[Session] = sessionmaker(
+            bind=self.engine,
+            expire_on_commit=False,
+            autoflush=False,
+            autocommit=False,
+            class_=Session,
+        )
+
+    def time_create_context(self) -> None:
+        session = self.sf()
+        _ = UnitOfWorkContext(session)
+        session.close()
+
+    def time_access_actions_repo(self) -> None:
+        session = self.sf()
+        ctx = UnitOfWorkContext(session)
+        _ = ctx.actions
+        session.close()
+
+    def time_access_lifecycle_plans_repo(self) -> None:
+        session = self.sf()
+        ctx = UnitOfWorkContext(session)
+        _ = ctx.lifecycle_plans
+        session.close()
+
+
+class TimeUowActionPlanRoundTrip:
+    """Benchmark creating action + plan via UoW and retrieving."""
+
+    timeout = 30.0
+
+    def setup(self) -> None:
+        self.engine = create_engine(
+            "sqlite:///:memory:",
+            echo=False,
+            future=True,
+            connect_args={"check_same_thread": False},
+        )
+
+        @event.listens_for(self.engine, "connect")
+        def _fk(dbapi_conn, _rec):  # type: ignore[no-untyped-def]
+            cursor = dbapi_conn.cursor()
+            cursor.execute("PRAGMA foreign_keys=ON")
+            cursor.close()
+
+        Base.metadata.create_all(self.engine)
+        self.sf: sessionmaker[Session] = sessionmaker(
+            bind=self.engine,
+            expire_on_commit=False,
+            autoflush=False,
+            autocommit=False,
+            class_=Session,
+        )
+        # Seed an action for plan creation benchmarks
+        session = self.sf()
+        ctx = UnitOfWorkContext(session)
+        ctx.actions.create(_make_bench_action("local/bench-uow-action"))
+        session.commit()
+        session.close()
+
+    def time_create_action_via_uow(self) -> None:
+        session = self.sf()
+        ctx = UnitOfWorkContext(session)
+        uid = _bench_ulid()
+        ctx.actions.create(_make_bench_action(f"local/bench-action-{uid}"))
+        session.commit()
+        session.close()
+
+    def time_create_plan_via_uow(self) -> None:
+        session = self.sf()
+        ctx = UnitOfWorkContext(session)
+        ctx.lifecycle_plans.create(_make_bench_plan())
+        session.commit()
+        session.close()
+
+    def time_create_action_and_plan_via_uow(self) -> None:
+        session = self.sf()
+        ctx = UnitOfWorkContext(session)
+        uid = _bench_ulid()
+        action_name = f"local/bench-combo-{uid}"
+        ctx.actions.create(_make_bench_action(action_name))
+        ctx.lifecycle_plans.create(_make_bench_plan(action_name=action_name))
+        session.commit()
+        session.close()

docs/reference/database_schema.md

@@ -132,7 +132,7 @@ hierarchical subplan trees via self-referencing foreign keys.
 | `action_name` | String(255) | No | -- | FK to `actions.namespaced_name` (RESTRICT) |
 | `namespaced_name` | String(255) | No | -- | Plan's own namespaced name |
 | `namespace` | String(100) | No | -- | Extracted namespace |
-| `phase` | String(20) | No | `'strategize'` | Current lifecycle phase |
+| `phase` | String(20) | No | `'action'` | Current lifecycle phase |
 | `processing_state` | String(20) | No | `'queued'` | Current processing state |
 | `attempt` | Integer | No | `1` | Retry attempt counter |
 | `description` | Text | No | -- | Rendered description (from action template + args) |
@@ -144,7 +144,7 @@ hierarchical subplan trees via self-referencing foreign keys.
 | `estimation_actor` | String(255) | Yes | NULL | Estimation actor |
 | `invariant_actor` | String(255) | Yes | NULL | Invariant reconciliation actor |
 | `automation_profile` | String(255) | Yes | NULL | Resolved automation profile name |
-| `automation_level` | String(30) | No | `'manual'` | `manual`, `review_before_apply`, `full_automation` |
+| `automation_level` | String(30) | No | `'manual'` | **Legacy.** `manual`, `review_before_apply`, `full_automation`. Retained for backward compatibility; prefer `automation_profile`. |
 | `reusable` | Boolean | No | `1` | Copied from action |
 | `read_only` | Boolean | No | `0` | Copied from action |
 | `inputs_schema_json` | Text | Yes | NULL | JSON Schema dict (copied from action) |
@@ -410,6 +410,7 @@ resources (PK: resource_id)
 | `a5_003_spec_aligned_actions` | Drops `lifecycle_plans` and `actions_v3`; creates `actions`, `action_invariants`, `action_arguments` |
 | `a5_004_spec_aligned_plans` | Creates `v3_plans`, `plan_projects`, `plan_arguments`, `plan_invariants` |
 | `b1_001_resource_registry` | Creates `resource_types`, `resources`, `resource_edges` |
+| `a5_005_rebaseline_plan_phases` | Rebaselines `v3_plans` phase/state CHECK constraints: adds `action` phase, removes `applied` phase, adds `applied`/`constrained` processing states, changes default phase to `action` |
 
 ## Source Location
 

docs/reference/repositories.md

@@ -0,0 +1,133 @@
+# Repository Reference
+
+## Overview
+
+Repositories implement the persistence contract defined in ADR-007 (Repository Pattern). Each repository targets a single aggregate root and follows the session-factory pattern: public methods obtain their own session from the factory, flush (but do **not** commit), and delegate transaction control to the caller or a `UnitOfWork` wrapper.
+
+All public methods are decorated with `@database_retry` (ADR-033) to handle transient database errors.
+
+## LifecyclePlanRepository
+
+**Module**: `src/cleveragents/infrastructure/database/repositories.py`
+
+Provides CRUD operations for v3 lifecycle plans (`v3_plans` table and its child tables: `plan_projects`, `plan_arguments`, `plan_invariants`).
+
+### Constructor
+
+```python
+LifecyclePlanRepository(session_factory: Callable[[], Session])
+```
+
+### Methods
+
+| Method | Signature | Description |
+|--------|-----------|-------------|
+| `create` | `create(plan) -> Plan` | Persist a new plan via `LifecyclePlanModel.from_domain()`. Handles child tables. Raises `DuplicatePlanError` on duplicate `plan_id`. |
+| `get` | `get(plan_id: str) -> Plan \| None` | Retrieve by ULID primary key. Returns domain model or `None`. |
+| `get_by_name` | `get_by_name(namespaced_name: str) -> Plan \| None` | Retrieve by full namespaced name. Returns domain model or `None`. |
+| `update` | `update(plan) -> Plan` | Update phase, processing_state, timestamps, actor refs, error fields, and replace all child rows. Raises `PlanNotFoundError` if the plan does not exist. |
+| `delete` | `delete(plan_id: str) -> bool` | Delete plan and cascade to child tables. Returns `True` if deleted, `False` if not found. |
+| `list_plans` | `list_plans(phase?, processing_state?, action_name?, project_name?, namespace?, limit=100, offset=0) -> list[Plan]` | Filter and paginate plans. Deterministic ordering by `created_at DESC`. |
+| `count` | `count(phase?, processing_state?) -> int` | Count matching plans with optional filters. |
+
+### Filters
+
+The `list_plans` method supports the following filters, matching CLI flags:
+
+- **phase**: Filter by lifecycle phase (`action`, `strategize`, `execute`, `apply`).
+- **processing_state**: Filter by processing state (`queued`, `processing`, `errored`, `complete`, `cancelled`, `applied`, `constrained`).
+- **action_name**: Filter by the action template name (FK to `actions` table).
+- **project_name**: Filter by project name (joins `plan_projects` child table).
+- **namespace**: Filter by plan namespace.
+
+Results are always ordered by `created_at DESC` for deterministic pagination.
+
+### Custom Exceptions
+
+| Exception | Parent | Description |
+|-----------|--------|-------------|
+| `DuplicatePlanError` | `DatabaseError` | Raised when creating a plan with a `plan_id` that already exists. |
+| `PlanNotFoundError` | `DatabaseError` | Raised when updating or deleting a plan that does not exist. |
+
+### Session and Transaction Pattern
+
+```python
+# Session-factory pattern: each method gets its own session
+repo = LifecyclePlanRepository(session_factory=lambda: session)
+
+# All mutations flush but do NOT commit
+plan = repo.create(plan_domain_object)
+# Caller must commit:
+session.commit()
+```
+
+### Child Table Handling
+
+- **plan_projects**: Stores plan-to-project links with alias and read_only flags.
+- **plan_arguments**: Stores resolved argument values with stable ordering via `position`.
+- **plan_invariants**: Stores invariant constraints with source provenance (`plan`, `action`, `project`, `global`).
+
+On `create`, child rows are populated from the domain model. On `update`, all child rows are replaced (clear + re-add) to ensure consistency.
+
+## ActionRepository
+
+**Module**: `src/cleveragents/infrastructure/database/repositories.py`
+
+Provides CRUD operations for v3 lifecycle actions (`actions` table and its child tables: `action_arguments`, `action_invariants`).
+
+See `ActionRepository` class documentation in `repositories.py` for full method signatures.
+
+### Custom Exceptions
+
+| Exception | Parent | Description |
+|-----------|--------|-------------|
+| `DuplicateActionError` | `DatabaseError` | Raised when creating an action with a name that already exists. |
+| `ActionInUseError` | `BusinessRuleViolation` | Raised when deleting an action still referenced by plans. |
+
+## UnitOfWork Lifecycle Usage
+
+**Module**: `src/cleveragents/infrastructure/database/unit_of_work.py`
+
+The `UnitOfWork` class provides transactional access to all repositories. The `UnitOfWorkContext` exposes both legacy and v3 lifecycle repositories within a single transaction boundary.
+
+### Available Repositories on UnitOfWorkContext
+
+| Property | Type | Description |
+|----------|------|-------------|
+| `actions` | `ActionRepository` | V3 lifecycle action persistence. |
+| `lifecycle_plans` | `LifecyclePlanRepository` | V3 lifecycle plan persistence. |
+| `projects` | `ProjectRepository` | Legacy project persistence. |
+| `plans` | `PlanRepository` | Legacy plan persistence (deprecated; use `lifecycle_plans`). |
+| `contexts` | `ContextRepository` | Context item persistence. |
+| `changes` | `ChangeRepository` | Change persistence. |
+| `debug_attempts` | `DebugAttemptRepository` | Debug attempt persistence. |
+| `actors` | `ActorRepository` | Actor persistence. |
+
+### Transaction Pattern
+
+```python
+from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
+
+uow = UnitOfWork(database_url="sqlite:///app.db")
+
+# All operations within the context manager are committed atomically
+with uow.transaction() as ctx:
+    ctx.actions.create(action)
+    ctx.lifecycle_plans.create(plan)
+    # Both persisted on successful exit; rolled back on exception
+```
+
+### Session-Factory Bridging
+
+The v3 repositories (`ActionRepository`, `LifecyclePlanRepository`) use a session-factory pattern (`Callable[[], Session]`), while legacy repositories take a `Session` directly. `UnitOfWorkContext` bridges both patterns by providing a `_session_factory()` method that returns the transaction's session, ensuring all repositories operate within the same transaction boundary.
+
+### Cross-Repository Atomicity
+
+Creating an action and a plan in the same transaction guarantees that either both persist or neither does:
+
+```python
+with uow.transaction() as ctx:
+    ctx.actions.create(action)
+    ctx.lifecycle_plans.create(plan)
+    # If plan creation fails, the action is also rolled back
+```

features/plan_phase_migration.feature

@@ -0,0 +1,32 @@
+@database @migration @plan_phase_rebaseline
+Feature: Plan phase/state constraint rebaseline
+  As the system operator
+  I want the v3_plans table to accept the Action phase and Apply terminal states
+  So that plans can follow the four-phase lifecycle: Action -> Strategize -> Execute -> Apply
+
+  Background:
+    Given the plan phase rebaseline database is initialized
+
+  Scenario: Phase constraint accepts 'action' phase
+    When I insert a plan with phase "action" and state "queued"
+    Then the plan should be persisted successfully with phase "action"
+
+  Scenario: Phase constraint rejects 'applied' as a phase
+    When I try to insert a plan with phase "applied" and state "queued"
+    Then the insert should fail with a constraint violation
+
+  Scenario: Processing state accepts 'applied' state
+    When I insert a plan with phase "apply" and state "applied"
+    Then the plan should be persisted successfully with state "applied"
+
+  Scenario: Processing state accepts 'constrained' state
+    When I insert a plan with phase "apply" and state "constrained"
+    Then the plan should be persisted successfully with state "constrained"
+
+  Scenario: Default phase is 'action'
+    When I insert a plan using default phase
+    Then the rebaselined plan phase should be "action"
+
+  Scenario: Phase constraint rejects legacy 'applied' phase value
+    When I try to insert a plan with phase "applied" and state "complete"
+    Then the insert should fail with a constraint violation

features/plan_repository.feature

@@ -0,0 +1,125 @@
+@phase1 @domain @repository @plan_repository
+Feature: Lifecycle Plan Repository Persistence
+  As a system operator managing lifecycle plans
+  I want the plan repository to reliably persist and retrieve plans
+  So that plan state survives process restarts and supports CLI list/filter commands
+
+  Background:
+    Given a clean in-memory database for plan repository tests
+    And a lifecycle plan repository backed by the database
+
+  # ---------------------------------------------------------------------------
+  # Create and retrieve
+  # ---------------------------------------------------------------------------
+
+  @plan_create @plan_get
+  Scenario: Create a plan and retrieve it by ID
+    Given a valid plan domain object with id "01HGZ6FE0AQDYTR4BX00000001"
+    When the plan is saved through the plan repository
+    Then the plan repository should not raise an error
+    And the plan retrieved by id "01HGZ6FE0AQDYTR4BX00000001" should exist
+    And the retrieved plan description should be "Test plan description"
+
+  @plan_create @plan_get_by_name
+  Scenario: Create a plan and retrieve it by name
+    Given a valid plan domain object with id "01HGZ6FE0AQDYTR4BX00000002"
+    When the plan is saved through the plan repository
+    Then the plan retrieved by name "local/test-plan" should exist
+
+  # ---------------------------------------------------------------------------
+  # Update
+  # ---------------------------------------------------------------------------
+
+  @plan_update
+  Scenario: Update plan phase and processing state
+    Given a valid plan domain object with id "01HGZ6FE0AQDYTR4BX00000003"
+    And the plan is saved through the plan repository
+    When the plan phase is updated to "strategize" with state "processing"
+    Then the retrieved plan should have phase "strategize"
+    And the retrieved plan should have processing state "processing"
+
+  # ---------------------------------------------------------------------------
+  # List filters
+  # ---------------------------------------------------------------------------
+
+  @plan_list @plan_filter_phase
+  Scenario: List plans filtered by phase
+    Given 3 plans exist in phase "action" and 2 plans in phase "strategize"
+    When plans are listed with phase filter "action"
+    Then the plan list should contain 3 plans
+
+  @plan_list @plan_filter_state
+  Scenario: List plans filtered by processing state
+    Given 2 plans exist in state "queued" and 1 plan in state "processing"
+    When plans are listed with processing state filter "queued"
+    Then the plan list should contain 2 plans
+
+  @plan_list @plan_filter_action
+  Scenario: List plans filtered by action name
+    Given 2 plans referencing action "local/deploy" and 1 plan referencing "local/review"
+    When plans are listed with action name filter "local/deploy"
+    Then the plan list should contain 2 plans
+
+  @plan_list @plan_filter_project
+  Scenario: List plans filtered by project name
+    Given a plan linked to project "local/api-service" and another with no project link
+    When plans are listed with project name filter "local/api-service"
+    Then the plan list should contain 1 plan
+
+  # ---------------------------------------------------------------------------
+  # Delete
+  # ---------------------------------------------------------------------------
+
+  @plan_delete
+  Scenario: Delete plan cascades to child tables
+    Given a valid plan domain object with id "01HGZ6FE0AQDYTR4BX00000010" and child rows
+    And the plan is saved through the plan repository
+    When the plan with id "01HGZ6FE0AQDYTR4BX00000010" is deleted
+    Then the plan retrieved by id "01HGZ6FE0AQDYTR4BX00000010" should not exist
+    And the child rows for plan "01HGZ6FE0AQDYTR4BX00000010" should be gone
+
+  # ---------------------------------------------------------------------------
+  # Error handling
+  # ---------------------------------------------------------------------------
+
+  @plan_create @error_handling
+  Scenario: Creating duplicate plan raises error
+    Given a valid plan domain object with id "01HGZ6FE0AQDYTR4BX00000011"
+    And the plan is saved through the plan repository
+    When a duplicate plan with id "01HGZ6FE0AQDYTR4BX00000011" is saved
+    Then a DuplicatePlanError should be raised mentioning "01HGZ6FE0AQDYTR4BX00000011"
+
+  @plan_get @error_handling
+  Scenario: Getting non-existent plan returns None
+    When the plan is retrieved by id "01HGZ6FE0AQDYTR4BXZZZZZZZZ"
+    Then the plan retrieve result should be None
+
+  # ---------------------------------------------------------------------------
+  # Phase and state persistence
+  # ---------------------------------------------------------------------------
+
+  @plan_phase @action_phase
+  Scenario: Plan with Action phase persists correctly
+    Given a plan domain object in phase "action" with state "queued" and id "01HGZ6FE0AQDYTR4BX00000012"
+    When the plan is saved through the plan repository
+    Then the retrieved plan should have phase "action"
+    And the retrieved plan should have processing state "queued"
+
+  @plan_phase @apply_terminal
+  Scenario: Plan with Apply terminal state persists correctly
+    Given a plan domain object in phase "apply" with state "applied" and id "01HGZ6FE0AQDYTR4BX00000013"
+    When the plan is saved through the plan repository
+    Then the retrieved plan should have phase "apply"
+    And the retrieved plan should have processing state "applied"
+
+  # ---------------------------------------------------------------------------
+  # Child row ordering
+  # ---------------------------------------------------------------------------
+
+  @plan_children @ordering
+  Scenario: Child rows (projects, arguments, invariants) persist in order
+    Given a plan with ordered children and id "01HGZ6FE0AQDYTR4BX00000014"
+    When the plan is saved through the plan repository
+    Then the retrieved plan should have arguments in order "alpha,beta,gamma"
+    And the retrieved plan should have invariants in order "first,second,third"
+    And the retrieved plan should have project links including "local/proj-a"

features/steps/plan_phase_migration_steps.py

@@ -0,0 +1,217 @@
+"""Step definitions for plan phase/state constraint rebaseline tests."""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+
+from behave import given, then, when
+from sqlalchemy import create_engine, text
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.infrastructure.database.models import (
+    Base,
+    LifecycleActionModel,
+    LifecyclePlanModel,
+)
+
+_VALID_ULID_PLAN = "01HV000000000000000000PH01"
+_VALID_ULID_COUNTER = 0
+
+
+def _next_ulid() -> str:
+    global _VALID_ULID_COUNTER
+    _VALID_ULID_COUNTER += 1
+    suffix = str(_VALID_ULID_COUNTER).zfill(4)
+    return f"01HV0000000000000000PH{suffix}"
+
+
+def _now_iso() -> str:
+    return datetime.now(tz=UTC).isoformat()
+
+
+def _ensure_action(session: Session) -> None:
+    """Ensure a parent action exists for FK references."""
+    existing = (
+        session.query(LifecycleActionModel)
+        .filter_by(namespaced_name="local/phase-test-action")
+        .first()
+    )
+    if existing is not None:
+        return
+
+    action = LifecycleActionModel()
+    action.namespaced_name = "local/phase-test-action"
+    action.namespace = "local"
+    action.name = "phase-test-action"
+    action.description = "Test action for phase migration"
+    action.definition_of_done = "Done"
+    action.strategy_actor = "local/s"
+    action.execution_actor = "local/e"
+    action.state = "available"
+    action.tags_json = "[]"
+    action.created_at = _now_iso()
+    action.updated_at = _now_iso()
+    session.add(action)
+    session.commit()
+
+
+@given("the plan phase rebaseline database is initialized")
+def step_plan_phase_rebaseline_db_init(context: Any) -> None:
+    """Set up an in-memory database with the schema including new constraints."""
+    engine = create_engine("sqlite:///:memory:")
+
+    # Enable FK enforcement
+    from sqlalchemy import event
+
+    @event.listens_for(engine, "connect")
+    def set_sqlite_pragma(dbapi_connection: Any, _: Any) -> None:
+        cursor = dbapi_connection.cursor()
+        cursor.execute("PRAGMA foreign_keys=ON")
+        cursor.close()
+
+    Base.metadata.create_all(engine)
+    factory = sessionmaker(bind=engine)
+    session = factory()
+    _ensure_action(session)
+    context.phase_rebaseline_engine = engine
+    context.phase_rebaseline_session = session
+
+
+@when('I insert a plan with phase "{phase}" and state "{state}"')
+def step_insert_plan_with_phase_and_state(context: Any, phase: str, state: str) -> None:
+    """Insert a plan with the specified phase and state."""
+    session: Session = context.phase_rebaseline_session
+    ulid = _next_ulid()
+    now = _now_iso()
+    plan = LifecyclePlanModel()
+    plan.plan_id = ulid
+    plan.action_name = "local/phase-test-action"
+    plan.namespaced_name = "local/phase-test-plan"
+    plan.namespace = "local"
+    plan.phase = phase
+    plan.processing_state = state
+    plan.description = "Test plan"
+    plan.tags_json = "[]"
+    plan.created_at = now
+    plan.updated_at = now
+    try:
+        session.add(plan)
+        session.commit()
+        context.phase_rebaseline_result = "ok"
+        context.phase_rebaseline_plan_id = ulid
+    except (IntegrityError, Exception) as exc:
+        session.rollback()
+        context.phase_rebaseline_result = "error"
+        context.phase_rebaseline_error = str(exc)
+
+
+@when('I try to insert a plan with phase "{phase}" and state "{state}"')
+def step_try_insert_plan_with_phase_and_state(
+    context: Any, phase: str, state: str
+) -> None:
+    """Try to insert a plan with the specified phase and state, expecting failure."""
+    session: Session = context.phase_rebaseline_session
+    ulid = _next_ulid()
+    now = _now_iso()
+    try:
+        session.execute(
+            text(
+                "INSERT INTO v3_plans "
+                "(plan_id, action_name, namespaced_name, namespace, "
+                "phase, processing_state, description, tags_json, "
+                "created_at, updated_at) "
+                "VALUES (:pid, :aname, :nname, :ns, :phase, :state, "
+                ":desc, :tags, :cat, :uat)"
+            ),
+            {
+                "pid": ulid,
+                "aname": "local/phase-test-action",
+                "nname": "local/try-plan",
+                "ns": "local",
+                "phase": phase,
+                "state": state,
+                "desc": "Test plan",
+                "tags": "[]",
+                "cat": now,
+                "uat": now,
+            },
+        )
+        session.commit()
+        context.phase_rebaseline_result = "ok"
+    except (IntegrityError, Exception) as exc:
+        session.rollback()
+        context.phase_rebaseline_result = "error"
+        context.phase_rebaseline_error = str(exc)
+
+
+@when("I insert a plan using default phase")
+def step_insert_plan_default_phase(context: Any) -> None:
+    """Insert a plan without specifying a phase to test the ORM default."""
+    session: Session = context.phase_rebaseline_session
+    ulid = _next_ulid()
+    now = _now_iso()
+    # Use ORM to create plan without specifying phase — should default to 'action'
+    plan = LifecyclePlanModel()
+    plan.plan_id = ulid
+    plan.action_name = "local/phase-test-action"
+    plan.namespaced_name = "local/default-phase-plan"
+    plan.namespace = "local"
+    plan.description = "Default phase plan"
+    plan.tags_json = "[]"
+    plan.created_at = now
+    plan.updated_at = now
+    session.add(plan)
+    session.commit()
+    context.phase_rebaseline_plan_id = ulid
+
+
+@then('the plan should be persisted successfully with phase "{phase}"')
+def step_verify_plan_persisted_phase(context: Any, phase: str) -> None:
+    """Verify the plan was persisted with the expected phase."""
+    assert context.phase_rebaseline_result == "ok", (
+        f"Expected ok, got {context.phase_rebaseline_result}"
+    )
+    session: Session = context.phase_rebaseline_session
+    plan = (
+        session.query(LifecyclePlanModel)
+        .filter_by(plan_id=context.phase_rebaseline_plan_id)
+        .one()
+    )
+    assert plan.phase == phase
+
+
+@then('the plan should be persisted successfully with state "{state}"')
+def step_verify_plan_persisted_state(context: Any, state: str) -> None:
+    """Verify the plan was persisted with the expected state."""
+    assert context.phase_rebaseline_result == "ok", (
+        f"Expected ok, got {context.phase_rebaseline_result}"
+    )
+    session: Session = context.phase_rebaseline_session
+    plan = (
+        session.query(LifecyclePlanModel)
+        .filter_by(plan_id=context.phase_rebaseline_plan_id)
+        .one()
+    )
+    assert plan.processing_state == state
+
+
+@then("the insert should fail with a constraint violation")
+def step_verify_constraint_violation(context: Any) -> None:
+    """Verify the insert failed."""
+    assert context.phase_rebaseline_result == "error", (
+        f"Expected error, got {context.phase_rebaseline_result}"
+    )
+
+
+@then('the rebaselined plan phase should be "{phase}"')
+def step_verify_rebaselined_plan_phase_default(context: Any, phase: str) -> None:
+    """Verify the plan's phase matches the expected value."""
+    session: Session = context.phase_rebaseline_session
+    plan = (
+        session.query(LifecyclePlanModel)
+        .filter_by(plan_id=context.phase_rebaseline_plan_id)
+        .one()
+    )
+    assert plan.phase == phase, f"Expected phase '{phase}', got '{plan.phase}'"

features/steps/plan_repository_steps.py

@@ -0,0 +1,497 @@
+"""Step definitions for LifecyclePlanRepository persistence coverage.
+
+Targets ``LifecyclePlanRepository`` in
+``src/cleveragents/infrastructure/database/repositories.py``.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.core.exceptions import DatabaseError
+from cleveragents.domain.models.core.action import Action, ActionState
+from cleveragents.domain.models.core.plan import (
+    AutomationLevel,
+    InvariantSource,
+    NamespacedName,
+    PlanIdentity,
+    PlanInvariant,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+    ProjectLink,
+)
+from cleveragents.domain.models.core.plan import Plan as V3Plan
+from cleveragents.infrastructure.database.models import (
+    Base,
+    PlanArgumentModel,
+    PlanInvariantModel,
+    PlanProjectModel,
+)
+from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
+    DuplicatePlanError,
+    LifecyclePlanRepository,
+)
+
+# Crockford base32 alphabet for ULIDs
+_CB32 = "0123456789ABCDEFGHJKMNPQRSTVWXYZ"
+_PLAN_REPO_ULID_COUNTER = 100
+
+
+def _next_plan_repo_ulid() -> str:
+    """Return a unique, valid ULID string for each call."""
+    global _PLAN_REPO_ULID_COUNTER
+    _PLAN_REPO_ULID_COUNTER += 1
+    n = _PLAN_REPO_ULID_COUNTER
+    suffix = ""
+    for _ in range(8):
+        suffix = _CB32[n % 32] + suffix
+        n //= 32
+    return f"01HGZ6FE0AQDYTR4BX{suffix}"
+
+
+def _ensure_action(ctx: Context, action_name: str = "local/test-action") -> None:
+    """Ensure the test action exists in the database."""
+    try:
+        actions_created = ctx._plan_repo_actions_created  # type: ignore[attr-defined]
+    except (AttributeError, KeyError):
+        actions_created: set[str] = set()
+        ctx._plan_repo_actions_created = actions_created  # type: ignore[attr-defined]
+    if action_name in actions_created:
+        return
+    parts = action_name.split("/", 1)
+    namespace = parts[0] if len(parts) == 2 else "local"
+    short_name = parts[1] if len(parts) == 2 else parts[0]
+    action = Action(
+        namespaced_name=NamespacedName(namespace=namespace, name=short_name),
+        description=f"Test action {short_name}",
+        long_description=None,
+        definition_of_done=f"Verify {short_name} completes",
+        strategy_actor="local/strategist",
+        execution_actor="local/executor",
+        estimation_actor=None,
+        review_actor=None,
+        arguments=[],
+        reusable=True,
+        read_only=False,
+        state=ActionState("available"),
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        created_by=None,
+        tags=[],
+    )
+    action_repo = ActionRepository(session_factory=ctx.db_session_factory)
+    action_repo.create(action)
+    ctx.db_session.commit()
+    ctx._plan_repo_actions_created.add(action_name)
+
+
+def _make_plan(
+    plan_id: str,
+    phase: str = "action",
+    processing_state: str = "queued",
+    action_name: str = "local/test-action",
+    description: str = "Test plan description",
+    project_links: list[ProjectLink] | None = None,
+    invariants: list[PlanInvariant] | None = None,
+    arguments: dict[str, Any] | None = None,
+    arguments_order: list[str] | None = None,
+) -> V3Plan:
+    """Create a minimal valid v3 Plan domain object."""
+    now = datetime.now()
+    return V3Plan(
+        identity=PlanIdentity(plan_id=plan_id, attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="test-plan"),
+        action_name=action_name,
+        description=description,
+        definition_of_done="All tests pass",
+        phase=PlanPhase(phase),
+        processing_state=ProcessingState(processing_state),
+        automation_level=AutomationLevel.MANUAL,
+        strategy_actor="local/strategist",
+        execution_actor="local/executor",
+        project_links=project_links or [],
+        invariants=invariants or [],
+        arguments=arguments or {},
+        arguments_order=arguments_order or list((arguments or {}).keys()),
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        created_by="test-user",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a clean in-memory database for plan repository tests")
+def step_clean_db_plan_repo(context: Context) -> None:
+    """Create a fresh in-memory SQLite database with all tables."""
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    context.db_engine = engine
+    session = sessionmaker(bind=engine)()
+    context.db_session = session
+    context.db_session_factory = lambda: session
+
+
+@given("a lifecycle plan repository backed by the database")
+def step_plan_repo(context: Context) -> None:
+    """Instantiate a LifecyclePlanRepository using the session factory."""
+    context.plan_repo = LifecyclePlanRepository(
+        session_factory=context.db_session_factory,
+    )
+    context.plan_error = None
+    context.plan_result = None
+    context.retrieved_plan = None
+
+
+# ---------------------------------------------------------------------------
+# Create plans
+# ---------------------------------------------------------------------------
+
+
+@given('a valid plan domain object with id "{plan_id}"')
+def step_make_plan(context: Context, plan_id: str) -> None:
+    """Build a Plan domain object with the given ULID."""
+    _ensure_action(context)
+    context.plan = _make_plan(plan_id=plan_id)
+
+
+@given("the plan is saved through the plan repository")
+def step_save_plan_given(context: Context) -> None:
+    """Persist the plan (as a Given step)."""
+    try:
+        context.plan_repo.create(context.plan)
+        context.db_session.commit()
+    except Exception as exc:
+        context.plan_error = exc
+
+
+@when("the plan is saved through the plan repository")
+def step_save_plan(context: Context) -> None:
+    """Persist the plan and capture any errors."""
+    try:
+        context.plan_repo.create(context.plan)
+        context.db_session.commit()
+    except Exception as exc:
+        context.plan_error = exc
+
+
+@when('a duplicate plan with id "{plan_id}" is saved')
+def step_save_duplicate_plan(context: Context, plan_id: str) -> None:
+    """Attempt to persist a second plan with the same ID."""
+    dup = _make_plan(plan_id=plan_id, description="Duplicate plan")
+    try:
+        context.plan_repo.create(dup)
+        context.db_session.commit()
+    except (DuplicatePlanError, DatabaseError) as exc:
+        context.plan_error = exc
+    except Exception as exc:
+        cause = exc.__cause__ or exc
+        context.plan_error = cause
+
+
+# ---------------------------------------------------------------------------
+# Retrieve plans
+# ---------------------------------------------------------------------------
+
+
+@then('the plan retrieved by id "{plan_id}" should exist')
+def step_get_plan_by_id_exists(context: Context, plan_id: str) -> None:
+    context.retrieved_plan = context.plan_repo.get(plan_id)
+    assert context.retrieved_plan is not None, f"Plan {plan_id} not found"
+
+
+@then('the plan retrieved by id "{plan_id}" should not exist')
+def step_get_plan_by_id_not_exists(context: Context, plan_id: str) -> None:
+    result = context.plan_repo.get(plan_id)
+    assert result is None, f"Plan {plan_id} should not exist but was found"
+
+
+@then('the retrieved plan description should be "{desc}"')
+def step_check_plan_description(context: Context, desc: str) -> None:
+    assert context.retrieved_plan.description == desc, (
+        f"Expected '{desc}', got '{context.retrieved_plan.description}'"
+    )
+
+
+@then('the plan retrieved by name "{name}" should exist')
+def step_get_plan_by_name(context: Context, name: str) -> None:
+    result = context.plan_repo.get_by_name(name)
+    assert result is not None, f"Plan with name {name} not found"
+
+
+@when('the plan is retrieved by id "{plan_id}"')
+def step_when_get_plan_by_id(context: Context, plan_id: str) -> None:
+    context.plan_result = context.plan_repo.get(plan_id)
+
+
+@then("the plan retrieve result should be None")
+def step_check_plan_none(context: Context) -> None:
+    assert context.plan_result is None, "Expected None but got a plan"
+
+
+# ---------------------------------------------------------------------------
+# Update plans
+# ---------------------------------------------------------------------------
+
+
+@when('the plan phase is updated to "{phase}" with state "{state}"')
+def step_update_plan(context: Context, phase: str, state: str) -> None:
+    plan = context.plan_repo.get(context.plan.identity.plan_id)
+    plan.phase = PlanPhase(phase)
+    plan.processing_state = ProcessingState(state)
+    plan.timestamps.updated_at = datetime.now()
+    context.plan_repo.update(plan)
+    context.db_session.commit()
+    context.retrieved_plan = context.plan_repo.get(context.plan.identity.plan_id)
+
+
+@then('the retrieved plan should have phase "{phase}"')
+def step_check_plan_phase(context: Context, phase: str) -> None:
+    # If retrieved_plan is not set yet, fetch it from the repo
+    if context.retrieved_plan is None:
+        context.retrieved_plan = context.plan_repo.get(context.plan.identity.plan_id)
+    actual = (
+        context.retrieved_plan.phase.value
+        if hasattr(context.retrieved_plan.phase, "value")
+        else context.retrieved_plan.phase
+    )
+    assert actual == phase, f"Expected phase '{phase}', got '{actual}'"
+
+
+@then('the retrieved plan should have processing state "{state}"')
+def step_check_plan_state(context: Context, state: str) -> None:
+    # If retrieved_plan is not set yet, fetch it from the repo
+    if context.retrieved_plan is None:
+        context.retrieved_plan = context.plan_repo.get(context.plan.identity.plan_id)
+    actual = (
+        context.retrieved_plan.processing_state.value
+        if hasattr(context.retrieved_plan.processing_state, "value")
+        else context.retrieved_plan.processing_state
+    )
+    assert actual == state, f"Expected state '{state}', got '{actual}'"
+
+
+# ---------------------------------------------------------------------------
+# List filters
+# ---------------------------------------------------------------------------
+
+
+@given('{n:d} plans exist in phase "{phase1}" and {m:d} plans in phase "{phase2}"')
+def step_create_plans_by_phase(
+    context: Context, n: int, phase1: str, m: int, phase2: str
+) -> None:
+    _ensure_action(context)
+    for _ in range(n):
+        plan = _make_plan(plan_id=_next_plan_repo_ulid(), phase=phase1)
+        context.plan_repo.create(plan)
+    for _ in range(m):
+        plan = _make_plan(plan_id=_next_plan_repo_ulid(), phase=phase2)
+        context.plan_repo.create(plan)
+    context.db_session.commit()
+
+
+@given('{n:d} plans exist in state "{state1}" and {m:d} plan in state "{state2}"')
+def step_create_plans_by_state(
+    context: Context, n: int, state1: str, m: int, state2: str
+) -> None:
+    _ensure_action(context)
+    for _ in range(n):
+        plan = _make_plan(plan_id=_next_plan_repo_ulid(), processing_state=state1)
+        context.plan_repo.create(plan)
+    for _ in range(m):
+        plan = _make_plan(plan_id=_next_plan_repo_ulid(), processing_state=state2)
+        context.plan_repo.create(plan)
+    context.db_session.commit()
+
+
+@given(
+    '{n:d} plans referencing action "{action1}" and {m:d} plan referencing "{action2}"'
+)
+def step_create_plans_by_action(
+    context: Context, n: int, action1: str, m: int, action2: str
+) -> None:
+    _ensure_action(context, action1)
+    _ensure_action(context, action2)
+    for _ in range(n):
+        plan = _make_plan(plan_id=_next_plan_repo_ulid(), action_name=action1)
+        context.plan_repo.create(plan)
+    for _ in range(m):
+        plan = _make_plan(plan_id=_next_plan_repo_ulid(), action_name=action2)
+        context.plan_repo.create(plan)
+    context.db_session.commit()
+
+
+@given('a plan linked to project "{project}" and another with no project link')
+def step_create_plans_with_project(context: Context, project: str) -> None:
+    _ensure_action(context)
+    plan_with = _make_plan(
+        plan_id=_next_plan_repo_ulid(),
+        project_links=[ProjectLink(project_name=project)],
+    )
+    context.plan_repo.create(plan_with)
+    plan_without = _make_plan(plan_id=_next_plan_repo_ulid())
+    context.plan_repo.create(plan_without)
+    context.db_session.commit()
+
+
+@when('plans are listed with phase filter "{phase}"')
+def step_list_by_phase(context: Context, phase: str) -> None:
+    context.plan_list_result = context.plan_repo.list_plans(phase=phase)
+
+
+@when('plans are listed with processing state filter "{state}"')
+def step_list_by_state(context: Context, state: str) -> None:
+    context.plan_list_result = context.plan_repo.list_plans(processing_state=state)
+
+
+@when('plans are listed with action name filter "{action}"')
+def step_list_by_action(context: Context, action: str) -> None:
+    context.plan_list_result = context.plan_repo.list_plans(action_name=action)
+
+
+@when('plans are listed with project name filter "{project}"')
+def step_list_by_project(context: Context, project: str) -> None:
+    context.plan_list_result = context.plan_repo.list_plans(project_name=project)
+
+
+@then("the plan list should contain {n:d} plans")
+def step_check_plan_list_count_plural(context: Context, n: int) -> None:
+    actual = len(context.plan_list_result)
+    assert actual == n, f"Expected {n} plans, got {actual}"
+
+
+@then("the plan list should contain {n:d} plan")
+def step_check_plan_list_count_singular(context: Context, n: int) -> None:
+    actual = len(context.plan_list_result)
+    assert actual == n, f"Expected {n} plan(s), got {actual}"
+
+
+# ---------------------------------------------------------------------------
+# Delete
+# ---------------------------------------------------------------------------
+
+
+@given('a valid plan domain object with id "{plan_id}" and child rows')
+def step_make_plan_with_children(context: Context, plan_id: str) -> None:
+    _ensure_action(context)
+    context.plan = _make_plan(
+        plan_id=plan_id,
+        project_links=[ProjectLink(project_name="local/child-proj")],
+        invariants=[PlanInvariant(text="Must pass tests", source=InvariantSource.PLAN)],
+        arguments={"key": "value"},
+        arguments_order=["key"],
+    )
+
+
+@when('the plan with id "{plan_id}" is deleted')
+def step_delete_plan(context: Context, plan_id: str) -> None:
+    context.plan_repo.delete(plan_id)
+    context.db_session.commit()
+
+
+@then('the child rows for plan "{plan_id}" should be gone')
+def step_check_children_gone(context: Context, plan_id: str) -> None:
+    session: Session = context.db_session
+    projects = session.query(PlanProjectModel).filter_by(plan_id=plan_id).count()
+    arguments = session.query(PlanArgumentModel).filter_by(plan_id=plan_id).count()
+    invariants = session.query(PlanInvariantModel).filter_by(plan_id=plan_id).count()
+    assert projects == 0, f"Expected 0 project links, got {projects}"
+    assert arguments == 0, f"Expected 0 arguments, got {arguments}"
+    assert invariants == 0, f"Expected 0 invariants, got {invariants}"
+
+
+# ---------------------------------------------------------------------------
+# Error handling
+# ---------------------------------------------------------------------------
+
+
+@then("the plan repository should not raise an error")
+def step_no_plan_error(context: Context) -> None:
+    assert context.plan_error is None, f"Unexpected error: {context.plan_error}"
+
+
+@then('a DuplicatePlanError should be raised mentioning "{plan_id}"')
+def step_check_duplicate_plan_error(context: Context, plan_id: str) -> None:
+    assert context.plan_error is not None, "Expected DuplicatePlanError but no error"
+    assert isinstance(context.plan_error, DuplicatePlanError), (
+        f"Expected DuplicatePlanError, got {type(context.plan_error).__name__}"
+    )
+    assert plan_id in str(context.plan_error), (
+        f"Error message should mention '{plan_id}': {context.plan_error}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Phase and state persistence
+# ---------------------------------------------------------------------------
+
+
+@given(
+    'a plan domain object in phase "{phase}" with state "{state}" and id "{plan_id}"'
+)
+def step_make_plan_phase_state(
+    context: Context, phase: str, state: str, plan_id: str
+) -> None:
+    _ensure_action(context)
+    context.plan = _make_plan(plan_id=plan_id, phase=phase, processing_state=state)
+
+
+# ---------------------------------------------------------------------------
+# Child row ordering
+# ---------------------------------------------------------------------------
+
+
+@given('a plan with ordered children and id "{plan_id}"')
+def step_make_plan_ordered_children(context: Context, plan_id: str) -> None:
+    _ensure_action(context)
+    context.plan = _make_plan(
+        plan_id=plan_id,
+        project_links=[
+            ProjectLink(project_name="local/proj-a"),
+            ProjectLink(project_name="local/proj-b"),
+        ],
+        invariants=[
+            PlanInvariant(text="first", source=InvariantSource.PLAN),
+            PlanInvariant(text="second", source=InvariantSource.ACTION),
+            PlanInvariant(text="third", source=InvariantSource.PROJECT),
+        ],
+        arguments={"alpha": "a", "beta": "b", "gamma": "g"},
+        arguments_order=["alpha", "beta", "gamma"],
+    )
+
+
+@then('the retrieved plan should have arguments in order "{expected}"')
+def step_check_arguments_order(context: Context, expected: str) -> None:
+    plan = context.plan_repo.get(context.plan.identity.plan_id)
+    expected_list = expected.split(",")
+    assert plan.arguments_order == expected_list, (
+        f"Expected arguments order {expected_list}, got {plan.arguments_order}"
+    )
+
+
+@then('the retrieved plan should have invariants in order "{expected}"')
+def step_check_invariants_order(context: Context, expected: str) -> None:
+    plan = context.plan_repo.get(context.plan.identity.plan_id)
+    expected_list = expected.split(",")
+    actual = [inv.text for inv in plan.invariants]
+    assert actual == expected_list, f"Expected invariants {expected_list}, got {actual}"
+
+
+@then('the retrieved plan should have project links including "{project}"')
+def step_check_project_links(context: Context, project: str) -> None:
+    plan = context.plan_repo.get(context.plan.identity.plan_id)
+    names = [pl.project_name for pl in plan.project_links]
+    assert project in names, f"Expected project '{project}' in links, got {names}"

features/steps/uow_lifecycle_steps.py

@@ -0,0 +1,353 @@
+"""Step definitions for UnitOfWork lifecycle repository wiring.
+
+Targets ``UnitOfWork`` and ``UnitOfWorkContext`` in
+``src/cleveragents/infrastructure/database/unit_of_work.py``.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any
+
+from behave import given, then, when
+from behave.runner import Context
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.domain.models.core.action import Action, ActionState
+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,
+    PlanRepository,
+)
+from cleveragents.infrastructure.database.unit_of_work import (
+    UnitOfWorkContext,
+)
+
+
+def _make_uow_action(name: str = "local/uow-action") -> Action:
+    """Create a minimal Action domain object for testing."""
+    parts = name.split("/", 1)
+    namespace = parts[0] if len(parts) == 2 else "local"
+    short_name = parts[1] if len(parts) == 2 else parts[0]
+    now = datetime.now()
+    return Action(
+        namespaced_name=NamespacedName(namespace=namespace, name=short_name),
+        description=f"UoW test action {short_name}",
+        long_description=None,
+        definition_of_done=f"Verify {short_name} completes",
+        strategy_actor="local/strategist",
+        execution_actor="local/executor",
+        estimation_actor=None,
+        review_actor=None,
+        arguments=[],
+        reusable=True,
+        read_only=False,
+        state=ActionState("available"),
+        created_at=now,
+        updated_at=now,
+        created_by=None,
+        tags=[],
+    )
+
+
+def _make_uow_plan(
+    plan_id: str,
+    action_name: str = "local/uow-action",
+) -> V3Plan:
+    """Create a minimal Plan domain object for testing."""
+    now = datetime.now()
+    return V3Plan(
+        identity=PlanIdentity(plan_id=plan_id, attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="uow-plan"),
+        action_name=action_name,
+        description="UoW test plan",
+        definition_of_done="Verify UoW plan completes",
+        strategy_actor="local/strategist",
+        execution_actor="local/executor",
+        phase=PlanPhase("action"),
+        processing_state=ProcessingState("queued"),
+        automation_level=AutomationLevel("manual"),
+        timestamps=PlanTimestamps(
+            created_at=now,
+            updated_at=now,
+        ),
+        project_links=[],
+        arguments={},
+        arguments_order=[],
+        invariants=[],
+        reusable=True,
+        read_only=False,
+        created_by=None,
+        tags=[],
+    )
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a UnitOfWork backed by an in-memory database for lifecycle tests")
+def step_uow_lifecycle_background(context: Context) -> None:
+    """Set up an in-memory database with schema for UoW tests.
+
+    We bypass the migration runner and use ``Base.metadata.create_all``
+    directly since this is a unit test with an in-memory DB.
+    """
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        future=True,
+        connect_args={"check_same_thread": False},
+    )
+
+    # Enable FK enforcement for SQLite
+    @event.listens_for(engine, "connect")
+    def _set_sqlite_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,
+    )
+
+    # Store on context for reuse
+    context.uow_engine = engine
+    context.uow_session_factory = sf
+
+
+# ---------------------------------------------------------------------------
+# Repository exposure
+# ---------------------------------------------------------------------------
+
+
+@when("I access the actions repository from the UoW context")
+def step_access_actions_repo(context: Context) -> None:
+    session = context.uow_session_factory()
+    ctx = UnitOfWorkContext(session)
+    context.uow_ctx_result = ctx.actions
+    session.close()
+
+
+@then("the actions repository should be an ActionRepository instance")
+def step_check_actions_repo_type(context: Context) -> None:
+    assert isinstance(context.uow_ctx_result, ActionRepository), (
+        f"Expected ActionRepository, got {type(context.uow_ctx_result)}"
+    )
+
+
+@when("I access the lifecycle_plans repository from the UoW context")
+def step_access_lifecycle_plans_repo(context: Context) -> None:
+    session = context.uow_session_factory()
+    ctx = UnitOfWorkContext(session)
+    context.uow_ctx_result = ctx.lifecycle_plans
+    session.close()
+
+
+@then("the lifecycle_plans repository should be a LifecyclePlanRepository instance")
+def step_check_lifecycle_plans_repo_type(context: Context) -> None:
+    assert isinstance(context.uow_ctx_result, LifecyclePlanRepository), (
+        f"Expected LifecyclePlanRepository, got {type(context.uow_ctx_result)}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# CRUD via UoW
+# ---------------------------------------------------------------------------
+
+
+@when('I create an action "{name}" via the UoW transaction')
+def step_create_action_via_uow(context: Context, name: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        action = _make_uow_action(name)
+        ctx.actions.create(action)
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+
+@then('the action "{name}" should be retrievable from the actions repository')
+def step_check_action_retrievable(context: Context, name: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        action = ctx.actions.get_by_name(name)
+        assert action is not None, f"Action {name} not found"
+    finally:
+        session.close()
+
+
+@given('an action "{name}" exists via the UoW')
+def step_ensure_action_exists(context: Context, name: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        existing = ctx.actions.get_by_name(name)
+        if existing is None:
+            action = _make_uow_action(name)
+            ctx.actions.create(action)
+            session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+
+@when('I create a lifecycle plan "{plan_id}" via the UoW transaction')
+def step_create_plan_via_uow(context: Context, plan_id: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        plan = _make_uow_plan(plan_id, action_name="local/uow-plan-action")
+        ctx.lifecycle_plans.create(plan)
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+
+@then('the plan "{plan_id}" should be retrievable from the lifecycle_plans repository')
+def step_check_plan_retrievable(context: Context, plan_id: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        plan = ctx.lifecycle_plans.get(plan_id)
+        assert plan is not None, f"Plan {plan_id} not found"
+    finally:
+        session.close()
+
+
+# ---------------------------------------------------------------------------
+# Cross-repository commit
+# ---------------------------------------------------------------------------
+
+
+@when(
+    'I create an action "{action_name}" and a plan "{plan_id}" in a single UoW transaction'
+)
+def step_create_action_and_plan(
+    context: Context, action_name: str, plan_id: str
+) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        action = _make_uow_action(action_name)
+        ctx.actions.create(action)
+        plan = _make_uow_plan(plan_id, action_name=action_name)
+        ctx.lifecycle_plans.create(plan)
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+
+@then('the action "{name}" should be retrievable in a new transaction')
+def step_check_action_new_txn(context: Context, name: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        action = ctx.actions.get_by_name(name)
+        assert action is not None, f"Action {name} not found in new transaction"
+    finally:
+        session.close()
+
+
+@then('the plan "{plan_id}" should be retrievable in a new transaction')
+def step_check_plan_new_txn(context: Context, plan_id: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        plan = ctx.lifecycle_plans.get(plan_id)
+        assert plan is not None, f"Plan {plan_id} not found in new transaction"
+    finally:
+        session.close()
+
+
+# ---------------------------------------------------------------------------
+# Rollback
+# ---------------------------------------------------------------------------
+
+
+@when('I create an action "{action_name}" and a plan "{plan_id}" but force a rollback')
+def step_create_and_rollback(context: Context, action_name: str, plan_id: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        action = _make_uow_action(action_name)
+        ctx.actions.create(action)
+        plan = _make_uow_plan(plan_id, action_name=action_name)
+        ctx.lifecycle_plans.create(plan)
+        # Force rollback instead of commit
+        session.rollback()
+    finally:
+        session.close()
+
+
+@then('the action "{name}" should not exist in a new transaction')
+def step_check_action_not_exists(context: Context, name: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        action = ctx.actions.get_by_name(name)
+        assert action is None, f"Action {name} should not exist after rollback"
+    finally:
+        session.close()
+
+
+@then('the plan "{plan_id}" should not exist in a new transaction')
+def step_check_plan_not_exists(context: Context, plan_id: str) -> None:
+    session = context.uow_session_factory()
+    try:
+        ctx = UnitOfWorkContext(session)
+        plan = ctx.lifecycle_plans.get(plan_id)
+        assert plan is None, f"Plan {plan_id} should not exist after rollback"
+    finally:
+        session.close()
+
+
+# ---------------------------------------------------------------------------
+# Legacy accessor
+# ---------------------------------------------------------------------------
+
+
+@when("I access the legacy plans repository from the UoW context")
+def step_access_legacy_plans_repo(context: Context) -> None:
+    session = context.uow_session_factory()
+    ctx = UnitOfWorkContext(session)
+    context.uow_ctx_result = ctx.plans
+    session.close()
+
+
+@then("the legacy plans repository should be a PlanRepository instance")
+def step_check_legacy_plans_repo_type(context: Context) -> None:
+    assert isinstance(context.uow_ctx_result, PlanRepository), (
+        f"Expected PlanRepository, got {type(context.uow_ctx_result)}"
+    )

features/uow_lifecycle.feature

@@ -0,0 +1,66 @@
+@phase1 @domain @unit_of_work @uow_lifecycle
+Feature: UnitOfWork lifecycle repository wiring
+  As a system operator managing v3 lifecycle entities
+  I want the UnitOfWork to expose actions and lifecycle_plans repositories
+  So that transactional persistence works across both repositories atomically
+
+  Background:
+    Given a UnitOfWork backed by an in-memory database for lifecycle tests
+
+  # ---------------------------------------------------------------------------
+  # Repository exposure
+  # ---------------------------------------------------------------------------
+
+  @uow_actions_repo
+  Scenario: UoW exposes actions repository
+    When I access the actions repository from the UoW context
+    Then the actions repository should be an ActionRepository instance
+
+  @uow_lifecycle_plans_repo
+  Scenario: UoW exposes lifecycle_plans repository
+    When I access the lifecycle_plans repository from the UoW context
+    Then the lifecycle_plans repository should be a LifecyclePlanRepository instance
+
+  # ---------------------------------------------------------------------------
+  # CRUD via UoW
+  # ---------------------------------------------------------------------------
+
+  @uow_create_action
+  Scenario: Create action via UoW and retrieve it
+    When I create an action "local/uow-action" via the UoW transaction
+    Then the action "local/uow-action" should be retrievable from the actions repository
+
+  @uow_create_plan
+  Scenario: Create plan via UoW and retrieve it
+    Given an action "local/uow-plan-action" exists via the UoW
+    When I create a lifecycle plan "01HGZ6FE0AQDYTR4BX00000E01" via the UoW transaction
+    Then the plan "01HGZ6FE0AQDYTR4BX00000E01" should be retrievable from the lifecycle_plans repository
+
+  # ---------------------------------------------------------------------------
+  # Cross-repository commit
+  # ---------------------------------------------------------------------------
+
+  @uow_commit_persists
+  Scenario: UoW commit persists changes across repositories
+    When I create an action "local/commit-action" and a plan "01HGZ6FE0AQDYTR4BX00000E02" in a single UoW transaction
+    Then the action "local/commit-action" should be retrievable in a new transaction
+    And the plan "01HGZ6FE0AQDYTR4BX00000E02" should be retrievable in a new transaction
+
+  # ---------------------------------------------------------------------------
+  # Rollback
+  # ---------------------------------------------------------------------------
+
+  @uow_rollback
+  Scenario: UoW rollback discards changes across repositories
+    When I create an action "local/rollback-action" and a plan "01HGZ6FE0AQDYTR4BX00000E03" but force a rollback
+    Then the action "local/rollback-action" should not exist in a new transaction
+    And the plan "01HGZ6FE0AQDYTR4BX00000E03" should not exist in a new transaction
+
+  # ---------------------------------------------------------------------------
+  # Legacy accessor still works
+  # ---------------------------------------------------------------------------
+
+  @uow_legacy_plans
+  Scenario: Legacy plans accessor still returns PlanRepository
+    When I access the legacy plans repository from the UoW context
+    Then the legacy plans repository should be a PlanRepository instance

implementation_plan.md

@@ -1634,65 +1634,65 @@ No standalone Q0-Advanced commits planned. Advanced QA enhancements are bundled
   - [X] Git [Jeff]: `git checkout master`
   - [X] Git [Jeff]: `git branch -d feature/m1-db-actions`
   - [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report` -- TOTAL 97%.
-- [ ] **COMMIT (Owner: Jeff | Group: A5.beta | Branch: feature/m1-db-plan-phase-rebaseline | Planned: Day 7 | Expected: Day 9) - Commit message: "feat(db): rebaseline plan phase/state enums for Action and Apply terminal states"**
-  - [ ] Git [Jeff]: `git checkout master`
-  - [ ] Git [Jeff]: `git pull origin master`
-  - [ ] Git [Jeff]: `git checkout -b feature/m1-db-plan-phase-rebaseline`
-  - [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
-  - [ ] Code [Jeff]: Add Alembic migration that updates `v3_plans` phase/state constraints to include `action` phase and Apply terminal outcomes (`applied`, `constrained`, `errored`, `cancelled`), and removes `applied` as a phase.
-  - [ ] Code [Jeff]: Rebuild `v3_plans` table safely for SQLite (create temp table → copy → drop → rename) to update CHECK constraints and defaults without data loss.
-  - [ ] Code [Jeff]: Update ORM `LifecyclePlanModel` to accept Action phase + Apply terminal states; ensure serialization/deserialization maps updated enums.
-  - [ ] Code [Jeff]: Mark `automation_level` column as legacy (no new writes); keep nullable for backward compatibility until removal in A6.legacy cleanup.
-  - [ ] Docs [Jeff]: Update `docs/reference/database_schema.md` with the new phase/state constraints and legacy automation_level note.
-  - [ ] Tests (Behave) [Jeff]: Add migration scenarios asserting phase/state constraints accept Action + Apply terminal values and reject legacy `applied` phase.
-  - [ ] Tests (Robot) [Jeff]: Add Robot migration smoke test that inserts a plan row with `phase=action` and `processing_state=queued`.
-  - [ ] Tests (ASV) [Jeff]: Add `benchmarks/plan_phase_migration_bench.py` for migration runtime baseline.
-  - [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
-  - [ ] Git [Jeff]: `git add .`
-  - [ ] Git [Jeff]: `git commit -m "feat(db): rebaseline plan phase/state enums for Action and Apply terminal states"`
-  - [ ] Forgejo PR [Jeff]: Open PR from `feature/m1-db-plan-phase-rebaseline` to `master` with description "Rebaseline plan phase/state constraints for Action phase and Apply terminal outcomes, with migration + tests.".
-  - [ ] Git [Jeff]: `git checkout master`
-  - [ ] Git [Jeff]: `git branch -d feature/m1-db-plan-phase-rebaseline`
-  - [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
-- [ ] **COMMIT (Owner: Jeff | Group: A5.delta | Branch: feature/m1-plan-repo | Planned: Day 8 | Expected: Day 10) - Commit message: "feat(repo): add lifecycle plan repository"**
-  - [ ] Git [Jeff]: `git checkout master`
-  - [ ] Git [Jeff]: `git pull origin master`
-  - [ ] Git [Jeff]: `git checkout -b feature/m1-plan-repo`
-  - [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
-  - [ ] Code [Jeff]: Implement `LifecyclePlanRepository` backed by `LifecyclePlanModel` (create/get/update/delete) with ordered child persistence for project links, arguments, and invariants.
-  - [ ] Code [Jeff]: Persist Action phase + Apply terminal states (`applied`, `constrained`, `errored`, `cancelled`) in phase/state filters and serialization.
-  - [ ] Code [Jeff]: Add list filters for phase, processing_state, action_name, project_name, and namespace (match CLI flags); ensure deterministic ordering.
-  - [ ] Code [Jeff]: Raise explicit errors for duplicate plan_id, missing plan, and invalid phase/state updates.
-  - [ ] Docs [Jeff]: Document plan repository contract + filters in `docs/reference/repositories.md`.
-  - [ ] Tests (Behave) [Jeff]: Add scenarios for create/get/list/update, filter behavior, and child row ordering.
-  - [ ] Tests (Robot) [Jeff]: Add Robot smoke test for repository CRUD via service layer.
-  - [ ] Tests (ASV) [Jeff]: Add `benchmarks/plan_repository_bench.py` for list/query performance.
-  - [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
-  - [ ] Git [Jeff]: `git add .`
-  - [ ] Git [Jeff]: `git commit -m "feat(repo): add lifecycle plan repository"`
+- [X] **COMMIT (Owner: Jeff | Group: A5.beta | Branch: feature/m1-db-plan-phase-rebaseline | Planned: Day 7 | Done: Day 7, February 15, 2026) - Commit message: "feat(db): rebaseline plan phase/state enums for Action and Apply terminal states"**
+  - [X] Git [Jeff]: `git checkout master`
+  - [X] Git [Jeff]: `git pull origin master`
+  - [X] Git [Jeff]: `git checkout -b feature/m1-db-plan-phase-rebaseline`
+  - [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
+  - [X] Code [Jeff]: Add Alembic migration that updates `v3_plans` phase/state constraints to include `action` phase and Apply terminal outcomes (`applied`, `constrained`, `errored`, `cancelled`), and removes `applied` as a phase.
+  - [X] Code [Jeff]: Rebuild `v3_plans` table safely for SQLite (create temp table → copy → drop → rename) to update CHECK constraints and defaults without data loss.
+  - [X] Code [Jeff]: Update ORM `LifecyclePlanModel` to accept Action phase + Apply terminal states; ensure serialization/deserialization maps updated enums.
+  - [X] Code [Jeff]: Mark `automation_level` column as legacy (no new writes); keep nullable for backward compatibility until removal in A6.legacy cleanup.
+  - [X] Docs [Jeff]: Update `docs/reference/database_schema.md` with the new phase/state constraints and legacy automation_level note.
+  - [X] Tests (Behave) [Jeff]: Add migration scenarios asserting phase/state constraints accept Action + Apply terminal values and reject legacy `applied` phase.
+  - [X] Tests (Robot) [Jeff]: Add Robot migration smoke test that inserts a plan row with `phase=action` and `processing_state=queued`.
+  - [X] Tests (ASV) [Jeff]: Add `benchmarks/plan_phase_migration_bench.py` for migration runtime baseline.
+  - [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
+  - [X] Git [Jeff]: `git add .`
+  - [X] Git [Jeff]: `git commit -m "feat(db): rebaseline plan phase/state enums for Action and Apply terminal states"`
+  - [X] Forgejo PR [Jeff]: Open PR from `feature/m1-db-plan-phase-rebaseline` to `master` with description "Rebaseline plan phase/state constraints for Action phase and Apply terminal outcomes, with migration + tests.".
+  - [X] Git [Jeff]: `git checkout master`
+  - [X] Git [Jeff]: `git branch -d feature/m1-db-plan-phase-rebaseline`
+  - [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%. Only mark this as complete once coverage is >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
+- [X] **COMMIT (Owner: Jeff | Group: A5.delta | Branch: feature/m1-plan-repo | Planned: Day 8 | Expected: Day 10) - Commit message: "feat(repo): add lifecycle plan repository"** Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git checkout master` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git pull origin master` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git checkout -b feature/m1-plan-repo` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit) Done: Day 7, February 15, 2026
+  - [X] Code [Jeff]: Implement `LifecyclePlanRepository` backed by `LifecyclePlanModel` (create/get/update/delete) with ordered child persistence for project links, arguments, and invariants. Done: Day 7, February 15, 2026
+  - [X] Code [Jeff]: Persist Action phase + Apply terminal states (`applied`, `constrained`, `errored`, `cancelled`) in phase/state filters and serialization. Done: Day 7, February 15, 2026
+  - [X] Code [Jeff]: Add list filters for phase, processing_state, action_name, project_name, and namespace (match CLI flags); ensure deterministic ordering. Done: Day 7, February 15, 2026
+  - [X] Code [Jeff]: Raise explicit errors for duplicate plan_id, missing plan, and invalid phase/state updates. Done: Day 7, February 15, 2026
+  - [X] Docs [Jeff]: Document plan repository contract + filters in `docs/reference/repositories.md`. Done: Day 7, February 15, 2026
+  - [X] Tests (Behave) [Jeff]: Add scenarios for create/get/list/update, filter behavior, and child row ordering. Done: Day 7, February 15, 2026
+  - [X] Tests (Robot) [Jeff]: Add Robot smoke test for repository CRUD via service layer. Done: Day 7, February 15, 2026
+  - [X] Tests (ASV) [Jeff]: Add `benchmarks/plan_repository_bench.py` for list/query performance. Done: Day 7, February 15, 2026
+  - [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes. Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git add .` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git commit -m "feat(repo): add lifecycle plan repository"` Done: Day 7, February 15, 2026
   - [ ] Forgejo PR [Jeff]: Open PR from `feature/m1-plan-repo` to `master` with description "Add lifecycle plan repository with filters and ordered child persistence.".
   - [ ] Git [Jeff]: `git checkout master`
   - [ ] Git [Jeff]: `git branch -d feature/m1-plan-repo`
-  - [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
+  - [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%. Done: Day 7, February 15, 2026
 
-- [ ] **COMMIT (Owner: Jeff | Group: A5.epsilon | Branch: feature/m1-uow-lifecycle | Planned: Day 9 | Expected: Day 11) - Commit message: "feat(uow): wire lifecycle repositories"**
-  - [ ] Git [Jeff]: `git checkout master`
-  - [ ] Git [Jeff]: `git pull origin master`
-  - [ ] Git [Jeff]: `git checkout -b feature/m1-uow-lifecycle`
-  - [ ] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit)
-  - [ ] Code [Jeff]: Update `UnitOfWork` to expose `actions` + `lifecycle_plans` repositories and remove legacy plan repository accessors.
-  - [ ] Code [Jeff]: Update `UnitOfWorkContext` to include new repos and remove legacy plan/change repositories from default access.
-  - [ ] Docs [Jeff]: Update `docs/reference/repositories.md` with UoW lifecycle usage patterns.
-  - [ ] Tests (Behave) [Jeff]: Add scenarios verifying UoW exposes action + lifecycle plan repositories.
-  - [ ] Tests (Robot) [Jeff]: Add Robot smoke test that creates an action + plan via UoW.
-  - [ ] Tests (ASV) [Jeff]: Add `benchmarks/uow_lifecycle_bench.py` for UoW overhead baseline.
-  - [ ] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes.
-  - [ ] Git [Jeff]: `git add .`
-  - [ ] Git [Jeff]: `git commit -m "feat(uow): wire lifecycle repositories"`
+- [X] **COMMIT (Owner: Jeff | Group: A5.epsilon | Branch: feature/m1-uow-lifecycle | Planned: Day 9 | Done: Day 7, February 15, 2026) - Commit message: "feat(uow): wire lifecycle repositories"**
+  - [X] Git [Jeff]: `git checkout master` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git pull origin master` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git checkout -b feature/m1-uow-lifecycle` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git fetch origin && git merge origin/master` (run before final tests and before commit) Done: Day 7, February 15, 2026
+  - [X] Code [Jeff]: Update `UnitOfWork` to expose `actions` + `lifecycle_plans` repositories and remove legacy plan repository accessors. Done: Day 7, February 15, 2026
+  - [X] Code [Jeff]: Update `UnitOfWorkContext` to include new repos and remove legacy plan/change repositories from default access. Done: Day 7, February 15, 2026
+  - [X] Docs [Jeff]: Update `docs/reference/repositories.md` with UoW lifecycle usage patterns. Done: Day 7, February 15, 2026
+  - [X] Tests (Behave) [Jeff]: Add scenarios verifying UoW exposes action + lifecycle plan repositories. Done: Day 7, February 15, 2026
+  - [X] Tests (Robot) [Jeff]: Add Robot smoke test that creates an action + plan via UoW. Done: Day 7, February 15, 2026
+  - [X] Tests (ASV) [Jeff]: Add `benchmarks/uow_lifecycle_bench.py` for UoW overhead baseline. Done: Day 7, February 15, 2026
+  - [X] Quality [Jeff]: Run `nox` (all default sessions, including benchmark), fix any errors if needed ensuring nox passes. Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git add .` Done: Day 7, February 15, 2026
+  - [X] Git [Jeff]: `git commit -m "feat(uow): wire lifecycle repositories"` Done: Day 7, February 15, 2026
   - [ ] Forgejo PR [Jeff]: Open PR from `feature/m1-uow-lifecycle` to `master` with description "Wire lifecycle repositories into UnitOfWork and remove legacy plan repo access.".
   - [ ] Git [Jeff]: `git checkout master`
   - [ ] Git [Jeff]: `git branch -d feature/m1-uow-lifecycle`
-  - [ ] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%.
+  - [X] Quality [Jeff]: Verify coverage >=97% via `nox -s coverage_report`. If coverage is <97% then review the current unit test coverage report at `build/coverage.xml` and use it to write new Behave based unit tests to improve code coverage. Specifically, write Behave style unit tests that are descriptively named and specifically improves coverage on whichever file has the most uncovered lines by writing tests that will target the uncovered lines in the report. Once that is done rerun `nox -s coverage_report` to verify all tests pass and coverage is above >=97%, if not repeat this task as many times as is needed until coverage reaches >=97%. Done: Day 7, February 15, 2026
 
 - [ ] **COMMIT (Owner: Jeff | Group: A5.zeta | Branch: feature/m1-lifecycle-persist | Planned: Day 10 | Expected: Day 12) - Commit message: "feat(service): persist plan lifecycle via repositories"**
   - [ ] Git [Jeff]: `git checkout master`

robot/helper_plan_phase_migration.py

@@ -0,0 +1,157 @@
+"""Helper script for plan phase migration Robot integration tests.
+
+Tests that the rebaselined v3_plans schema accepts Action phase
+and Apply terminal processing states.
+"""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+from datetime import UTC, datetime
+from pathlib import Path
+
+from sqlalchemy.orm import Session
+
+from cleveragents.domain.models.core.action import Action, ActionState
+from cleveragents.domain.models.core.plan import (
+    AutomationLevel,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    PlanPhase,
+    PlanTimestamps,
+    ProcessingState,
+)
+from cleveragents.infrastructure.database.models import (
+    LifecycleActionModel,
+    LifecyclePlanModel,
+    init_database,
+)
+
+
+def _create_temp_db() -> tuple[Session, str]:
+    """Create a temporary SQLite database for testing."""
+    tmp = tempfile.mktemp(suffix=".db")
+    db_url = f"sqlite:///{tmp}"
+    engine = init_database(db_url)
+    session = Session(bind=engine)
+    return session, tmp
+
+
+def _ensure_action(session: Session) -> None:
+    """Ensure a parent action exists for FK references."""
+    action = Action(
+        namespaced_name=NamespacedName.parse("local/phase-migration-action"),
+        description="Migration 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),
+    )
+    db_model = LifecycleActionModel.from_domain(action)
+    session.add(db_model)
+    session.commit()
+
+
+def _action_queued() -> None:
+    """Test: insert plan with phase='action', processing_state='queued'."""
+    session, tmp = _create_temp_db()
+    try:
+        _ensure_action(session)
+
+        plan = Plan(
+            identity=PlanIdentity(plan_id="01HV000000000000000000MG01"),
+            action_name="local/phase-migration-action",
+            namespaced_name=NamespacedName.parse("local/phase-migration-action"),
+            description="Phase migration test plan",
+            phase=PlanPhase.ACTION,
+            processing_state=ProcessingState.QUEUED,
+            automation_level=AutomationLevel.MANUAL,
+            strategy_actor="local/s",
+            execution_actor="local/e",
+            timestamps=PlanTimestamps(
+                created_at=datetime(2026, 2, 15, tzinfo=UTC),
+                updated_at=datetime(2026, 2, 15, tzinfo=UTC),
+            ),
+        )
+
+        plan_db = LifecyclePlanModel.from_domain(
+            plan, action_name="local/phase-migration-action"
+        )
+        session.add(plan_db)
+        session.commit()
+
+        loaded = (
+            session.query(LifecyclePlanModel)
+            .filter_by(plan_id="01HV000000000000000000MG01")
+            .one()
+        )
+        assert loaded.phase == "action"
+        assert loaded.processing_state == "queued"
+
+        print("action-queued-ok")
+    finally:
+        session.close()
+        Path(tmp).unlink(missing_ok=True)
+
+
+def _apply_applied() -> None:
+    """Test: insert plan with phase='apply', processing_state='applied'."""
+    session, tmp = _create_temp_db()
+    try:
+        _ensure_action(session)
+
+        plan = Plan(
+            identity=PlanIdentity(plan_id="01HV000000000000000000MG02"),
+            action_name="local/phase-migration-action",
+            namespaced_name=NamespacedName.parse("local/phase-migration-action"),
+            description="Apply terminal test plan",
+            phase=PlanPhase.APPLY,
+            processing_state=ProcessingState.APPLIED,
+            automation_level=AutomationLevel.MANUAL,
+            strategy_actor="local/s",
+            execution_actor="local/e",
+            timestamps=PlanTimestamps(
+                created_at=datetime(2026, 2, 15, tzinfo=UTC),
+                updated_at=datetime(2026, 2, 15, tzinfo=UTC),
+            ),
+        )
+
+        plan_db = LifecyclePlanModel.from_domain(
+            plan, action_name="local/phase-migration-action"
+        )
+        session.add(plan_db)
+        session.commit()
+
+        loaded = (
+            session.query(LifecyclePlanModel)
+            .filter_by(plan_id="01HV000000000000000000MG02")
+            .one()
+        )
+        assert loaded.phase == "apply"
+        assert loaded.processing_state == "applied"
+
+        print("apply-applied-ok")
+    finally:
+        session.close()
+        Path(tmp).unlink(missing_ok=True)
+
+
+def main() -> None:
+    if len(sys.argv) < 2:
+        raise SystemExit("Expected command argument")
+    command = sys.argv[1]
+    commands = {
+        "action-queued": _action_queued,
+        "apply-applied": _apply_applied,
+    }
+    if command not in commands:
+        raise SystemExit(f"Unknown command: {command}")
+    commands[command]()
+
+
+if __name__ == "__main__":
+    main()

robot/helper_plan_repository.py

@@ -0,0 +1,110 @@
+"""Helper script for plan_repository.robot smoke test."""
+
+from __future__ import annotations
+
+import sys
+
+sys.path.insert(0, "src")
+sys.path.insert(0, ".")
+
+from datetime import datetime
+
+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,
+    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,
+)
+
+
+def main() -> None:
+    engine = create_engine("sqlite:///:memory:")
+    Base.metadata.create_all(engine)
+    session = sessionmaker(bind=engine)()
+    factory = lambda: session  # noqa: E731
+
+    # Create prerequisite action
+    action_repo = ActionRepository(session_factory=factory)
+    action = Action(
+        namespaced_name=NamespacedName(namespace="local", name="robot-action"),
+        description="Robot test action",
+        long_description=None,
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        estimation_actor=None,
+        review_actor=None,
+        arguments=[],
+        reusable=True,
+        read_only=False,
+        state=ActionState("available"),
+        created_at=datetime.now(),
+        updated_at=datetime.now(),
+        created_by=None,
+        tags=[],
+    )
+    action_repo.create(action)
+    session.commit()
+
+    # Test plan CRUD
+    repo = LifecyclePlanRepository(session_factory=factory)
+    now = datetime.now()
+    plan = V3Plan(
+        identity=PlanIdentity(plan_id="01KHG51GK1G5TJ678GT8JEF55Y", attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="robot-plan"),
+        action_name="local/robot-action",
+        description="Robot test",
+        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="robot",
+        tags=[],
+        reusable=True,
+        read_only=False,
+    )
+
+    # Create
+    repo.create(plan)
+    session.commit()
+
+    # Get
+    got = repo.get("01KHG51GK1G5TJ678GT8JEF55Y")
+    assert got is not None, "Plan not found"
+    assert got.description == "Robot test", f"Bad desc: {got.description}"
+
+    # List
+    plans = repo.list_plans(phase="action")
+    assert len(plans) == 1, f"Expected 1, got {len(plans)}"
+
+    # Count
+    count = repo.count(phase="action")
+    assert count == 1, f"Expected count 1, got {count}"
+
+    # Delete
+    deleted = repo.delete("01KHG51GK1G5TJ678GT8JEF55Y")
+    session.commit()
+    assert deleted is True, "Delete failed"
+
+    gone = repo.get("01KHG51GK1G5TJ678GT8JEF55Y")
+    assert gone is None, "Plan still exists"
+
+    print("PASS: plan_repository smoke test")
+
+
+if __name__ == "__main__":
+    main()

robot/helper_uow_lifecycle.py

@@ -0,0 +1,128 @@
+"""Helper script for uow_lifecycle.robot smoke test."""
+
+from __future__ import annotations
+
+import sys
+
+sys.path.insert(0, "src")
+sys.path.insert(0, ".")
+
+from datetime import datetime
+
+from sqlalchemy import create_engine, event
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.domain.models.core.action import Action, ActionState
+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,
+)
+from cleveragents.infrastructure.database.unit_of_work import UnitOfWorkContext
+
+
+def main() -> None:
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        future=True,
+        connect_args={"check_same_thread": False},
+    )
+
+    @event.listens_for(engine, "connect")
+    def _fk(dbapi_conn, _rec):  # type: ignore[no-untyped-def]
+        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,
+    )
+
+    # --- Test 1: create action + plan via UoW ---
+    session = sf()
+    ctx = UnitOfWorkContext(session)
+
+    assert isinstance(ctx.actions, ActionRepository), "actions not ActionRepository"
+    assert isinstance(ctx.lifecycle_plans, LifecyclePlanRepository), (
+        "lifecycle_plans not LifecyclePlanRepository"
+    )
+
+    now = datetime.now()
+    action = Action(
+        namespaced_name=NamespacedName(namespace="local", name="robot-uow-action"),
+        description="Robot UoW test action",
+        long_description=None,
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        estimation_actor=None,
+        review_actor=None,
+        arguments=[],
+        reusable=True,
+        read_only=False,
+        state=ActionState("available"),
+        created_at=now,
+        updated_at=now,
+        created_by=None,
+        tags=[],
+    )
+    ctx.actions.create(action)
+
+    plan = V3Plan(
+        identity=PlanIdentity(plan_id="01HGZ6FE0AQDYTR4BX00000R01", attempt=1),
+        namespaced_name=NamespacedName(namespace="local", name="robot-uow-plan"),
+        action_name="local/robot-uow-action",
+        description="Robot UoW test plan",
+        definition_of_done="Done",
+        strategy_actor="local/s",
+        execution_actor="local/e",
+        phase=PlanPhase("action"),
+        processing_state=ProcessingState("queued"),
+        automation_level=AutomationLevel("manual"),
+        timestamps=PlanTimestamps(created_at=now, updated_at=now),
+        project_links=[],
+        arguments={},
+        arguments_order=[],
+        invariants=[],
+        reusable=True,
+        read_only=False,
+        created_by=None,
+        tags=[],
+    )
+    ctx.lifecycle_plans.create(plan)
+    session.commit()
+    session.close()
+
+    # --- Test 2: verify retrieval in new session ---
+    session2 = sf()
+    ctx2 = UnitOfWorkContext(session2)
+    found_action = ctx2.actions.get_by_name("local/robot-uow-action")
+    assert found_action is not None, "Action not found after commit"
+    found_plan = ctx2.lifecycle_plans.get("01HGZ6FE0AQDYTR4BX00000R01")
+    assert found_plan is not None, "Plan not found after commit"
+    session2.close()
+
+    print("PASS: uow_lifecycle smoke test")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as exc:
+        print(f"FAIL: {exc}", file=sys.stderr)
+        sys.exit(1)

robot/plan_phase_migration.robot

@@ -0,0 +1,25 @@
+*** Settings ***
+Documentation     Integration tests for plan phase/state rebaseline migration (a5_005).
+...               Verifies that the rebaselined v3_plans table accepts the Action phase
+...               and Apply terminal processing states.
+Resource          ${CURDIR}/common.resource
+Suite Setup       Setup Test Environment
+Suite Teardown    Cleanup Test Environment
+
+*** Variables ***
+${HELPER_SCRIPT}    robot/helper_plan_phase_migration.py
+
+*** Test Cases ***
+Insert Plan With Action Phase And Queued State
+    [Documentation]    Verify a plan with phase='action' and processing_state='queued' persists
+    [Tags]    database    migration    plan_phase
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    action-queued    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    action-queued-ok
+
+Insert Plan With Apply Phase And Applied State
+    [Documentation]    Verify a plan with phase='apply' and processing_state='applied' persists
+    [Tags]    database    migration    plan_phase
+    ${result}=    Run Process    ${PYTHON}    ${HELPER_SCRIPT}    apply-applied    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    apply-applied-ok

robot/plan_repository.robot

@@ -0,0 +1,17 @@
+*** Settings ***
+Documentation    Smoke test for LifecyclePlanRepository CRUD operations
+Library          Process
+Library          OperatingSystem
+Library          String
+
+*** Variables ***
+${PYTHON}    python
+
+*** Test Cases ***
+Plan Repository CRUD Via Helper Script
+    [Documentation]    Create, retrieve, list, count, and delete a plan via LifecyclePlanRepository
+    ${result}=    Run Process    ${PYTHON}    ${CURDIR}/helper_plan_repository.py
+    ...    stdout=STDOUT    stderr=STDOUT    timeout=30s
+    Log    ${result.stdout}
+    Should Contain    ${result.stdout}    PASS: plan_repository smoke test
+    Should Be Equal As Integers    ${result.rc}    0

robot/uow_lifecycle.robot

@@ -0,0 +1,17 @@
+*** Settings ***
+Documentation    Smoke test for UnitOfWork lifecycle repository wiring
+Library          Process
+Library          OperatingSystem
+Library          String
+
+*** Variables ***
+${PYTHON}    python
+
+*** Test Cases ***
+UoW Lifecycle Action And Plan Via Helper Script
+    [Documentation]    Create action + plan via UoW, verify retrieval in new session
+    ${result}=    Run Process    ${PYTHON}    ${CURDIR}/helper_uow_lifecycle.py
+    ...    stdout=STDOUT    stderr=STDOUT    timeout=30s
+    Log    ${result.stdout}
+    Should Contain    ${result.stdout}    PASS: uow_lifecycle smoke test
+    Should Be Equal As Integers    ${result.rc}    0

src/cleveragents/infrastructure/database/__init__.py

@@ -28,6 +28,9 @@ from .repositories import (
     ChangeRepository,
     ContextRepository,
     DuplicateActionError,
+    DuplicatePlanError,
+    LifecyclePlanRepository,
+    PlanNotFoundError,
     PlanRepository,
     ProjectRepository,
 )
@@ -44,11 +47,14 @@ __all__ = [
     "ContextModel",
     "ContextRepository",
     "DuplicateActionError",
+    "DuplicatePlanError",
     "LifecycleActionModel",
     "LifecyclePlanModel",
+    "LifecyclePlanRepository",
     "PlanArgumentModel",
     "PlanInvariantModel",
     "PlanModel",
+    "PlanNotFoundError",
     "PlanProjectModel",
     "PlanRepository",
     "ProjectModel",

src/cleveragents/infrastructure/database/models.py

@@ -524,9 +524,13 @@ class LifecyclePlanModel(Base):  # type: ignore[misc]
     namespaced_name = Column(String(255), nullable=False)
     namespace = Column(String(100), nullable=False)
 
-    # Lifecycle phase and processing state (spec-aligned)
-    phase = Column(String(20), nullable=False, default="strategize")
-    processing_state = Column(String(20), nullable=False, default="queued")
+    # Lifecycle phase and processing state (spec-aligned, rebaselined in a5_005)
+    phase = Column(
+        String(20), nullable=False, default="action", server_default="action"
+    )
+    processing_state = Column(
+        String(20), nullable=False, default="queued", server_default="queued"
+    )
     attempt = Column(Integer, nullable=False, default=1)
 
     # Rendered description and DoD
@@ -543,6 +547,8 @@ class LifecyclePlanModel(Base):  # type: ignore[misc]
 
     # Policy / profile
     automation_profile = Column(Text, nullable=True)
+    # Legacy: automation_level is retained for backward compatibility.
+    # New code should use automation_profile. Will be removed in A6.legacy.
     automation_level = Column(String(30), nullable=False, default="manual")
 
     # Behavior
@@ -614,6 +620,20 @@ class LifecyclePlanModel(Base):  # type: ignore[misc]
     )
 
     __table_args__ = (
+        CheckConstraint(
+            "phase IN ('action', 'strategize', 'execute', 'apply')",
+            name="ck_v3_plans_phase",
+        ),
+        CheckConstraint(
+            "processing_state IN "
+            "('queued', 'processing', 'errored', 'complete', 'cancelled', "
+            "'applied', 'constrained')",
+            name="ck_v3_plans_state",
+        ),
+        CheckConstraint(
+            "automation_level IN ('manual', 'review_before_apply', 'full_automation')",
+            name="ck_v3_plans_automation",
+        ),
         Index("ix_v3_plans_phase", "phase"),
         Index("ix_v3_plans_state", "processing_state"),
         Index("ix_v3_plans_parent", "parent_plan_id"),

src/cleveragents/infrastructure/database/repositories.py

@@ -6,6 +6,7 @@ Now includes retry patterns for database operations based on ADR-033.
 
 from __future__ import annotations
 
+import json
 from collections.abc import Callable
 from datetime import datetime
 from pathlib import Path
@@ -35,7 +36,11 @@ from cleveragents.infrastructure.database.models import (
     ContextModel,
     DebugAttemptModel,
     LifecycleActionModel,
+    LifecyclePlanModel,
+    PlanArgumentModel,
+    PlanInvariantModel,
     PlanModel,
+    PlanProjectModel,
     ProjectModel,
 )
 
@@ -1050,3 +1055,368 @@ class ActionRepository:
             raise DatabaseError(
                 f"Failed to delete action {action_name}: {exc}"
             ) from exc
+
+
+# ---------------------------------------------------------------------------
+# V3 Lifecycle Plan Repository
+# ---------------------------------------------------------------------------
+
+
+class DuplicatePlanError(DatabaseError):
+    """Raised when creating a plan with an ID that already exists."""
+
+    def __init__(self, plan_id: str):
+        super().__init__(f"Plan with id '{plan_id}' already exists")
+        self.plan_id = plan_id
+
+
+class PlanNotFoundError(DatabaseError):
+    """Raised when a plan cannot be found for update/delete."""
+
+    def __init__(self, plan_id: str):
+        super().__init__(f"Plan '{plan_id}' not found")
+        self.plan_id = plan_id
+
+
+class LifecyclePlanRepository:
+    """Repository for v3 lifecycle plan persistence.
+
+    Uses a session-factory pattern: each public method obtains its own
+    session from the factory, ensuring proper session lifecycle management.
+
+    All mutating methods flush (but do NOT commit); the caller or a
+    ``UnitOfWork`` wrapper is responsible for committing the transaction.
+    """
+
+    def __init__(self, session_factory: Callable[[], Session]) -> None:
+        """Initialise with a callable that returns a new SQLAlchemy Session."""
+        self._session_factory = session_factory
+
+    def _session(self) -> Session:
+        """Convenience helper to obtain a session."""
+        return self._session_factory()
+
+    @database_retry
+    def create(self, plan: Any) -> Any:
+        """Persist a new ``Plan`` domain object.
+
+        Args:
+            plan: A v3 ``Plan`` domain model instance.
+
+        Returns:
+            The same ``Plan`` after persistence.
+
+        Raises:
+            DuplicatePlanError: If a plan with the same ID already exists.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            db_model = LifecyclePlanModel.from_domain(plan)
+            session.add(db_model)
+            session.flush()
+            return plan
+        except IntegrityError as exc:
+            session.rollback()
+            exc_str = str(exc).upper()
+            if "UNIQUE" in exc_str or "PRIMARY" in exc_str:
+                raise DuplicatePlanError(str(plan.identity.plan_id)) from exc
+            raise DatabaseError(f"Failed to create plan: {exc}") from exc
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to create plan: {exc}") from exc
+
+    @database_retry
+    def get(self, plan_id: str) -> Any | None:
+        """Retrieve a plan by its ULID primary key.
+
+        Returns:
+            The ``Plan`` domain object, or ``None`` if not found.
+        """
+        session = self._session()
+        try:
+            row = session.query(LifecyclePlanModel).filter_by(plan_id=plan_id).first()
+            if row is None:
+                return None
+            return row.to_domain()
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to get plan {plan_id}: {exc}") from exc
+
+    @database_retry
+    def get_by_name(self, namespaced_name: str) -> Any | None:
+        """Retrieve a plan by its full namespaced name string.
+
+        Returns:
+            The ``Plan`` domain object, or ``None`` if not found.
+        """
+        session = self._session()
+        try:
+            row = (
+                session.query(LifecyclePlanModel)
+                .filter_by(namespaced_name=namespaced_name)
+                .first()
+            )
+            if row is None:
+                return None
+            return row.to_domain()
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(
+                f"Failed to get plan by name '{namespaced_name}': {exc}"
+            ) from exc
+
+    @database_retry
+    def update(self, plan: Any) -> Any:
+        """Update all mutable fields of an existing plan.
+
+        Args:
+            plan: The ``Plan`` domain model with updated fields.
+
+        Returns:
+            The same ``Plan`` after persistence.
+
+        Raises:
+            PlanNotFoundError: If the plan is not found.
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        plan_id_str = str(plan.identity.plan_id)
+        try:
+            row = (
+                session.query(LifecyclePlanModel).filter_by(plan_id=plan_id_str).first()
+            )
+            if row is None:
+                raise PlanNotFoundError(plan_id_str)
+
+            # Overwrite mutable columns from domain model
+            row.phase = (  # type: ignore[assignment]
+                plan.phase.value if hasattr(plan.phase, "value") else plan.phase
+            )
+            row.processing_state = (  # type: ignore[assignment]
+                plan.processing_state.value
+                if hasattr(plan.processing_state, "value")
+                else str(plan.processing_state)
+            )
+            row.attempt = plan.identity.attempt  # type: ignore[assignment]
+            row.description = plan.description  # type: ignore[assignment]
+            row.definition_of_done = plan.definition_of_done  # type: ignore[assignment]
+            row.strategy_actor = plan.strategy_actor  # type: ignore[assignment]
+            row.execution_actor = plan.execution_actor  # type: ignore[assignment]
+            row.review_actor = getattr(plan, "review_actor", None)  # type: ignore[assignment]
+            row.apply_actor = getattr(plan, "apply_actor", None)  # type: ignore[assignment]
+            row.estimation_actor = getattr(plan, "estimation_actor", None)  # type: ignore[assignment]
+            row.invariant_actor = getattr(plan, "invariant_actor", None)  # type: ignore[assignment]
+
+            # Serialize automation profile
+            automation_profile_json = (
+                json.dumps(plan.automation_profile.model_dump(mode="json"))
+                if plan.automation_profile
+                else None
+            )
+            row.automation_profile = automation_profile_json  # type: ignore[assignment]
+
+            row.error_message = plan.error_message  # type: ignore[assignment]
+            row.error_details_json = (  # type: ignore[assignment]
+                json.dumps(plan.error_details)
+                if getattr(plan, "error_details", None) is not None
+                else None
+            )
+            row.changeset_id = getattr(plan, "changeset_id", None)  # type: ignore[assignment]
+            row.sandbox_refs_json = json.dumps(  # type: ignore[assignment]
+                getattr(plan, "sandbox_refs", []) or []
+            )
+            row.validation_summary_json = (  # type: ignore[assignment]
+                json.dumps(plan.validation_summary)
+                if getattr(plan, "validation_summary", None) is not None
+                else None
+            )
+            row.decision_root_id = getattr(plan, "decision_root_id", None)  # type: ignore[assignment]
+
+            row.reusable = plan.reusable  # type: ignore[assignment]
+            row.read_only = plan.read_only  # type: ignore[assignment]
+            row.created_by = plan.created_by  # type: ignore[assignment]
+            row.tags_json = json.dumps(plan.tags)  # type: ignore[assignment]
+
+            # Timestamps
+            row.updated_at = plan.timestamps.updated_at.isoformat()  # type: ignore[assignment]
+            row.completed_at = LifecyclePlanModel._to_iso(  # type: ignore[assignment]
+                plan.timestamps.applied_at
+            )
+            row.strategize_started_at = LifecyclePlanModel._to_iso(  # type: ignore[assignment]
+                plan.timestamps.strategize_started_at
+            )
+            row.strategize_completed_at = LifecyclePlanModel._to_iso(  # type: ignore[assignment]
+                plan.timestamps.strategize_completed_at
+            )
+            row.execute_started_at = LifecyclePlanModel._to_iso(  # type: ignore[assignment]
+                plan.timestamps.execute_started_at
+            )
+            row.execute_completed_at = LifecyclePlanModel._to_iso(  # type: ignore[assignment]
+                plan.timestamps.execute_completed_at
+            )
+            row.apply_started_at = LifecyclePlanModel._to_iso(  # type: ignore[assignment]
+                plan.timestamps.apply_started_at
+            )
+            row.applied_at = LifecyclePlanModel._to_iso(  # type: ignore[assignment]
+                plan.timestamps.applied_at
+            )
+
+            # Replace child project links
+            row.project_links_rel.clear()  # type: ignore[union-attr]
+            now_iso = plan.timestamps.updated_at.isoformat()
+            for pl in getattr(plan, "project_links", []) or []:
+                row.project_links_rel.append(  # type: ignore[union-attr]
+                    PlanProjectModel(
+                        project_name=pl.project_name,
+                        alias=pl.alias,
+                        read_only=pl.read_only,
+                        created_at=now_iso,
+                    )
+                )
+
+            # Replace child arguments
+            row.arguments_rel.clear()  # type: ignore[union-attr]
+            arguments_dict: dict[str, Any] = getattr(plan, "arguments", {}) or {}
+            arguments_order: list[str] = getattr(plan, "arguments_order", []) or list(
+                arguments_dict.keys()
+            )
+            for idx, arg_name in enumerate(arguments_order):
+                if arg_name in arguments_dict:
+                    row.arguments_rel.append(  # type: ignore[union-attr]
+                        PlanArgumentModel(
+                            name=arg_name,
+                            value_json=json.dumps(arguments_dict[arg_name]),
+                            value_type="string",
+                            position=idx,
+                        )
+                    )
+
+            # Replace child invariants
+            row.invariants_rel.clear()  # type: ignore[union-attr]
+            for idx, inv in enumerate(getattr(plan, "invariants", []) or []):
+                inv_text = inv.text if hasattr(inv, "text") else str(inv)
+                raw_source: Any = inv.source if hasattr(inv, "source") else "plan"
+                inv_source_str: str = (
+                    raw_source.value
+                    if hasattr(raw_source, "value")
+                    else str(raw_source)
+                )
+                row.invariants_rel.append(  # type: ignore[union-attr]
+                    PlanInvariantModel(
+                        invariant_text=inv_text,
+                        source_scope=inv_source_str,
+                        source_name=None,
+                        position=idx,
+                        created_at=now_iso,
+                    )
+                )
+
+            session.flush()
+            return plan
+        except PlanNotFoundError:
+            raise
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to update plan {plan_id_str}: {exc}") from exc
+
+    @database_retry
+    def delete(self, plan_id: str) -> bool:
+        """Delete a plan by ULID, cascading to child tables.
+
+        Args:
+            plan_id: The ULID string of the plan to delete.
+
+        Returns:
+            ``True`` if the plan was deleted, ``False`` if not found.
+
+        Raises:
+            DatabaseError: On transient or unexpected DB errors.
+        """
+        session = self._session()
+        try:
+            row = session.query(LifecyclePlanModel).filter_by(plan_id=plan_id).first()
+            if row is None:
+                return False
+            session.delete(row)
+            session.flush()
+            return True
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to delete plan {plan_id}: {exc}") from exc
+
+    @database_retry
+    def list_plans(
+        self,
+        phase: str | None = None,
+        processing_state: str | None = None,
+        action_name: str | None = None,
+        project_name: str | None = None,
+        namespace: str | None = None,
+        limit: int = 100,
+        offset: int = 0,
+    ) -> list[Any]:
+        """List plans with optional filters, ordered by created_at DESC.
+
+        Args:
+            phase: Optional phase filter (e.g., ``"action"``).
+            processing_state: Optional state filter (e.g., ``"queued"``).
+            action_name: Optional action name filter.
+            project_name: Optional project name filter (joins plan_projects).
+            namespace: Optional namespace filter.
+            limit: Maximum number of results (default 100).
+            offset: Number of results to skip (default 0).
+
+        Returns:
+            List of ``Plan`` domain objects.
+        """
+        session = self._session()
+        try:
+            query = session.query(LifecyclePlanModel)
+            if phase is not None:
+                query = query.filter(LifecyclePlanModel.phase == phase)
+            if processing_state is not None:
+                query = query.filter(
+                    LifecyclePlanModel.processing_state == processing_state
+                )
+            if action_name is not None:
+                query = query.filter(LifecyclePlanModel.action_name == action_name)
+            if namespace is not None:
+                query = query.filter(LifecyclePlanModel.namespace == namespace)
+            if project_name is not None:
+                query = query.join(
+                    PlanProjectModel,
+                    PlanProjectModel.plan_id == LifecyclePlanModel.plan_id,
+                ).filter(PlanProjectModel.project_name == project_name)
+            query = query.order_by(LifecyclePlanModel.created_at.desc())
+            query = query.offset(offset).limit(limit)
+            rows = query.all()
+            return [row.to_domain() for row in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to list plans: {exc}") from exc
+
+    @database_retry
+    def count(
+        self,
+        phase: str | None = None,
+        processing_state: str | None = None,
+    ) -> int:
+        """Count matching plans with optional filters.
+
+        Args:
+            phase: Optional phase filter.
+            processing_state: Optional state filter.
+
+        Returns:
+            Number of matching plans.
+        """
+        session = self._session()
+        try:
+            query = session.query(LifecyclePlanModel)
+            if phase is not None:
+                query = query.filter(LifecyclePlanModel.phase == phase)
+            if processing_state is not None:
+                query = query.filter(
+                    LifecyclePlanModel.processing_state == processing_state
+                )
+            return cast(int, query.count())
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to count plans: {exc}") from exc

src/cleveragents/infrastructure/database/unit_of_work.py

@@ -14,10 +14,12 @@ from sqlalchemy.orm import Session, sessionmaker
 
 from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
 from cleveragents.infrastructure.database.repositories import (
+    ActionRepository,
     ActorRepository,
     ChangeRepository,
     ContextRepository,
     DebugAttemptRepository,
+    LifecyclePlanRepository,
     PlanRepository,
     ProjectRepository,
 )
@@ -180,6 +182,12 @@ class UnitOfWorkContext:
         self._changes: ChangeRepository | None = None
         self._debug_attempts: DebugAttemptRepository | None = None
         self._actors: ActorRepository | None = None
+        self._actions: ActionRepository | None = None
+        self._lifecycle_plans: LifecyclePlanRepository | None = None
+
+    def _session_factory(self) -> Session:
+        """Return the transaction's session for factory-pattern repositories."""
+        return self._session
 
     @property
     def projects(self) -> ProjectRepository:
@@ -190,11 +198,39 @@ class UnitOfWorkContext:
 
     @property
     def plans(self) -> PlanRepository:
-        """Get plan repository for this transaction."""
+        """Get legacy plan repository for this transaction.
+
+        .. deprecated::
+            Use :attr:`lifecycle_plans` for v3 lifecycle plans.
+            This accessor targets the legacy ``plans`` table and will
+            be removed in a future release.
+        """
         if self._plans is None:
             self._plans = PlanRepository(self._session)
         return self._plans
 
+    @property
+    def actions(self) -> ActionRepository:
+        """Get v3 action repository for this transaction.
+
+        Uses the session-factory pattern required by ActionRepository.
+        """
+        if self._actions is None:
+            self._actions = ActionRepository(session_factory=self._session_factory)
+        return self._actions
+
+    @property
+    def lifecycle_plans(self) -> LifecyclePlanRepository:
+        """Get v3 lifecycle plan repository for this transaction.
+
+        Uses the session-factory pattern required by LifecyclePlanRepository.
+        """
+        if self._lifecycle_plans is None:
+            self._lifecycle_plans = LifecyclePlanRepository(
+                session_factory=self._session_factory,
+            )
+        return self._lifecycle_plans
+
     @property
     def contexts(self) -> ContextRepository:
         """Get context repository for this transaction."""