features/db_migration_lifecycle.feature

@@ -0,0 +1,50 @@
+Feature: Database migration lifecycle
+  As a CleverAgents administrator
+  I want to manage database schema via Alembic migrations
+  So that I can safely evolve the schema with rollback capability
+
+  Background:
+    Given a fresh in-memory database for migration lifecycle testing
+
+  Scenario: Apply all migrations forward on a fresh database
+    When I run all migrations forward to head
+    Then the database should have all expected tables
+    And the current revision should be the head revision
+
+  Scenario: Roll back all migrations to base
+    Given all migrations have been applied
+    When I downgrade all migrations to base
+    Then the current revision should be None
+    And the database should have no application tables
+
+  Scenario: Round-trip forward then rollback to initial
+    Given all migrations have been applied
+    When I downgrade to the initial migration "001_initial_schema"
+    Then the current revision should be "001_initial_schema"
+    When I upgrade back to head
+    Then the current revision should be the head revision
+
+  Scenario: CLI db upgrade command applies migrations
+    When I invoke the db upgrade CLI command
+    Then the CLI should report a successful upgrade
+    And the database should have all expected tables
+
+  Scenario: CLI db current command shows revision
+    Given all migrations have been applied
+    When I invoke the db current CLI command
+    Then the CLI should display the current revision
+
+  Scenario: CLI db downgrade command rolls back
+    Given all migrations have been applied
+    When I invoke the db downgrade CLI command with revision "m6_004_container_metadata_column"
+    Then the CLI should report a successful downgrade
+
+  Scenario: Stamp logic detects pre-Alembic database
+    Given a database with legacy tables but no alembic_version
+    When I run init_or_upgrade on the legacy database
+    Then the database should be stamped with alembic_version
+    And the database revision should be at head
+
+  Scenario: init_database creates a valid schema
+    When I call init_database with an in-memory URL
+    Then the resulting engine should have a valid schema

features/steps/db_migration_lifecycle_steps.py

