fix(persistence): persist reversion_count, last_completed_step, and last_checkpoint_id on LifecyclePlanModel

Implemented persistence for three previously missing Plan fields to ensure proper resume and reversion behavior across restarts. The changes address the gaps in the database, ORM mapping, repository logic, and tests.

- Alembic migration
  - File: alembic/versions/m9_002_plan_resume_fields.py
  - Adds three columns to the v3_plans table:
    - reversion_count: INTEGER NOT NULL DEFAULT 0 (server_default)
    - last_completed_step: INTEGER NOT NULL DEFAULT -1 (server_default)
    - last_checkpoint_id: TEXT nullable (no server_default)
  - Migration chains from m9_001_session_name_column to align with existing plan schema evolution.

- SQLAlchemy model
  - File: src/cleveragents/infrastructure/database/models.py
  - Updated LifecyclePlanModel to include reversion_count, last_completed_step, and last_checkpoint_id columns.
  - from_domain(): updated to serialize all three fields.
  - to_domain(): updated to deserialize all three fields with proper cast() typing, ensuring correct domain conversions.

- Repository fix
  - File: src/cleveragents/infrastructure/database/repositories.py
  - Fixed LifecyclePlanRepository.update() which was silently dropping the three fields on every update.
  - This remediation ensures updates preserve reversion_count, last_completed_step, and last_checkpoint_id.

- Behave tests
  - Files: features/plan_resume_fields_persistence.feature and associated steps
  - Added 7 scenarios validating individual field persistence, combined persistence, default values, cross-reconnection persistence, and update persistence.
  - Tests ensure correctness of persistence behavior across restarts and updates.

Key Design Decisions
- server_default used for migration columns to backfill defaults automatically for existing rows without a separate backfill step.
- last_checkpoint_id is nullable (no server_default) because None is the correct default in the domain model.
- The update() fix in LifecyclePlanRepository was a bonus discovery; without it, updates could silently drop the resume fields and break persistence guarantees.

ISSUES CLOSED: #2864

alembic/versions/m9_002_plan_resume_fields.py

@@ -0,0 +1,64 @@
+"""Add reversion_count, last_completed_step, last_checkpoint_id columns to v3_plans.
+
+These columns are required for plan resume and reversion-limit functionality.
+Without them, ``Plan.reversion_count``, ``last_completed_step``, and
+``last_checkpoint_id`` are not persisted to the database, causing plan resume
+and reversion limits to fail after a process restart.
+
+Revision ID: m9_002_plan_resume_fields
+Revises: m9_001_session_name_column
+Create Date: 2026-04-05 00:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "m9_002_plan_resume_fields"
+down_revision: str | Sequence[str] | None = "m9_001_session_name_column"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Add resume-tracking columns to ``v3_plans``.
+
+    Adds ``reversion_count``, ``last_completed_step``, and
+    ``last_checkpoint_id``.
+    """
+    op.add_column(
+        "v3_plans",
+        sa.Column(
+            "reversion_count",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+        ),
+    )
+    op.add_column(
+        "v3_plans",
+        sa.Column(
+            "last_completed_step",
+            sa.Integer(),
+            nullable=False,
+            server_default="-1",
+        ),
+    )
+    op.add_column(
+        "v3_plans",
+        sa.Column("last_checkpoint_id", sa.Text(), nullable=True),
+    )
+
+
+def downgrade() -> None:
+    """Remove resume-tracking columns from ``v3_plans``.
+
+    Drops ``reversion_count``, ``last_completed_step``, and
+    ``last_checkpoint_id``.
+    """
+    op.drop_column("v3_plans", "last_checkpoint_id")
+    op.drop_column("v3_plans", "last_completed_step")
+    op.drop_column("v3_plans", "reversion_count")

features/plan_resume_fields_persistence.feature