@@ -0,0 +1,346 @@
+"""Step definitions for db_migration_lifecycle.feature.
+
+Exercises the full Alembic migration lifecycle: forward application,
+rollback, round-trip testing, CLI wrappers, stamp logic, and the
+init_database function.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+from typing import Any
+from unittest.mock import patch
+
+from behave import given, then, when
+
+
+def _create_temp_db_path() -> str:
+    """Create a temporary file path for a SQLite database.
+
+    Returns the path as a string; the caller is responsible for cleanup.
+    """
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    return path
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+# Tables that are created by the Alembic migration chain.
+# NOTE: ``llm_traces`` and ``audit_log`` are defined as ORM models but
+# do not have dedicated migration scripts yet — they were historically
+# created only via ``Base.metadata.create_all()``.  They are excluded
+# from this set until their migration scripts are added.
+_EXPECTED_TABLES = {
+    "projects",
+    "plans",
+    "contexts",
+    "changes",
+    "debug_attempts",
+    "actors",
+    "actions",
+    "action_invariants",
+    "action_arguments",
+    "v3_plans",
+    "plan_projects",
+    "plan_arguments",
+    "plan_invariants",
+    "ns_projects",
+    "project_resource_links",
+    "resource_types",
+    "resources",
+    "resource_edges",
+    "resource_links",
+    "tools",
+    "tool_resource_bindings",
+    "validation_attachments",
+    "sessions",
+    "session_messages",
+    "automation_profiles",
+    "skills",
+    "skill_items",
+    "locks",
+    "decisions",
+    "decision_dependencies",
+    "checkpoint_metadata",
+    "changeset_entries",
+    "tool_invocations",
+    "async_jobs",
+    "repo_indexes",
+    "indexed_files",
+}
+
+
+def _get_table_names(engine: Any) -> set[str]:
+    from sqlalchemy import inspect as sa_inspect
+
+    inspector = sa_inspect(engine)
+    return set(inspector.get_table_names())
+
+
+def _make_engine(url: str = "sqlite:///:memory:") -> Any:
+    from sqlalchemy import create_engine
+
+    return create_engine(url, connect_args={"check_same_thread": False})
+
+
+# ---------------------------------------------------------------------------
+# Background
+# ---------------------------------------------------------------------------
+
+
+@given("a fresh in-memory database for migration lifecycle testing")
+def step_fresh_in_memory_db(context: Any) -> None:
+    context.db_url = "sqlite:///:memory:"
+    context.engine = _make_engine(context.db_url)
+    # Ensure the migration runner module's engine cache is clean for this test
+    from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
+
+    MEMORY_ENGINES.pop(context.db_url, None)
+
+
+# ---------------------------------------------------------------------------
+# Given
+# ---------------------------------------------------------------------------
+
+
+@given("all migrations have been applied")
+def step_all_migrations_applied(context: Any) -> None:
+    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+
+    runner = MigrationRunner(context.db_url)
+    runner.run_migrations(engine=context.engine)
+    context.runner = runner
+
+
+@given("a database with legacy tables but no alembic_version")
+def step_legacy_db_no_alembic(context: Any) -> None:
+    from cleveragents.infrastructure.database.models import Base
+
+    # Create all tables via create_all (simulates pre-Alembic state)
+    Base.metadata.create_all(context.engine)
+    # Verify alembic_version does NOT exist yet
+    tables = _get_table_names(context.engine)
+    assert "alembic_version" not in tables
+
+
+# ---------------------------------------------------------------------------
+# When
+# ---------------------------------------------------------------------------
+
+
+@when("I run all migrations forward to head")
+def step_run_migrations_forward(context: Any) -> None:
+    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+
+    runner = MigrationRunner(context.db_url)
+    runner.run_migrations(engine=context.engine)
+    context.runner = runner
+
+
+@when("I downgrade all migrations to base")
+def step_downgrade_to_base(context: Any) -> None:
+    from alembic import command
+
+    runner = context.runner
+    with context.engine.connect() as conn:
+        runner.alembic_cfg.attributes["connection"] = conn
+        try:
+            command.downgrade(runner.alembic_cfg, "base")
+            conn.commit()
+        finally:
+            runner.alembic_cfg.attributes.pop("connection", None)
+
+
+@when('I downgrade to the initial migration "{revision}"')
+def step_downgrade_to_revision(context: Any, revision: str) -> None:
+    from alembic import command
+
+    runner = context.runner
+    with context.engine.connect() as conn:
+        runner.alembic_cfg.attributes["connection"] = conn
+        try:
+            command.downgrade(runner.alembic_cfg, revision)
+            conn.commit()
+        finally:
+            runner.alembic_cfg.attributes.pop("connection", None)
+
+
+@when("I upgrade back to head")
+def step_upgrade_to_head(context: Any) -> None:
+    runner = context.runner
+    runner.run_migrations(engine=context.engine)
+
+
+@when("I invoke the db upgrade CLI command")
+def step_invoke_db_upgrade(context: Any) -> None:
+
+    from cleveragents.cli.main import main
+
+    # CLI commands create their own engine, so we must use a file-based
+    # DB to persist across engine instances.
+    db_path = _create_temp_db_path()
+    context.cli_db_path = db_path
+    cli_url = f"sqlite:///{db_path}"
+    with patch.dict(os.environ, {"CLEVERAGENTS_DATABASE_URL": cli_url}):
+        context.cli_exit_code = main(["db", "upgrade"])
+    # Re-read the DB to verify tables
+    context.engine = _make_engine(cli_url)
+
+
+@when("I invoke the db current CLI command")
+def step_invoke_db_current(context: Any) -> None:
+
+    from cleveragents.cli.main import main
+
+    # Use a file-based DB so the CLI sees the migrated schema
+    db_path = _create_temp_db_path()
+    cli_url = f"sqlite:///{db_path}"
+    # First apply migrations so there's something to report
+    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+
+    MigrationRunner(cli_url).init_or_upgrade()
+    with patch.dict(os.environ, {"CLEVERAGENTS_DATABASE_URL": cli_url}):
+        context.cli_exit_code = main(["db", "current"])
+
+
+@when('I invoke the db downgrade CLI command with revision "{revision}"')
+def step_invoke_db_downgrade(context: Any, revision: str) -> None:
+
+    from cleveragents.cli.main import main
+    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+
+    # Use a file-based DB so the CLI sees the migrated schema
+    db_path = _create_temp_db_path()
+    cli_url = f"sqlite:///{db_path}"
+    MigrationRunner(cli_url).init_or_upgrade()
+    with patch.dict(os.environ, {"CLEVERAGENTS_DATABASE_URL": cli_url}):
+        context.cli_exit_code = main(["db", "downgrade", revision])
+
+
+@when("I run init_or_upgrade on the legacy database")
+def step_init_or_upgrade_legacy(context: Any) -> None:
+    from cleveragents.infrastructure.database.engine_cache import MEMORY_ENGINES
+    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+
+    MEMORY_ENGINES[context.db_url] = context.engine
+    try:
+        runner = MigrationRunner(context.db_url)
+        runner.init_or_upgrade()
+        context.runner = runner
+    finally:
+        MEMORY_ENGINES.pop(context.db_url, None)
+
+
+@when("I call init_database with an in-memory URL")
+def step_call_init_database(context: Any) -> None:
+
+    from cleveragents.infrastructure.database.models import init_database
+
+    # Use a file-based temporary database so the engine can be reopened
+    # and the tables inspected after init_database returns.
+    db_path = _create_temp_db_path()
+    file_url = f"sqlite:///{db_path}"
+    context.result_engine = init_database(file_url)
+
+
+# ---------------------------------------------------------------------------
+# Then
+# ---------------------------------------------------------------------------
+
+
+@then("the database should have all expected tables")
+def step_db_has_expected_tables(context: Any) -> None:
+    tables = _get_table_names(context.engine)
+    missing = _EXPECTED_TABLES - tables
+    assert not missing, f"Missing tables: {missing}"
+
+
+@then("the current revision should be the head revision")
+def step_current_is_head(context: Any) -> None:
+    from alembic.script import ScriptDirectory
+
+    runner = context.runner
+    head = ScriptDirectory.from_config(runner.alembic_cfg).get_current_head()
+
+    from alembic.runtime.migration import MigrationContext
+
+    with context.engine.connect() as conn:
+        current = MigrationContext.configure(conn).get_current_revision()
+
+    assert current == head, f"Expected {head}, got {current}"
+
+
+@then("the current revision should be None")
+def step_current_is_none(context: Any) -> None:
+    from alembic.runtime.migration import MigrationContext
+
+    with context.engine.connect() as conn:
+        current = MigrationContext.configure(conn).get_current_revision()
+    assert current is None, f"Expected None, got {current}"
+
+
+@then("the database should have no application tables")
+def step_db_has_no_app_tables(context: Any) -> None:
+    tables = _get_table_names(context.engine)
+    # Only alembic_version may remain after full downgrade
+    app_tables = tables - {"alembic_version"}
+    assert not app_tables, f"Unexpected tables remain: {app_tables}"
+
+
+@then('the current revision should be "{expected}"')
+def step_current_revision_is(context: Any, expected: str) -> None:
+    from alembic.runtime.migration import MigrationContext
+
+    with context.engine.connect() as conn:
+        current = MigrationContext.configure(conn).get_current_revision()
+    assert current == expected, f"Expected {expected}, got {current}"
+
+
+@then("the CLI should report a successful upgrade")
+def step_cli_upgrade_success(context: Any) -> None:
+    assert context.cli_exit_code == 0, f"CLI exited with code {context.cli_exit_code}"
+
+
+@then("the CLI should display the current revision")
+def step_cli_displays_revision(context: Any) -> None:
+    assert context.cli_exit_code == 0, f"CLI exited with code {context.cli_exit_code}"
+
+
+@then("the CLI should report a successful downgrade")
+def step_cli_downgrade_success(context: Any) -> None:
+    assert context.cli_exit_code == 0, f"CLI exited with code {context.cli_exit_code}"
+
+
+@then("the database should be stamped with alembic_version")
+def step_db_stamped_with_alembic(context: Any) -> None:
+    tables = _get_table_names(context.engine)
+    assert "alembic_version" in tables, f"alembic_version not found in {tables}"
+
+
+@then("the database revision should be at head")
+def step_db_revision_at_head(context: Any) -> None:
+    from alembic.runtime.migration import MigrationContext
+    from alembic.script import ScriptDirectory
+
+    runner = context.runner
+    head = ScriptDirectory.from_config(runner.alembic_cfg).get_current_head()
+    with context.engine.connect() as conn:
+        current = MigrationContext.configure(conn).get_current_revision()
+    assert current == head, f"Expected head {head}, got {current}"
+
+
+@then("the resulting engine should have a valid schema")
+def step_engine_has_valid_schema(context: Any) -> None:
+    tables = _get_table_names(context.result_engine)
+    # At minimum, core tables should exist
+    assert "actors" in tables or "projects" in tables, f"Got tables: {tables}"
+
+
+@then("the alembic_version table should exist")
+def step_alembic_version_exists(context: Any) -> None:
+    tables = _get_table_names(context.result_engine)
+    assert "alembic_version" in tables, f"alembic_version not in {tables}"

features/steps/migration_runner_steps.py

@@ -20,6 +20,7 @@ class FakeConnection:
         self.entered = False
         self.exit_called = False
         self.closed_direct = False
+        self.committed = False
 
     def __enter__(self) -> FakeConnection:
         self.entered = True
@@ -32,6 +33,9 @@ class FakeConnection:
     def close(self) -> None:
         self.closed_direct = True
 
+    def commit(self) -> None:
+        self.committed = True
+
 
 class FakeEngine:
     def __init__(self) -> None:
@@ -230,7 +234,7 @@ def step_when_legacy_stamping(context) -> None:
 @then("the stamp command should run using the active connection")
 def step_then_stamp_uses_connection(context) -> None:
     assert len(context.legacy_stamp_calls) == 1
-    assert context.legacy_stamp_revision == "001_initial_schema"
+    assert context.legacy_stamp_revision == "head"
     assert context.legacy_stamp_connection_has_attr is True
     assert len(context.legacy_upgrade_calls) == 0
     assert "connection" not in context.legacy_connection_attr_after

src/cleveragents/cli/commands/db.py

@@ -0,0 +1,174 @@
+"""Database management commands for CleverAgents CLI.
+
+The ``agents db`` command group exposes Alembic-managed schema operations
+so that administrators can inspect, apply, and roll back database migrations.
+
+## Commands
+
+| Command                           | Description                              |
+|-----------------------------------|------------------------------------------|
+| ``agents db migrate``             | Generate a new migration revision        |
+| ``agents db upgrade [REVISION]``  | Apply pending migrations (default: head) |
+| ``agents db downgrade REVISION``  | Roll back to a specific revision         |
+| ``agents db current``             | Show the current migration revision      |
+| ``agents db history``             | Show migration history                   |
+
+These commands delegate to the :class:`MigrationRunner` which wraps the
+Alembic ``command`` API with project-aware database URL resolution.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Annotated
+
+import typer
+
+from cleveragents.cli.formatting import format_output
+
+if TYPE_CHECKING:
+    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+
+app = typer.Typer(help="Database migration management.")
+_FORMAT_HELP = "Output format: json, yaml, plain, table, or rich (default: rich)"
+
+
+def _get_runner() -> MigrationRunner:
+    """Build a :class:`MigrationRunner` using the resolved database URL."""
+    from cleveragents.application.container import get_database_url
+    from cleveragents.infrastructure.database.migration_runner import MigrationRunner
+
+    return MigrationRunner(get_database_url())
+
+
+@app.command()
+def migrate(
+    message: Annotated[
+        str,
+        typer.Option(
+            "--message",
+            "-m",
+            help="Short description for the migration revision",
+        ),
+    ] = "auto",
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Generate a new Alembic migration revision (autogenerate).
+
+    Compares the current ORM models against the database schema and
+    produces a migration script for any detected differences.
+    """
+    from alembic import command
+
+    runner = _get_runner()
+    command.revision(runner.alembic_cfg, message=message, autogenerate=True)
+    typer.echo(f"Migration revision created: {message}")
+
+
+@app.command()
+def upgrade(
+    revision: Annotated[
+        str,
+        typer.Argument(help="Target revision (default: 'head')"),
+    ] = "head",
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Apply pending database migrations up to REVISION (default: head).
+
+    On a fresh database this creates the full schema.  On an existing
+    database it applies only the migrations that have not yet been run.
+    """
+    runner = _get_runner()
+
+    if revision == "head":
+        runner.init_or_upgrade()
+    else:
+        from alembic import command
+
+        command.upgrade(runner.alembic_cfg, revision)
+
+    current = runner.get_current_revision()
+    data = {"status": "ok", "current_revision": current}
+    if fmt == "rich":
+        typer.echo(f"Database upgraded to revision: {current}")
+    else:
+        typer.echo(format_output(data, fmt))
+
+
+@app.command()
+def downgrade(
+    revision: Annotated[
+        str,
+        typer.Argument(help="Target revision to downgrade to (e.g. '-1' or a rev id)"),
+    ],
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Roll the database back to REVISION.
+
+    Accepts Alembic revision identifiers such as a hex revision id,
+    ``-1`` (one step back), or ``base`` (roll back everything).
+
+    WARNING: Downgrading may cause data loss.  Always back up the
+    database before running this command.
+    """
+    from alembic import command
+
+    runner = _get_runner()
+    command.downgrade(runner.alembic_cfg, revision)
+
+    current = runner.get_current_revision()
+    data = {"status": "ok", "current_revision": current}
+    if fmt == "rich":
+        typer.echo(f"Database downgraded to revision: {current}")
+    else:
+        typer.echo(format_output(data, fmt))
+
+
+@app.command()
+def current(
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Show the current migration revision of the database."""
+    runner = _get_runner()
+    rev = runner.get_current_revision()
+    pending = runner.get_pending_migrations()
+
+    data = {
+        "current_revision": rev,
+        "pending_count": len(pending),
+        "pending_revisions": pending,
+    }
+
+    if fmt == "rich":
+        typer.echo(f"Current revision : {rev or '(none)'}")
+        typer.echo(f"Pending migrations: {len(pending)}")
+        if pending:
+            for p in pending:
+                typer.echo(f"  - {p}")
+    else:
+        typer.echo(format_output(data, fmt))
+
+
+@app.command()
+def history(
+    fmt: Annotated[str, typer.Option("--format", "-f", help=_FORMAT_HELP)] = "rich",
+) -> None:
+    """Show the migration revision history."""
+    from alembic import command
+
+    runner = _get_runner()
+
+    if fmt == "rich":
+        command.history(runner.alembic_cfg)
+    else:
+        from alembic.script import ScriptDirectory
+
+        script = ScriptDirectory.from_config(runner.alembic_cfg)
+        revisions = [
+            {
+                "revision": rev.revision,
+                "down_revision": rev.down_revision,
+                "description": rev.doc or "",
+            }
+            for rev in script.walk_revisions()
+        ]
+        typer.echo(format_output(revisions, fmt))

src/cleveragents/cli/main.py

@@ -96,6 +96,7 @@ def _register_subcommands() -> None:
             validation,
         )
         from cleveragents.cli.commands.auto_debug import app as auto_debug_app
+        from cleveragents.cli.commands.db import app as db_app
         from cleveragents.cli.commands.repl import _repl_app
         from cleveragents.cli.commands.server import app as server_app
     except Exception as exc:  # pragma: no cover
@@ -130,6 +131,11 @@ def _register_subcommands() -> None:
         help="Manage actions (reusable plan templates) for the v3 plan lifecycle",
     )
     app.add_typer(auto_debug_app, name="auto-debug", help="Auto-debug operations")
+    app.add_typer(
+        db_app,
+        name="db",
+        help="Database migration management (Alembic)",
+    )
     app.add_typer(
         resource.app,
         name="resource",
@@ -230,6 +236,7 @@ def _print_basic_help() -> None:
     typer.echo("  tell         Create a plan (shortcut)")
     typer.echo("  build        Build the current plan")
     typer.echo("  apply        Apply plan changes")
+    typer.echo("  db           Database migration management")
     typer.echo("  auto-debug   Auto-debug operations")
     typer.echo("  repl         Interactive REPL session")
     typer.echo("  tui          Textual terminal UI")
@@ -605,6 +612,7 @@ def main(args: list[str] | None = None) -> int:
             "plan",
             "actor",
             "action",  # v3 plan lifecycle actions
+            "db",  # Database migration management
             "resource",  # Resource registry management
             "skill",  # Skill management
             "lsp",  # LSP server management

src/cleveragents/infrastructure/database/migration_runner.py

@@ -269,16 +269,20 @@ class MigrationRunner:
                     # on existing databases
                     if len(tables) > 0:
                         # Database has tables but no alembic_version
-                        # This might be a legacy database
-                        # For now, we'll stamp it with the initial migration
-                        # Pass the connection to stamp command as well
+                        # This is a pre-Alembic legacy database.
+                        # Stamp it at the initial migration to record the
+                        # existing schema, then run any subsequent migrations
+                        # so the database reaches the current head.
                         require_user_approval(
                             "Existing database has tables but no migration history. "
                             "Stamp the initial schema before continuing."
                         )
                         self.alembic_cfg.attributes["connection"] = connection
                         try:
-                            command.stamp(self.alembic_cfg, "001_initial_schema")
+                            # Stamp at head since create_all creates all
+                            # tables matching the current ORM models.
+                            command.stamp(self.alembic_cfg, "head")
+                            connection.commit()
                         finally:
                             self.alembic_cfg.attributes.pop("connection", None)
                     else:

vulture_whitelist.py

@@ -38,6 +38,12 @@ _action_spec_dict  # noqa: B018, F821
 _plan_spec_dict  # noqa: B018, F821
 _FORMAT_HELP  # noqa: B018, F821
 
+# Database CLI commands — public API registered via Typer (db.py)
+migrate  # noqa: B018, F821
+upgrade  # noqa: B018, F821
+downgrade  # noqa: B018, F821
+history  # noqa: B018, F821
+
 # Resource repository error classes — public API for service layer
 ResourceTypeNotFoundError  # noqa: B018, F821
 ResourceNotFoundRepoError  # noqa: B018, F821