@@ -0,0 +1,70 @@
+Feature: Plan resume fields round-trip persistence
+  As a developer
+  I want reversion_count, last_completed_step, and last_checkpoint_id to persist through the database
+  So that plan resume and reversion limits work correctly after process restarts
+
+  Background:
+    Given a fresh in-memory plan persistence database
+    And a prerequisite action "local/persist-action" exists in the database
+
+  Scenario: reversion_count persists through database round-trip
+    Given a new lifecycle plan with ID "01HV000000000000000000RP01"
+    And the plan has reversion_count 2
+    When I persist the plan via the plan repository
+    And I retrieve the plan by ID "01HV000000000000000000RP01"
+    Then the retrieved plan reversion_count should be 2
+
+  Scenario: last_completed_step persists through database round-trip
+    Given a new lifecycle plan with ID "01HV000000000000000000RP02"
+    And the plan has last_completed_step 4
+    When I persist the plan via the plan repository
+    And I retrieve the plan by ID "01HV000000000000000000RP02"
+    Then the retrieved plan last_completed_step should be 4
+
+  Scenario: last_checkpoint_id persists through database round-trip
+    Given a new lifecycle plan with ID "01HV000000000000000000RP03"
+    And the plan has last_checkpoint_id "01HV000000000000000000CP01"
+    When I persist the plan via the plan repository
+    And I retrieve the plan by ID "01HV000000000000000000RP03"
+    Then the retrieved plan last_checkpoint_id should be "01HV000000000000000000CP01"
+
+  Scenario: All three resume fields persist together through database round-trip
+    Given a new lifecycle plan with ID "01HV000000000000000000RP04"
+    And the plan has reversion_count 1
+    And the plan has last_completed_step 3
+    And the plan has last_checkpoint_id "01HV000000000000000000CP02"
+    When I persist the plan via the plan repository
+    And I retrieve the plan by ID "01HV000000000000000000RP04"
+    Then the retrieved plan reversion_count should be 1
+    And the retrieved plan last_completed_step should be 3
+    And the retrieved plan last_checkpoint_id should be "01HV000000000000000000CP02"
+
+  Scenario: Default values persist correctly when fields are not set
+    Given a new lifecycle plan with ID "01HV000000000000000000RP05"
+    When I persist the plan via the plan repository
+    And I retrieve the plan by ID "01HV000000000000000000RP05"
+    Then the retrieved plan reversion_count should be 0
+    And the retrieved plan last_completed_step should be -1
+    And the retrieved plan last_checkpoint_id should be None
+
+  Scenario: Resume fields persist across database reconnection
+    Given the plan persistence database is file-based
+    And a prerequisite action "local/persist-action" exists in the database
+    And a new lifecycle plan with ID "01HV000000000000000000RP06"
+    And the plan has reversion_count 3
+    And the plan has last_completed_step 5
+    And the plan has last_checkpoint_id "01HV000000000000000000CP03"
+    When I persist the plan via the plan repository
+    And I close and reopen the persistence database
+    And I retrieve the plan by ID "01HV000000000000000000RP06"
+    Then the retrieved plan reversion_count should be 3
+    And the retrieved plan last_completed_step should be 5
+    And the retrieved plan last_checkpoint_id should be "01HV000000000000000000CP03"
+
+  Scenario: Updated reversion_count persists after plan update
+    Given a new lifecycle plan with ID "01HV000000000000000000RP07"
+    And the plan has reversion_count 0
+    When I persist the plan via the plan repository
+    And I update the plan reversion_count to 2
+    And I retrieve the plan by ID "01HV000000000000000000RP07"
+    Then the retrieved plan reversion_count should be 2

features/steps/plan_resume_fields_persistence_steps.py

@@ -0,0 +1,113 @@
+"""Step definitions for plan resume fields persistence feature.
+
+Tests round-trip persistence of reversion_count, last_completed_step,
+and last_checkpoint_id through the LifecyclePlanRepository.
+
+Background steps (fresh in-memory DB, prerequisite action, file-based DB,
+close-and-reopen, new lifecycle plan, persist plan) are reused from
+plan_persistence_steps.py — they share the same step text so Behave
+automatically picks them up.
+"""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from behave.runner import Context
+
+# ---------------------------------------------------------------------------
+# Given: mutate the plan under test
+# ---------------------------------------------------------------------------
+
+
+@given("the plan has reversion_count {count:d}")
+def step_plan_has_reversion_count(context: Context, count: int) -> None:
+    """Set reversion_count on the plan being built."""
+    context._pp_plan = context._pp_plan.model_copy(update={"reversion_count": count})
+
+
+@given("the plan has last_completed_step {step:d}")
+def step_plan_has_last_completed_step(context: Context, step: int) -> None:
+    """Set last_completed_step on the plan being built."""
+    context._pp_plan = context._pp_plan.model_copy(update={"last_completed_step": step})
+
+
+@given('the plan has last_checkpoint_id "{checkpoint_id}"')
+def step_plan_has_last_checkpoint_id(context: Context, checkpoint_id: str) -> None:
+    """Set last_checkpoint_id on the plan being built."""
+    context._pp_plan = context._pp_plan.model_copy(
+        update={"last_checkpoint_id": checkpoint_id}
+    )
+
+
+# ---------------------------------------------------------------------------
+# When: retrieve the plan
+# ---------------------------------------------------------------------------
+
+
+@when('I retrieve the plan by ID "{plan_id}"')
+def step_retrieve_plan_by_id(context: Context, plan_id: str) -> None:
+    """Retrieve the plan from the repository and store it for assertions."""
+    context.retrieved_plan = context._pp_plan_repo.get(plan_id)
+    assert context.retrieved_plan is not None, (
+        f"Plan {plan_id!r} not found in repository"
+    )
+
+
+# ---------------------------------------------------------------------------
+# When: update reversion_count on an already-persisted plan
+# ---------------------------------------------------------------------------
+
+
+@when("I update the plan reversion_count to {count:d}")
+def step_update_plan_reversion_count(context: Context, count: int) -> None:
+    """Update reversion_count on the persisted plan.
+
+    Uses ``context._pp_plan`` (the plan that was just persisted) so this
+    step can be called immediately after "I persist the plan via the plan
+    repository" without a prior retrieve step.
+    """
+    updated = context._pp_plan.model_copy(update={"reversion_count": count})
+    context._pp_plan_repo.update(updated)
+    context._pp_session.commit()
+
+
+# ---------------------------------------------------------------------------
+# Then: assert resume field values
+# ---------------------------------------------------------------------------
+
+
+@then("the retrieved plan reversion_count should be {count:d}")
+def step_retrieved_plan_reversion_count(context: Context, count: int) -> None:
+    """Assert the retrieved plan's reversion_count."""
+    assert context.retrieved_plan.reversion_count == count, (
+        f"Expected reversion_count={count}, got {context.retrieved_plan.reversion_count}"
+    )
+
+
+@then("the retrieved plan last_completed_step should be {step:d}")
+def step_retrieved_plan_last_completed_step(context: Context, step: int) -> None:
+    """Assert the retrieved plan's last_completed_step."""
+    assert context.retrieved_plan.last_completed_step == step, (
+        f"Expected last_completed_step={step}, "
+        f"got {context.retrieved_plan.last_completed_step}"
+    )
+
+
+@then('the retrieved plan last_checkpoint_id should be "{checkpoint_id}"')
+def step_retrieved_plan_last_checkpoint_id(
+    context: Context, checkpoint_id: str
+) -> None:
+    """Assert the retrieved plan's last_checkpoint_id matches the expected value."""
+    assert context.retrieved_plan.last_checkpoint_id == checkpoint_id, (
+        f"Expected last_checkpoint_id={checkpoint_id!r}, "
+        f"got {context.retrieved_plan.last_checkpoint_id!r}"
+    )
+
+
+@then("the retrieved plan last_checkpoint_id should be None")
+def step_retrieved_plan_last_checkpoint_id_none(context: Context) -> None:
+    """Assert the retrieved plan's last_checkpoint_id is None."""
+    assert context.retrieved_plan.last_checkpoint_id is None, (
+        f"Expected last_checkpoint_id=None, "
+        f"got {context.retrieved_plan.last_checkpoint_id!r}"
+    )

src/cleveragents/infrastructure/database/models.py

@@ -639,6 +639,13 @@ class LifecyclePlanModel(Base):  # type: ignore[misc]
     )
     attempt = Column(Integer, nullable=False, default=1)
 
+    # Resume tracking fields
+    reversion_count = Column(Integer, nullable=False, default=0, server_default="0")
+    last_completed_step = Column(
+        Integer, nullable=False, default=-1, server_default="-1"
+    )
+    last_checkpoint_id = Column(String(26), nullable=True)
+
     # Rendered description and DoD
     description = Column(Text, nullable=False)
     definition_of_done = Column(Text, nullable=True)
@@ -1015,6 +1022,9 @@ class LifecyclePlanModel(Base):  # type: ignore[misc]
             tags=tags_list,
             reusable=cast(bool, self.reusable),
             read_only=cast(bool, self.read_only),
+            reversion_count=cast(int, self.reversion_count),
+            last_completed_step=cast(int, self.last_completed_step),
+            last_checkpoint_id=cast("str | None", self.last_checkpoint_id),
             execution_environment=cast("str | None", self.execution_environment),
             execution_env_priority=(
                 ExecutionEnvPriority(cast(str, self.execution_env_priority))
@@ -1076,6 +1086,9 @@ class LifecyclePlanModel(Base):  # type: ignore[misc]
             phase=plan.phase.value if hasattr(plan.phase, "value") else plan.phase,
             processing_state=state_str,
             attempt=plan.identity.attempt,
+            reversion_count=plan.reversion_count,
+            last_completed_step=plan.last_completed_step,
+            last_checkpoint_id=plan.last_checkpoint_id,
             description=plan.description,
             definition_of_done=plan.definition_of_done,
             strategy_actor=plan.strategy_actor,

src/cleveragents/infrastructure/database/repositories.py

@@ -1400,6 +1400,11 @@ class LifecyclePlanRepository:
             )
             row.decision_root_id = getattr(plan, "decision_root_id", None)  # type: ignore[assignment]
 
+            # Resume tracking fields
+            row.reversion_count = plan.reversion_count  # type: ignore[assignment]
+            row.last_completed_step = plan.last_completed_step  # type: ignore[assignment]
+            row.last_checkpoint_id = plan.last_checkpoint_id  # type: ignore[assignment]
+
             row.reusable = plan.reusable  # type: ignore[assignment]
             row.read_only = plan.read_only  # type: ignore[assignment]
             row.execution_environment = plan.execution_environment  # type: ignore[assignment]