benchmarks/db_resource_bench.py

@@ -0,0 +1,180 @@
+"""ASV benchmarks for database resource types and transaction sandbox.
+
+Measures the performance of:
+- Database type definition lookup
+- Connection argument resolution
+- Connection validation (SQLite)
+- TransactionSandbox create/commit/rollback lifecycle
+- SandboxFactory routing for transaction_rollback
+"""
+
+from __future__ import annotations
+
+import importlib
+import os
+import sqlite3
+import sys
+import tempfile
+from pathlib import Path
+
+# Ensure the local *source* tree is importable even when ASV has an
+# older build of the package installed.
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+# Force-reload so ASV picks up the source tree version.
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.infrastructure.sandbox.factory import (  # noqa: E402
+    SandboxFactory,
+)
+from cleveragents.infrastructure.sandbox.transaction_sandbox import (  # noqa: E402
+    TransactionSandbox,
+)
+from cleveragents.resource.handlers.database import (  # noqa: E402
+    DATABASE_TYPE_DEFS,
+    DatabaseResourceHandler,
+    resolve_connection_args,
+    validate_connection,
+)
+
+
+# ---------------------------------------------------------------------------
+# Type definition benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TypeDefinitionSuite:
+    """Benchmark database type definition operations."""
+
+    timeout = 60
+
+    def time_lookup_all_type_defs(self) -> None:
+        """Time iterating all database type definitions."""
+        for td in DATABASE_TYPE_DEFS:
+            _ = td["name"]
+
+    def time_resolve_postgres_args(self) -> None:
+        """Time resolving postgres connection args."""
+        resolve_connection_args(
+            {"host": "db.example.com", "port": 5432, "dbname": "mydb"},
+            "postgres",
+        )
+
+    def time_resolve_sqlite_args(self) -> None:
+        """Time resolving sqlite connection args."""
+        resolve_connection_args({"path": "/tmp/test.db"}, "sqlite")
+
+    def time_handler_instantiation(self) -> None:
+        """Time creating a DatabaseResourceHandler."""
+        DatabaseResourceHandler()
+
+
+# ---------------------------------------------------------------------------
+# Connection validation benchmarks
+# ---------------------------------------------------------------------------
+
+
+class ConnectionValidationSuite:
+    """Benchmark connection validation operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Create a temp SQLite database for benchmarking."""
+        fd, self.db_path = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("CREATE TABLE bench (id INTEGER PRIMARY KEY)")
+        conn.commit()
+        conn.close()
+
+    def teardown(self) -> None:
+        """Remove the temp database."""
+        if hasattr(self, "db_path") and os.path.exists(self.db_path):
+            os.unlink(self.db_path)
+
+    def time_validate_sqlite_file(self) -> None:
+        """Time validating a file-based SQLite connection."""
+        validate_connection("sqlite", {"path": self.db_path})
+
+    def time_validate_sqlite_memory(self) -> None:
+        """Time validating an in-memory SQLite connection."""
+        validate_connection("sqlite", {"path": ":memory:"})
+
+    def time_validate_postgres_stub(self) -> None:
+        """Time validating a postgres stub (no driver)."""
+        validate_connection(
+            "postgres",
+            {"host": "localhost", "port": 5432, "dbname": "bench"},
+        )
+
+
+# ---------------------------------------------------------------------------
+# Transaction sandbox benchmarks
+# ---------------------------------------------------------------------------
+
+
+class TransactionSandboxSuite:
+    """Benchmark TransactionSandbox lifecycle operations."""
+
+    timeout = 60
+
+    def setup(self) -> None:
+        """Create a temp SQLite database for benchmarking."""
+        fd, self.db_path = tempfile.mkstemp(suffix=".db")
+        os.close(fd)
+        conn = sqlite3.connect(self.db_path)
+        conn.execute("CREATE TABLE bench (id INTEGER PRIMARY KEY, val TEXT)")
+        conn.commit()
+        conn.close()
+
+    def teardown(self) -> None:
+        """Remove the temp database."""
+        if hasattr(self, "db_path") and os.path.exists(self.db_path):
+            os.unlink(self.db_path)
+
+    def time_create_and_cleanup(self) -> None:
+        """Time sandbox create + cleanup cycle."""
+        sandbox = TransactionSandbox(resource_id="bench-1", original_path=self.db_path)
+        sandbox.create(plan_id="BENCH-PLAN")
+        sandbox.cleanup()
+
+    def time_create_insert_commit(self) -> None:
+        """Time full sandbox lifecycle: create, insert, commit, cleanup."""
+        sandbox = TransactionSandbox(resource_id="bench-2", original_path=self.db_path)
+        sandbox.create(plan_id="BENCH-PLAN")
+        sandbox.execute("INSERT INTO bench (id, val) VALUES (1, 'bench')")
+        sandbox.commit("bench commit")
+        sandbox.cleanup()
+
+    def time_create_insert_rollback(self) -> None:
+        """Time sandbox lifecycle with rollback."""
+        sandbox = TransactionSandbox(resource_id="bench-3", original_path=self.db_path)
+        sandbox.create(plan_id="BENCH-PLAN")
+        sandbox.execute("INSERT INTO bench (id, val) VALUES (2, 'rollback')")
+        sandbox.rollback()
+        sandbox.cleanup()
+
+
+# ---------------------------------------------------------------------------
+# Factory routing benchmarks
+# ---------------------------------------------------------------------------
+
+
+class FactoryRoutingSuite:
+    """Benchmark SandboxFactory routing for database strategies."""
+
+    timeout = 60
+
+    def time_factory_transaction_rollback(self) -> None:
+        """Time factory creating a TransactionSandbox."""
+        factory = SandboxFactory()
+        factory.create_sandbox(
+            resource_id="bench-factory",
+            original_path=":memory:",
+            sandbox_strategy="transaction_rollback",
+        )

features/consolidated_sandbox.feature

@@ -738,10 +738,10 @@ Feature: Consolidated Sandbox
     # --- Unimplemented strategies ---
 
 
-  Scenario: The transaction rollback strategy is not yet available
+  Scenario: The transaction rollback strategy creates a TransactionSandbox
     Given the sandbox factory is available
     When a sandbox is requested for resource "db-main" at "postgres://db/main" using the transaction rollback strategy
-    Then the factory should indicate the strategy is not yet implemented
+    Then the factory should produce a transaction sandbox in the pending state
 
 
   Scenario: The snapshot strategy is not yet available

features/database_resources.feature

@@ -0,0 +1,228 @@
+Feature: Database resource types and transaction sandbox
+  As a CleverAgents developer
+  I want database resource types with transaction-based sandboxing
+  So that I can manage database resources with isolation and rollback
+
+  # ---- Resource Type Definitions ----
+
+  Scenario: Postgres resource type is defined
+    Given the database type definitions are loaded
+    Then the "postgres" type definition should exist
+    And the "postgres" type should have sandbox strategy "transaction_rollback"
+    And the "postgres" type should have a "host" cli argument
+    And the "postgres" type should have a "port" cli argument
+    And the "postgres" type should have a "dbname" cli argument
+    And the "postgres" type should have a "user" cli argument
+    And the "postgres" type should have a "password" cli argument
+    And the "postgres" type should have a "connection-string" cli argument
+
+  Scenario: MySQL resource type is defined
+    Given the database type definitions are loaded
+    Then the "mysql" type definition should exist
+    And the "mysql" type should have sandbox strategy "transaction_rollback"
+    And the "mysql" type should have a "host" cli argument
+    And the "mysql" type should have a "port" cli argument
+
+  Scenario: SQLite resource type is defined
+    Given the database type definitions are loaded
+    Then the "sqlite" type definition should exist
+    And the "sqlite" type should have sandbox strategy "transaction_rollback"
+    And the "sqlite" type should have a "path" cli argument
+
+  Scenario: DuckDB resource type is defined
+    Given the database type definitions are loaded
+    Then the "duckdb" type definition should exist
+    And the "duckdb" type should have sandbox strategy "transaction_rollback"
+    And the "duckdb" type should have a "path" cli argument
+
+  # ---- Connection Argument Validation ----
+
+  Scenario: Resolve postgres connection args from properties
+    Given postgres connection properties with host "db.example.com" port 5432 dbname "mydb"
+    When I resolve connection args for "postgres"
+    Then the resolved args should have host "db.example.com"
+    And the resolved args should have port 5432
+    And the resolved args should have dbname "mydb"
+
+  Scenario: Resolve postgres connection args from connection string
+    Given postgres properties with connection string "postgresql://admin:secret@db:5432/app"
+    When I resolve connection args for "postgres"
+    Then the resolved args should have connection string "postgresql://admin:secret@db:5432/app"
+
+  Scenario: Resolve postgres credentials from env vars
+    Given postgres properties without explicit credentials
+    And the env var "CLEVERAGENTS_DB_USER" is set to "envuser"
+    And the env var "CLEVERAGENTS_DB_PASSWORD" is set to "envpass"
+    When I resolve connection args for "postgres"
+    Then the resolved args should have user "envuser"
+    And the resolved args should have password "envpass"
+
+  Scenario: Resolve sqlite connection args
+    Given sqlite properties with path "/tmp/test.db"
+    When I resolve connection args for "sqlite"
+    Then the resolved args should have path "/tmp/test.db"
+
+  Scenario: Resolve duckdb connection args with default memory
+    Given duckdb properties with no path
+    When I resolve connection args for "duckdb"
+    Then the resolved args should have path ":memory:"
+
+  # ---- Credential Masking in Error Messages ----
+
+  Scenario: Credential masking in postgres validation message
+    Given postgres connection properties with user "admin" and password "s3cret"
+    When I validate the "postgres" connection
+    Then the validation message should not contain "s3cret"
+    And the validation result should be successful
+
+  Scenario: Connection string credentials are masked
+    Given postgres properties with connection string "postgresql://admin:s3cret@db:5432/app"
+    When I validate the "postgres" connection
+    Then the validation message should not contain "s3cret"
+
+  # ---- Read-Only Enforcement ----
+
+  Scenario: Read-only transaction sandbox rejects writes
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox in read-only mode
+    When I attempt to insert data via the sandbox
+    Then the sandbox should reject the write operation
+
+  # ---- Transaction Sandbox Create / Commit / Rollback ----
+
+  Scenario: Transaction sandbox lifecycle create and commit
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I create the sandbox with plan "PLAN-TX-001"
+    And I insert a row via the sandbox
+    And I commit the sandbox
+    Then the row should be persisted in the database
+
+  Scenario: Transaction sandbox lifecycle create and rollback
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I create the sandbox with plan "PLAN-TX-002"
+    And I insert a row via the sandbox
+    And I rollback the sandbox
+    Then the row should not be persisted in the database
+
+  Scenario: Transaction sandbox cleanup closes connection
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I create the sandbox with plan "PLAN-TX-003"
+    And I cleanup the sandbox
+    Then the sandbox status should be "cleaned_up"
+
+  # ---- SQLite Resource Handler Roundtrip ----
+
+  Scenario: SQLite validate_connection success
+    Given a temporary SQLite database
+    When I validate the sqlite connection
+    Then the sqlite validation should succeed
+
+  Scenario: SQLite validate_connection with memory database
+    When I validate an in-memory sqlite connection
+    Then the sqlite validation should succeed
+
+  # ---- DatabaseResourceHandler Protocol ----
+
+  Scenario: DatabaseResourceHandler satisfies ResourceHandler protocol
+    Given a DatabaseResourceHandler instance
+    Then the handler should satisfy the ResourceHandler protocol
+
+  Scenario: DatabaseResourceHandler uses transaction_rollback strategy
+    Given a DatabaseResourceHandler instance
+    Then the handler default strategy should be "transaction_rollback"
+
+  # ---- SandboxFactory transaction_rollback routing ----
+
+  Scenario: SandboxFactory creates TransactionSandbox for transaction_rollback
+    Given the sandbox factory is available
+    When a sandbox is requested for resource "db-1" at ":memory:" using transaction_rollback
+    Then the created sandbox should be a TransactionSandbox
+
+  Scenario: SandboxFactory reports transaction_rollback as supported
+    Given the sandbox factory is available
+    Then the factory should report "transaction_rollback" as supported
+
+  # ---- TransactionSandbox edge cases ----
+
+  Scenario: TransactionSandbox rejects empty resource_id
+    When I create a TransactionSandbox with empty resource_id
+    Then the edge error should be a ValueError mentioning "resource_id"
+
+  Scenario: TransactionSandbox rejects empty original_path
+    When I create a TransactionSandbox with empty original_path
+    Then the edge error should be a ValueError mentioning "original_path"
+
+  Scenario: TransactionSandbox rejects empty plan_id
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I try to create the sandbox with an empty plan_id
+    Then the edge error should be a ValueError mentioning "plan_id"
+
+  Scenario: TransactionSandbox get_path returns db path
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I create the sandbox with plan "PLAN-PATH"
+    Then get_path should return the database path
+
+  Scenario: TransactionSandbox get_path rejects traversal
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I create the sandbox with plan "PLAN-TRAV"
+    Then get_path with traversal should raise ValueError
+
+  Scenario: TransactionSandbox get_path in wrong state raises
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    Then get_path before create should raise SandboxStateError
+
+  Scenario: TransactionSandbox commit in wrong state raises
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    Then commit before create should raise SandboxStateError
+
+  Scenario: TransactionSandbox rollback in wrong state raises
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    Then rollback before create should raise SandboxStateError
+
+  Scenario: TransactionSandbox cleanup is idempotent
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I create the sandbox with plan "PLAN-IDEM"
+    And I cleanup the sandbox
+    And I cleanup the sandbox again
+    Then the sandbox status should be "cleaned_up"
+
+  Scenario: TransactionSandbox execute in wrong state raises
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    Then execute before create should raise SandboxStateError
+
+  Scenario: TransactionSandbox cleanup with active transaction
+    Given a temporary SQLite database with a test table
+    And a transaction sandbox for the database
+    When I create the sandbox with plan "PLAN-ACT"
+    And I insert a row via the sandbox
+    And I cleanup the sandbox
+    Then the sandbox status should be "cleaned_up"
+
+  Scenario: DuckDB validate_connection stub
+    When I validate a duckdb connection with path ":memory:"
+    Then the duckdb validation should succeed with a stub message
+
+  Scenario: MySQL validate_connection stub
+    Given mysql connection properties with host "db.example.com" and port 3306
+    When I validate the "mysql" connection
+    Then the validation result should be successful
+
+  Scenario: Unknown database type validation
+    When I validate an unknown database type
+    Then the unknown validation should fail
+
+  Scenario: Resolve mysql connection args from connection string
+    Given mysql properties with connection string "mysql://user:pass@host:3306/db"
+    When I resolve connection args for "mysql"
+    Then the resolved args should have connection string "mysql://user:pass@host:3306/db"

features/steps/database_resources_steps.py

@@ -0,0 +1,534 @@
+"""Step definitions for database resource types and transaction sandbox.
+
+Covers:
+- Database type definitions (postgres, mysql, sqlite, duckdb)
+- Connection argument resolution and env var fallback
+- Credential masking in error/validation messages
+- Read-only enforcement
+- Transaction sandbox create/commit/rollback lifecycle
+- SQLite resource handler roundtrip
+- DatabaseResourceHandler protocol conformance
+- SandboxFactory transaction_rollback routing
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import tempfile
+
+from behave import given, then, when
+
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.protocol import SandboxStatus
+from cleveragents.infrastructure.sandbox.transaction_sandbox import (
+    TransactionSandbox,
+)
+from cleveragents.resource.handlers.database import (
+    DATABASE_TYPE_DEFS,
+    DatabaseResourceHandler,
+    resolve_connection_args,
+    validate_connection,
+)
+from cleveragents.resource.handlers.protocol import ResourceHandler
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+
+def _find_type_def(name):
+    """Find a database type definition by name."""
+    for td in DATABASE_TYPE_DEFS:
+        if td["name"] == name:
+            return td
+    return None
+
+
+# ---------------------------------------------------------------------------
+# Resource Type Definitions
+# ---------------------------------------------------------------------------
+
+
+@given("the database type definitions are loaded")
+def given_db_type_defs_loaded(context):
+    """Load database type definitions into context."""
+    context.db_type_defs = {td["name"]: td for td in DATABASE_TYPE_DEFS}
+
+
+@then('the "{name}" type definition should exist')
+def then_type_def_exists(context, name):
+    assert name in context.db_type_defs, f"Type '{name}' not found in definitions"
+
+
+@then('the "{name}" type should have sandbox strategy "{strategy}"')
+def then_type_strategy(context, name, strategy):
+    assert context.db_type_defs[name]["sandbox_strategy"] == strategy
+
+
+@then('the "{name}" type should have a "{arg_name}" cli argument')
+def then_type_has_arg(context, name, arg_name):
+    cli_args = context.db_type_defs[name]["cli_args"]
+    arg_names = [a["name"] for a in cli_args]
+    assert arg_name in arg_names, f"Argument '{arg_name}' not found in {arg_names}"
+
+
+# ---------------------------------------------------------------------------
+# Connection Argument Resolution
+# ---------------------------------------------------------------------------
+
+
+@given(
+    'postgres connection properties with host "{host}" port {port:d} dbname "{dbname}"'
+)
+def given_pg_props(context, host, port, dbname):
+    context.conn_props = {"host": host, "port": port, "dbname": dbname}
+
+
+@given('postgres properties with connection string "{conn_str}"')
+def given_pg_conn_str(context, conn_str):
+    context.conn_props = {"connection-string": conn_str}
+
+
+@given("postgres properties without explicit credentials")
+def given_pg_no_creds(context):
+    context.conn_props = {"host": "localhost", "port": 5432}
+
+
+@given('sqlite properties with path "{path}"')
+def given_sqlite_props(context, path):
+    context.conn_props = {"path": path}
+
+
+@given("duckdb properties with no path")
+def given_duckdb_no_path(context):
+    context.conn_props = {}
+
+
+@when('I resolve connection args for "{resource_type}"')
+def when_resolve_args(context, resource_type):
+    context.resolved_args = resolve_connection_args(context.conn_props, resource_type)
+
+
+@then('the resolved args should have host "{host}"')
+def then_resolved_host(context, host):
+    assert context.resolved_args["host"] == host
+
+
+@then("the resolved args should have port {port:d}")
+def then_resolved_port(context, port):
+    assert context.resolved_args["port"] == port
+
+
+@then('the resolved args should have dbname "{dbname}"')
+def then_resolved_dbname(context, dbname):
+    assert context.resolved_args["dbname"] == dbname
+
+
+@then('the resolved args should have connection string "{conn_str}"')
+def then_resolved_conn_str(context, conn_str):
+    assert context.resolved_args["connection-string"] == conn_str
+
+
+@then('the resolved args should have user "{user}"')
+def then_resolved_user(context, user):
+    args = context.resolved_args
+    assert args["user"] == user, f"Expected user '{user}', got '{args['user']}'"
+    _cleanup_env(context)
+
+
+@then('the resolved args should have password "{password}"')
+def then_resolved_password(context, password):
+    args = context.resolved_args
+    assert args["password"] == password
+    _cleanup_env(context)
+
+
+@then('the resolved args should have path "{path}"')
+def then_resolved_path(context, path):
+    assert context.resolved_args["path"] == path
+
+
+def _cleanup_env(context):
+    """Restore saved environment variables."""
+    saved = getattr(context, "_saved_env", {})
+    for var, old_value in saved.items():
+        if old_value is None:
+            os.environ.pop(var, None)
+        else:
+            os.environ[var] = old_value
+    context._saved_env = {}
+
+
+# ---------------------------------------------------------------------------
+# Credential Masking
+# ---------------------------------------------------------------------------
+
+
+@given('postgres connection properties with user "{user}" and password "{password}"')
+def given_pg_with_creds(context, user, password):
+    context.conn_props = {
+        "host": "localhost",
+        "user": user,
+        "password": password,
+    }
+
+
+@when('I validate the "{resource_type}" connection')
+def when_validate_connection(context, resource_type):
+    resolved = resolve_connection_args(context.conn_props, resource_type)
+    success, message = validate_connection(resource_type, resolved)
+    context.validation_success = success
+    context.validation_message = message
+
+
+@then('the validation message should not contain "{secret}"')
+def then_message_no_secret(context, secret):
+    msg = context.validation_message
+    assert secret not in msg, f"Secret '{secret}' found in validation message: {msg}"
+
+
+@then("the validation result should be successful")
+def then_validation_success(context):
+    assert context.validation_success is True
+
+
+# ---------------------------------------------------------------------------
+# Read-Only Enforcement
+# ---------------------------------------------------------------------------
+
+
+@given("a temporary SQLite database with a test table")
+def given_temp_sqlite_db(context):
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    conn = sqlite3.connect(path)
+    conn.execute("CREATE TABLE test_data (id INTEGER PRIMARY KEY, value TEXT)")
+    conn.commit()
+    conn.close()
+    context.db_path = path
+
+
+@given("a transaction sandbox in read-only mode")
+def given_ro_sandbox(context):
+    sandbox = TransactionSandbox(
+        resource_id="res-ro-1",
+        original_path=context.db_path,
+        read_only=True,
+    )
+    sandbox.create(plan_id="PLAN-RO")
+    context.tx_sandbox = sandbox
+
+
+@when("I attempt to insert data via the sandbox")
+def when_insert_readonly(context):
+    sandbox = context.tx_sandbox
+    try:
+        sandbox.execute("INSERT INTO test_data (id, value) VALUES (1, 'forbidden')")
+        context.write_rejected = False
+    except Exception:
+        context.write_rejected = True
+    finally:
+        sandbox.cleanup()
+
+
+@then("the sandbox should reject the write operation")
+def then_write_rejected(context):
+    assert context.write_rejected is True
+
+
+# ---------------------------------------------------------------------------
+# Transaction Sandbox Lifecycle
+# ---------------------------------------------------------------------------
+
+
+@given("a transaction sandbox for the database")
+def given_tx_sandbox(context):
+    sandbox = TransactionSandbox(
+        resource_id="res-tx-1",
+        original_path=context.db_path,
+    )
+    context.tx_sandbox = sandbox
+
+
+@when('I create the sandbox with plan "{plan_id}"')
+def when_create_sandbox(context, plan_id):
+    context.tx_sandbox.create(plan_id=plan_id)
+
+
+@when("I insert a row via the sandbox")
+def when_insert_row(context):
+    context.tx_sandbox.execute(
+        "INSERT INTO test_data (id, value) VALUES (99, 'test-value')"
+    )
+
+
+@when("I commit the sandbox")
+def when_commit_sandbox(context):
+    context.commit_result = context.tx_sandbox.commit("test commit")
+    context.tx_sandbox.cleanup()
+
+
+@when("I rollback the sandbox")
+def when_rollback_sandbox(context):
+    context.tx_sandbox.rollback()
+    context.tx_sandbox.cleanup()
+
+
+@when("I cleanup the sandbox")
+def when_cleanup_sandbox(context):
+    context.tx_sandbox.cleanup()
+
+
+@then("the row should be persisted in the database")
+def then_row_persisted(context):
+    conn = sqlite3.connect(context.db_path)
+    rows = conn.execute("SELECT value FROM test_data WHERE id = 99").fetchall()
+    conn.close()
+    assert len(rows) == 1, f"Expected 1 row, got {len(rows)}"
+    assert rows[0][0] == "test-value"
+
+
+@then("the row should not be persisted in the database")
+def then_row_not_persisted(context):
+    conn = sqlite3.connect(context.db_path)
+    rows = conn.execute("SELECT value FROM test_data WHERE id = 99").fetchall()
+    conn.close()
+    assert len(rows) == 0, f"Expected 0 rows, got {len(rows)}"
+
+
+@then('the sandbox status should be "{status}"')
+def then_sandbox_status(context, status):
+    assert context.tx_sandbox.status == SandboxStatus(status)
+
+
+# ---------------------------------------------------------------------------
+# SQLite Validation
+# ---------------------------------------------------------------------------
+
+
+@given("a temporary SQLite database")
+def given_temp_sqlite(context):
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    context.sqlite_path = path
+
+
+@when("I validate the sqlite connection")
+def when_validate_sqlite(context):
+    success, msg = validate_connection("sqlite", {"path": context.sqlite_path})
+    context.sqlite_valid_ok = success
+    context.sqlite_valid_msg = msg
+
+
+@when("I validate an in-memory sqlite connection")
+def when_validate_sqlite_memory(context):
+    success, msg = validate_connection("sqlite", {"path": ":memory:"})
+    context.sqlite_valid_ok = success
+    context.sqlite_valid_msg = msg
+
+
+@then("the sqlite validation should succeed")
+def then_sqlite_valid(context):
+    assert context.sqlite_valid_ok is True
+
+
+# ---------------------------------------------------------------------------
+# DatabaseResourceHandler Protocol
+# ---------------------------------------------------------------------------
+
+
+@given("a DatabaseResourceHandler instance")
+def given_db_handler(context):
+    context.db_handler = DatabaseResourceHandler()
+
+
+@then("the handler should satisfy the ResourceHandler protocol")
+def then_handler_protocol(context):
+    assert isinstance(context.db_handler, ResourceHandler)
+
+
+@then('the handler default strategy should be "{strategy}"')
+def then_handler_strategy(context, strategy):
+    assert context.db_handler._default_strategy.value == strategy
+
+
+# ---------------------------------------------------------------------------
+# SandboxFactory transaction_rollback routing
+# ---------------------------------------------------------------------------
+
+
+@when(
+    'a sandbox is requested for resource "{res_id}" '
+    'at "{path}" using transaction_rollback'
+)
+def when_factory_tx(context, res_id, path):
+    context.factory_sandbox = context.sandbox_factory.create_sandbox(
+        resource_id=res_id,
+        original_path=path,
+        sandbox_strategy="transaction_rollback",
+    )
+
+
+@then("the created sandbox should be a TransactionSandbox")
+def then_sandbox_is_tx(context):
+    assert isinstance(context.factory_sandbox, TransactionSandbox)
+
+
+@then('the factory should report "{strategy}" as supported')
+def then_factory_supported(context, strategy):
+    assert SandboxFactory.is_supported(strategy) is True
+
+
+# ---------------------------------------------------------------------------
+# TransactionSandbox edge cases
+# ---------------------------------------------------------------------------
+
+
+@when("I create a TransactionSandbox with empty resource_id")
+def when_tx_empty_res_id(context):
+    try:
+        TransactionSandbox(resource_id="", original_path="/tmp/x")
+        context.edge_error = None
+    except ValueError as exc:
+        context.edge_error = exc
+
+
+@when("I create a TransactionSandbox with empty original_path")
+def when_tx_empty_path(context):
+    try:
+        TransactionSandbox(resource_id="res-1", original_path="")
+        context.edge_error = None
+    except ValueError as exc:
+        context.edge_error = exc
+
+
+@then('the edge error should be a ValueError mentioning "{fragment}"')
+def then_value_error_with_msg(context, fragment):
+    assert context.edge_error is not None, "Expected a ValueError"
+    assert isinstance(context.edge_error, ValueError)
+    assert fragment in str(context.edge_error)
+
+
+@when("I try to create the sandbox with an empty plan_id")
+def when_empty_plan_id(context):
+    try:
+        context.tx_sandbox.create(plan_id="")
+        context.edge_error = None
+    except ValueError as exc:
+        context.edge_error = exc
+
+
+@then("get_path should return the database path")
+def then_get_path_returns_db(context):
+    sandbox = context.tx_sandbox
+    result = sandbox.get_path("table")
+    assert result == context.db_path
+
+
+@then("get_path with traversal should raise ValueError")
+def then_get_path_traversal(context):
+    sandbox = context.tx_sandbox
+    raised = False
+    try:
+        sandbox.get_path("../../etc/passwd")
+    except ValueError:
+        raised = True
+    assert raised, "Expected ValueError for path traversal"
+
+
+@then("get_path before create should raise SandboxStateError")
+def then_get_path_wrong_state(context):
+    from cleveragents.infrastructure.sandbox.protocol import SandboxStateError
+
+    sandbox = context.tx_sandbox
+    raised = False
+    try:
+        sandbox.get_path("test")
+    except SandboxStateError:
+        raised = True
+    assert raised, "Expected SandboxStateError"
+
+
+@then("commit before create should raise SandboxStateError")
+def then_commit_wrong_state(context):
+    from cleveragents.infrastructure.sandbox.protocol import SandboxStateError
+
+    sandbox = context.tx_sandbox
+    raised = False
+    try:
+        sandbox.commit("test")
+    except SandboxStateError:
+        raised = True
+    assert raised, "Expected SandboxStateError"
+
+
+@then("rollback before create should raise SandboxStateError")
+def then_rollback_wrong_state(context):
+    from cleveragents.infrastructure.sandbox.protocol import SandboxStateError
+
+    sandbox = context.tx_sandbox
+    raised = False
+    try:
+        sandbox.rollback()
+    except SandboxStateError:
+        raised = True
+    assert raised, "Expected SandboxStateError"
+
+
+@when("I cleanup the sandbox again")
+def when_cleanup_again(context):
+    context.tx_sandbox.cleanup()
+
+
+@then("execute before create should raise SandboxStateError")
+def then_execute_wrong_state(context):
+    from cleveragents.infrastructure.sandbox.protocol import SandboxStateError
+
+    sandbox = context.tx_sandbox
+    raised = False
+    try:
+        sandbox.execute("SELECT 1")
+    except SandboxStateError:
+        raised = True
+    assert raised, "Expected SandboxStateError"
+
+
+@when('I validate a duckdb connection with path "{path}"')
+def when_validate_duckdb(context, path):
+    success, msg = validate_connection("duckdb", {"path": path})
+    context.duckdb_valid_ok = success
+    context.duckdb_valid_msg = msg
+
+
+@then("the duckdb validation should succeed with a stub message")
+def then_duckdb_stub(context):
+    assert context.duckdb_valid_ok is True
+    # Either driver works or stub message about not installed
+    assert (
+        "DuckDB" in context.duckdb_valid_msg
+        or "duckdb" in context.duckdb_valid_msg.lower()
+    )
+
+
+@given('mysql connection properties with host "{host}" and port {port:d}')
+def given_mysql_props(context, host, port):
+    context.conn_props = {"host": host, "port": port}
+
+
+@when("I validate an unknown database type")
+def when_validate_unknown_db(context):
+    success, msg = validate_connection("unknown_db", {})
+    context.unknown_valid_ok = success
+    context.unknown_valid_msg = msg
+
+
+@then("the unknown validation should fail")
+def then_unknown_fail(context):
+    assert context.unknown_valid_ok is False
+    assert "Unknown" in context.unknown_valid_msg
+
+
+@given('mysql properties with connection string "{conn_str}"')
+def given_mysql_conn_str(context, conn_str):
+    context.conn_props = {"connection-string": conn_str}

features/steps/sandbox_factory_coverage_steps.py

@@ -5,7 +5,8 @@ Covers lines in ``factory.py``:
 - none strategy -> NoSandbox
 - git_worktree strategy -> GitWorktreeSandbox
 - copy_on_write strategy -> CopyOnWriteSandbox
-- transaction_rollback / snapshot -> NotImplementedError
+- transaction_rollback -> TransactionSandbox
+- snapshot -> NotImplementedError
 - overlay / versioning (removed) -> ValueError (unknown)
 - Unknown strategy -> ValueError
 - is_supported() for all strategies
@@ -21,6 +22,7 @@ from cleveragents.infrastructure.sandbox.factory import SandboxFactory
 from cleveragents.infrastructure.sandbox.git_worktree import GitWorktreeSandbox
 from cleveragents.infrastructure.sandbox.no_sandbox import NoSandbox
 from cleveragents.infrastructure.sandbox.protocol import SandboxStatus
+from cleveragents.infrastructure.sandbox.transaction_sandbox import TransactionSandbox
 
 # ---------------------------------------------------------------------------
 # Givens
@@ -131,14 +133,11 @@ def when_factory_overlay(context, res_id: str, path: str):
 def when_factory_transaction_rollback(context, res_id: str, path: str):
     """Request a sandbox with the 'transaction_rollback' strategy."""
     context.factory_error = None
-    try:
-        context.sandbox_factory.create_sandbox(
-            resource_id=res_id,
-            original_path=path,
-            sandbox_strategy="transaction_rollback",
-        )
-    except NotImplementedError as exc:
-        context.factory_error = exc
+    context.factory_sandbox = context.sandbox_factory.create_sandbox(
+        resource_id=res_id,
+        original_path=path,
+        sandbox_strategy="transaction_rollback",
+    )
 
 
 @when(
@@ -338,6 +337,13 @@ def then_factory_not_implemented(context):
     assert isinstance(context.factory_error, NotImplementedError)
 
 
+@then("the factory should produce a transaction sandbox in the pending state")
+def then_factory_produces_transaction_sandbox(context):
+    """Assert the returned sandbox is a TransactionSandbox in PENDING."""
+    assert isinstance(context.factory_sandbox, TransactionSandbox)
+    assert context.factory_sandbox.status == SandboxStatus.PENDING
+
+
 # ---------------------------------------------------------------------------
 # Thens - is_supported
 # ---------------------------------------------------------------------------

robot/database_resources.robot

@@ -0,0 +1,73 @@
+*** Settings ***
+Documentation    Integration smoke tests for database resource types and transaction sandbox
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_database_resources.py
+
+*** Test Cases ***
+Database Handler Protocol Conformance
+    [Documentation]    Verify DatabaseResourceHandler satisfies ResourceHandler
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    protocol-check    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    protocol-check-ok
+
+Database Type Definitions Present
+    [Documentation]    Verify all 4 database type definitions are present
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    type-defs    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    type-defs-ok
+
+SQLite Connection Validation
+    [Documentation]    Verify SQLite connection validation succeeds
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    sqlite-validate    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    sqlite-validate-ok
+
+Transaction Sandbox Commit
+    [Documentation]    Verify TransactionSandbox commit persists data
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    tx-commit    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tx-commit-ok
+
+Transaction Sandbox Rollback
+    [Documentation]    Verify TransactionSandbox rollback discards data
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    tx-rollback    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    tx-rollback-ok
+
+Credential Masking In Validation
+    [Documentation]    Verify credentials are masked in validation output
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    cred-mask    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    cred-mask-ok
+
+SandboxFactory Transaction Rollback Routing
+    [Documentation]    Verify SandboxFactory creates TransactionSandbox for transaction_rollback
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    factory-routing    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    factory-routing-ok
+
+Read Only Enforcement
+    [Documentation]    Verify read-only mode prevents writes
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    read-only    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    read-only-ok

robot/helper_database_resources.py

@@ -0,0 +1,190 @@
+"""Helper utilities for database resource Robot smoke tests.
+
+Each command prints a single ``<command>-ok`` token on success so the
+calling Robot test can assert on ``stdout``.  Only uses sqlite3 (stdlib)
+for concrete database testing.
+"""
+
+from __future__ import annotations
+
+import os
+import sqlite3
+import sys
+import tempfile
+
+from cleveragents.infrastructure.sandbox.factory import SandboxFactory
+from cleveragents.infrastructure.sandbox.transaction_sandbox import (
+    TransactionSandbox,
+)
+from cleveragents.resource.handlers.database import (
+    DATABASE_TYPE_DEFS,
+    DATABASE_TYPE_NAMES,
+    DatabaseResourceHandler,
+    resolve_connection_args,
+    validate_connection,
+)
+from cleveragents.resource.handlers.protocol import ResourceHandler
+from cleveragents.shared.redaction import mask_database_url
+
+
+def _protocol_check() -> None:
+    """Verify DatabaseResourceHandler satisfies ResourceHandler."""
+    handler = DatabaseResourceHandler()
+    assert isinstance(handler, ResourceHandler), (
+        "DatabaseResourceHandler not a ResourceHandler"
+    )
+    print("protocol-check-ok")
+
+
+def _type_defs() -> None:
+    """Verify all 4 database type definitions are present."""
+    names = {td["name"] for td in DATABASE_TYPE_DEFS}
+    for expected in ("postgres", "mysql", "sqlite", "duckdb"):
+        assert expected in names, f"Missing type def: {expected}"
+    assert names == DATABASE_TYPE_NAMES
+    print("type-defs-ok")
+
+
+def _sqlite_validate() -> None:
+    """Validate SQLite connection with a temp file."""
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    try:
+        success, msg = validate_connection("sqlite", {"path": path})
+        assert success, f"SQLite validation failed: {msg}"
+        print("sqlite-validate-ok")
+    finally:
+        os.unlink(path)
+
+
+def _tx_commit() -> None:
+    """Test transaction sandbox commit persists data."""
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    try:
+        # Setup table
+        conn = sqlite3.connect(path)
+        conn.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT)")
+        conn.commit()
+        conn.close()
+
+        # Use sandbox
+        sandbox = TransactionSandbox(resource_id="robot-tx-1", original_path=path)
+        sandbox.create(plan_id="PLAN-ROBOT-COMMIT")
+        sandbox.execute("INSERT INTO items (id, name) VALUES (1, 'committed')")
+        result = sandbox.commit("robot commit")
+        assert result.success
+        sandbox.cleanup()
+
+        # Verify data persisted
+        conn = sqlite3.connect(path)
+        rows = conn.execute("SELECT name FROM items WHERE id = 1").fetchall()
+        conn.close()
+        assert len(rows) == 1
+        assert rows[0][0] == "committed"
+        print("tx-commit-ok")
+    finally:
+        os.unlink(path)
+
+
+def _tx_rollback() -> None:
+    """Test transaction sandbox rollback discards data."""
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    try:
+        # Setup table
+        conn = sqlite3.connect(path)
+        conn.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT)")
+        conn.commit()
+        conn.close()
+
+        # Use sandbox
+        sandbox = TransactionSandbox(resource_id="robot-tx-2", original_path=path)
+        sandbox.create(plan_id="PLAN-ROBOT-ROLLBACK")
+        sandbox.execute("INSERT INTO items (id, name) VALUES (2, 'rolled-back')")
+        # Transition to ACTIVE by executing, then rollback
+        sandbox.rollback()
+        sandbox.cleanup()
+
+        # Verify data NOT persisted
+        conn = sqlite3.connect(path)
+        rows = conn.execute("SELECT name FROM items WHERE id = 2").fetchall()
+        conn.close()
+        assert len(rows) == 0
+        print("tx-rollback-ok")
+    finally:
+        os.unlink(path)
+
+
+def _cred_mask() -> None:
+    """Verify credential masking in validation messages."""
+    props = {"connection-string": "postgresql://admin:supersecret@db:5432/app"}
+    resolved = resolve_connection_args(props, "postgres")
+    _success, msg = validate_connection("postgres", resolved)
+    assert "supersecret" not in msg, f"Secret leaked: {msg}"
+    # Also test mask_database_url directly
+    masked = mask_database_url("postgresql://admin:supersecret@db:5432/app")
+    assert "supersecret" not in masked
+    print("cred-mask-ok")
+
+
+def _factory_routing() -> None:
+    """Verify SandboxFactory routes transaction_rollback correctly."""
+    factory = SandboxFactory()
+    sandbox = factory.create_sandbox(
+        resource_id="factory-test",
+        original_path=":memory:",
+        sandbox_strategy="transaction_rollback",
+    )
+    assert isinstance(sandbox, TransactionSandbox)
+    assert SandboxFactory.is_supported("transaction_rollback")
+    print("factory-routing-ok")
+
+
+def _read_only() -> None:
+    """Verify read-only sandbox prevents writes."""
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    try:
+        conn = sqlite3.connect(path)
+        conn.execute("CREATE TABLE items (id INTEGER PRIMARY KEY, name TEXT)")
+        conn.commit()
+        conn.close()
+
+        sandbox = TransactionSandbox(
+            resource_id="ro-test", original_path=path, read_only=True
+        )
+        sandbox.create(plan_id="PLAN-RO-ROBOT")
+        rejected = False
+        try:
+            sandbox.execute("INSERT INTO items (id, name) VALUES (1, 'forbidden')")
+        except Exception:
+            rejected = True
+        finally:
+            sandbox.cleanup()
+        assert rejected, "Write was not rejected in read-only mode"
+        print("read-only-ok")
+    finally:
+        os.unlink(path)
+
+
+_COMMANDS = {
+    "protocol-check": _protocol_check,
+    "type-defs": _type_defs,
+    "sqlite-validate": _sqlite_validate,
+    "tx-commit": _tx_commit,
+    "tx-rollback": _tx_rollback,
+    "cred-mask": _cred_mask,
+    "factory-routing": _factory_routing,
+    "read-only": _read_only,
+}
+
+
+if __name__ == "__main__":
+    if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS:
+        print(
+            f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>",
+            file=sys.stderr,
+        )
+        sys.exit(1)
+    _COMMANDS[sys.argv[1]]()

robot/helper_sandbox_integration.py

@@ -155,17 +155,17 @@ def _factory_create_none_strategy() -> None:
     )
     assert isinstance(cow_sandbox, CopyOnWriteSandbox)
 
-    # Verify unimplemented strategies still raise
-    try:
-        factory.create_sandbox(
-            resource_id="res-006",
-            original_path="/tmp/test",
-            sandbox_strategy="transaction_rollback",
-        )
-        print("FAIL: Expected NotImplementedError for transaction_rollback")
-        return
-    except NotImplementedError:
-        pass
+    # Verify transaction_rollback now produces a TransactionSandbox
+    from cleveragents.infrastructure.sandbox.transaction_sandbox import (
+        TransactionSandbox,
+    )
+
+    tx_sandbox = factory.create_sandbox(
+        resource_id="res-006",
+        original_path=":memory:",
+        sandbox_strategy="transaction_rollback",
+    )
+    assert isinstance(tx_sandbox, TransactionSandbox)
 
     print("factory-create-none-ok")
 
@@ -177,8 +177,10 @@ def _factory_supported_strategies() -> None:
     assert SandboxFactory.is_supported("git_worktree")
     assert SandboxFactory.is_supported("copy_on_write")
 
+    # transaction_rollback is now supported
+    assert SandboxFactory.is_supported("transaction_rollback")
+
     # Unimplemented strategies are not supported
-    assert not SandboxFactory.is_supported("transaction_rollback")
     assert not SandboxFactory.is_supported("snapshot")
 
     # Resource type strategy mapping (spec-aligned)

src/cleveragents/application/services/_resource_registry_data.py

@@ -28,6 +28,7 @@ from cleveragents.infrastructure.database.models import (
     ResourceModel,
     ResourceTypeModel,
 )
+from cleveragents.resource.handlers.database import DATABASE_TYPE_DEFS
 
 __all__ = [
     "BUILTIN_TYPES",
@@ -194,6 +195,7 @@ BUILTIN_TYPES: list[dict[str, Any]] = [
             "checkpoint": False,
         },
     },
+    *DATABASE_TYPE_DEFS,
 ]
 
 

src/cleveragents/domain/models/core/resource_type.py

@@ -162,6 +162,10 @@ class ResourceTypeSpec(BaseModel):
             "container-instance",
             "devcontainer-instance",
             "devcontainer-file",
+            "postgres",
+            "mysql",
+            "sqlite",
+            "duckdb",
         }
     )
 

src/cleveragents/infrastructure/sandbox/__init__.py

@@ -40,6 +40,7 @@ from cleveragents.infrastructure.sandbox.protocol import (
     SandboxError,
     SandboxStatus,
 )
+from cleveragents.infrastructure.sandbox.transaction_sandbox import TransactionSandbox
 
 __all__ = [
     "BoundaryCache",
@@ -63,6 +64,7 @@ __all__ = [
     "SandboxManager",
     "SandboxStatus",
     "SequentialMergeStrategy",
+    "TransactionSandbox",
     "compute_sandbox_domains",
     "is_sandbox_boundary",
     "sandbox_boundary",

src/cleveragents/infrastructure/sandbox/factory.py

@@ -19,6 +19,7 @@ from cleveragents.infrastructure.sandbox.no_sandbox import NoSandbox
 from cleveragents.infrastructure.sandbox.protocol import (
     Sandbox,
 )
+from cleveragents.infrastructure.sandbox.transaction_sandbox import TransactionSandbox
 
 logger = logging.getLogger(__name__)
 
@@ -39,7 +40,7 @@ SandboxStrategyStr = Literal[
 
 # Strategies that have concrete implementations
 _IMPLEMENTED_STRATEGIES: frozenset[str] = frozenset(
-    {"none", "git_worktree", "copy_on_write"}
+    {"none", "git_worktree", "copy_on_write", "transaction_rollback"}
 )
 
 # Resource type to supported strategies mapping (spec-aligned)
@@ -50,6 +51,10 @@ _SUPPORTED_STRATEGIES: dict[str, list[SandboxStrategyStr]] = {
     "fs-directory": ["copy_on_write", "none"],
     "fs-file": ["copy_on_write", "none"],
     "api_endpoint": ["none"],
+    "postgres": ["transaction_rollback", "none"],
+    "mysql": ["transaction_rollback", "none"],
+    "sqlite": ["transaction_rollback", "none"],
+    "duckdb": ["transaction_rollback", "none"],
 }
 
 
@@ -62,8 +67,9 @@ class SandboxFactory:
     - ``"none"`` -> :class:`NoSandbox`
     - ``"git_worktree"`` -> :class:`GitWorktreeSandbox`
     - ``"copy_on_write"`` -> :class:`CopyOnWriteSandbox`
+    - ``"transaction_rollback"`` -> :class:`TransactionSandbox`
 
-    ``"transaction_rollback"`` and ``"snapshot"`` raise ``NotImplementedError``.
+    ``"snapshot"`` raises ``NotImplementedError``.
     """
 
     def create_sandbox(
@@ -112,8 +118,9 @@ class SandboxFactory:
             )
 
         if sandbox_strategy == STRATEGY_TRANSACTION_ROLLBACK:
-            raise NotImplementedError(
-                "Database transaction sandbox not yet implemented"
+            return TransactionSandbox(
+                resource_id=resource_id,
+                original_path=original_path,
             )
 
         if sandbox_strategy == STRATEGY_SNAPSHOT:

src/cleveragents/infrastructure/sandbox/transaction_sandbox.py

@@ -0,0 +1,370 @@
+"""Transaction-rollback sandbox for database resources.
+
+Provides isolation through database transaction semantics
+(``BEGIN`` / ``ROLLBACK`` / ``COMMIT``).  The concrete backend
+uses Python's stdlib ``sqlite3`` module; PostgreSQL and MySQL
+backends are stubs that raise ``NotImplementedError``.
+
+Read-only mode is supported via SQLite's ``PRAGMA query_only``.
+
+Implements the
+:class:`~cleveragents.infrastructure.sandbox.protocol.Sandbox`
+protocol.
+
+Lifecycle::
+
+    sandbox = TransactionSandbox(resource_id, original_path)
+    ctx = sandbox.create(plan_id)       # opens connection, BEGIN
+    sandbox.execute("INSERT ...")        # within transaction
+    result = sandbox.commit("msg")      # COMMIT
+    sandbox.cleanup()                   # close connection
+
+Based on:
+    - implementation_plan.md group M7.post-resource-db
+    - Sandbox protocol (B3.1)
+"""
+
+from __future__ import annotations
+
+import contextlib
+import logging
+import sqlite3
+from datetime import datetime
+
+from ulid import ULID
+
+from cleveragents.infrastructure.sandbox.protocol import (
+    CommitResult,
+    SandboxCommitError,
+    SandboxContext,
+    SandboxCreationError,
+    SandboxRollbackError,
+    SandboxStateError,
+    SandboxStatus,
+)
+
+logger = logging.getLogger(__name__)
+
+
+class TransactionSandbox:
+    """Sandbox that isolates changes using database transactions.
+
+    Opens a database connection and wraps all operations in a
+    transaction.  ``commit()`` issues ``COMMIT``; ``rollback()``
+    issues ``ROLLBACK``.
+
+    Currently only SQLite is fully implemented via the stdlib
+    ``sqlite3`` module.  The ``original_path`` parameter is
+    interpreted as the SQLite database file path (or ``:memory:``).
+
+    Implements the
+    :class:`~cleveragents.infrastructure.sandbox.protocol.Sandbox`
+    protocol.
+    """
+
+    def __init__(
+        self,
+        resource_id: str,
+        original_path: str,
+        *,
+        read_only: bool = False,
+    ) -> None:
+        """Initialise a transaction sandbox.
+
+        Args:
+            resource_id: Identifier of the database resource.
+            original_path: Database file path or connection string.
+            read_only: When ``True``, enforce read-only mode on the
+                connection (SQLite ``PRAGMA query_only = ON``).
+
+        Raises:
+            ValueError: If *resource_id* is empty.
+            ValueError: If *original_path* is empty.
+        """
+        if not resource_id:
+            raise ValueError("resource_id cannot be empty")
+        if not original_path:
+            raise ValueError("original_path cannot be empty")
+
+        self._sandbox_id: str = str(ULID())
+        self._resource_id: str = resource_id
+        self._original_path: str = original_path
+        self._read_only: bool = read_only
+        self._status: SandboxStatus = SandboxStatus.PENDING
+        self._context: SandboxContext | None = None
+        self._connection: sqlite3.Connection | None = None
+
+    # -- protocol properties -------------------------------------------------
+
+    @property
+    def sandbox_id(self) -> str:
+        """Unique identifier for this sandbox instance."""
+        return self._sandbox_id
+
+    @property
+    def status(self) -> SandboxStatus:
+        """Current lifecycle status."""
+        return self._status
+
+    @property
+    def context(self) -> SandboxContext | None:
+        """Context after creation, ``None`` before ``create``."""
+        return self._context
+
+    # -- protocol methods ----------------------------------------------------
+
+    def create(self, plan_id: str) -> SandboxContext:
+        """Open a database connection and begin a transaction.
+
+        Args:
+            plan_id: The plan that owns this sandbox.
+
+        Returns:
+            A :class:`SandboxContext` with metadata about the
+            transaction sandbox.
+
+        Raises:
+            ValueError: If *plan_id* is empty.
+            SandboxStateError: If not in ``PENDING`` status.
+            SandboxCreationError: If the database connection fails.
+        """
+        if not plan_id:
+            raise ValueError("plan_id cannot be empty")
+
+        SandboxStatus.assert_transition(self._status, SandboxStatus.CREATED)
+
+        try:
+            self._connection = sqlite3.connect(
+                self._original_path,
+                isolation_level=None,  # manual transaction control
+            )
+            # Begin explicit transaction
+            self._connection.execute("BEGIN")
+
+            if self._read_only:
+                self._connection.execute("PRAGMA query_only = ON")
+
+        except sqlite3.Error as exc:
+            self._status = SandboxStatus.ERRORED
+            raise SandboxCreationError(
+                f"Failed to open database for resource {self._resource_id}: {exc}"
+            ) from exc
+
+        self._context = SandboxContext(
+            sandbox_id=self._sandbox_id,
+            sandbox_path=self._original_path,
+            original_path=self._original_path,
+            resource_id=self._resource_id,
+            plan_id=plan_id,
+            created_at=datetime.now(),
+            metadata={
+                "strategy": "transaction_rollback",
+                "read_only": self._read_only,
+                "db_path": self._original_path,
+            },
+        )
+        self._status = SandboxStatus.CREATED
+
+        logger.info(
+            "Created transaction sandbox: plan=%s resource=%s path=%s read_only=%s",
+            plan_id,
+            self._resource_id,
+            self._original_path,
+            self._read_only,
+        )
+
+        return self._context
+
+    def get_path(self, resource_path: str) -> str:
+        """Return the database path (resource_path is ignored for DB).
+
+        For database sandboxes the ``resource_path`` argument is not
+        meaningful in the filesystem sense.  This method returns the
+        database file path for compatibility with the protocol.
+
+        Args:
+            resource_path: Ignored (kept for protocol compatibility).
+
+        Returns:
+            The database file path.
+
+        Raises:
+            SandboxStateError: If sandbox is not in a usable status.
+            ValueError: If *resource_path* attempts directory traversal.
+        """
+        if self._status not in (
+            SandboxStatus.CREATED,
+            SandboxStatus.ACTIVE,
+        ):
+            raise SandboxStateError(
+                f"Cannot resolve path in status {self._status.value}"
+            )
+
+        if ".." in resource_path.split("/"):
+            raise ValueError(f"Path traversal not allowed: {resource_path}")
+
+        if self._status == SandboxStatus.CREATED:
+            self._status = SandboxStatus.ACTIVE
+
+        return self._original_path
+
+    def commit(self, message: str | None = None) -> CommitResult:
+        """Commit the database transaction.
+
+        Issues ``COMMIT`` on the underlying connection.
+
+        Args:
+            message: Optional log message.
+
+        Returns:
+            A :class:`CommitResult` describing the outcome.
+
+        Raises:
+            SandboxCommitError: If the COMMIT fails.
+            SandboxStateError: If sandbox is not in a committable status.
+        """
+        if self._status not in (
+            SandboxStatus.CREATED,
+            SandboxStatus.ACTIVE,
+        ):
+            raise SandboxStateError(f"Cannot commit from status {self._status.value}")
+
+        SandboxStatus.assert_transition(self._status, SandboxStatus.COMMITTED)
+
+        try:
+            if self._connection is not None:
+                self._connection.execute("COMMIT")
+        except sqlite3.Error as exc:
+            self._status = SandboxStatus.ERRORED
+            raise SandboxCommitError(
+                f"Failed to commit transaction for sandbox {self._sandbox_id}: {exc}"
+            ) from exc
+
+        self._status = SandboxStatus.COMMITTED
+
+        logger.info(
+            "Committed transaction sandbox: sandbox_id=%s message=%s",
+            self._sandbox_id,
+            message,
+        )
+
+        return CommitResult(
+            sandbox_id=self._sandbox_id,
+            success=True,
+            commit_ref=None,
+            changed_files=[],
+            added_files=[],
+            deleted_files=[],
+            error=None,
+            timestamp=datetime.now(),
+        )
+
+    def rollback(self) -> None:
+        """Roll back the database transaction.
+
+        Issues ``ROLLBACK`` on the underlying connection and begins a
+        new transaction so the sandbox can be re-used.
+
+        Raises:
+            SandboxRollbackError: If the ROLLBACK fails.
+            SandboxStateError: If called in an invalid status.
+        """
+        if self._status != SandboxStatus.ACTIVE:
+            raise SandboxStateError(f"Cannot rollback from status {self._status.value}")
+
+        SandboxStatus.assert_transition(self._status, SandboxStatus.ROLLED_BACK)
+
+        try:
+            if self._connection is not None:
+                self._connection.execute("ROLLBACK")
+                # Begin a new transaction so the sandbox can be reused
+                self._connection.execute("BEGIN")
+        except sqlite3.Error as exc:
+            self._status = SandboxStatus.ERRORED
+            raise SandboxRollbackError(
+                f"Failed to rollback transaction for sandbox {self._sandbox_id}: {exc}"
+            ) from exc
+
+        self._status = SandboxStatus.ROLLED_BACK
+
+        logger.info(
+            "Rolled back transaction sandbox: sandbox_id=%s",
+            self._sandbox_id,
+        )
+
+    def cleanup(self) -> None:
+        """Close the database connection.
+
+        Attempts a ROLLBACK before closing to discard any uncommitted
+        changes.  Idempotent -- safe to call multiple times.
+
+        Raises:
+            SandboxError: On unexpected errors during cleanup.
+        """
+        if self._status == SandboxStatus.CLEANED_UP:
+            return
+
+        logger.debug(
+            "Cleaning up transaction sandbox: sandbox_id=%s path=%s",
+            self._sandbox_id,
+            self._original_path,
+        )
+
+        if self._connection is not None:
+            with contextlib.suppress(sqlite3.Error):
+                # Discard any uncommitted work
+                if self._status in (
+                    SandboxStatus.CREATED,
+                    SandboxStatus.ACTIVE,
+                    SandboxStatus.ROLLED_BACK,
+                ):
+                    self._connection.execute("ROLLBACK")
+            with contextlib.suppress(sqlite3.Error):
+                self._connection.close()
+            self._connection = None
+
+        self._status = SandboxStatus.CLEANED_UP
+
+        logger.info(
+            "Cleaned up transaction sandbox: sandbox_id=%s",
+            self._sandbox_id,
+        )
+
+    # -- database helper methods ---------------------------------------------
+
+    def execute(
+        self, sql: str, params: tuple[object, ...] = ()
+    ) -> list[tuple[object, ...]]:
+        """Execute a SQL statement within the sandbox transaction.
+
+        This is a convenience method for database sandbox usage.
+        It is not part of the Sandbox protocol.
+
+        Args:
+            sql: SQL statement to execute.
+            params: Bind parameters.
+
+        Returns:
+            List of result rows (empty for non-SELECT statements).
+
+        Raises:
+            SandboxStateError: If sandbox is not in a usable status.
+            sqlite3.Error: If the SQL execution fails.
+        """
+        if self._status not in (
+            SandboxStatus.CREATED,
+            SandboxStatus.ACTIVE,
+        ):
+            raise SandboxStateError(
+                f"Cannot execute SQL in status {self._status.value}"
+            )
+
+        if self._status == SandboxStatus.CREATED:
+            self._status = SandboxStatus.ACTIVE
+
+        if self._connection is None:
+            raise SandboxStateError("Database connection not available")
+
+        cursor = self._connection.execute(sql, params)
+        rows: list[tuple[object, ...]] = cursor.fetchall()
+        return rows

src/cleveragents/resource/handlers/__init__.py

@@ -27,6 +27,7 @@ Handler strings stored on :class:`ResourceTypeSpec` use the format
 dynamically imports the module and returns an instance.
 """
 
+from cleveragents.resource.handlers.database import DatabaseResourceHandler
 from cleveragents.resource.handlers.devcontainer import DevcontainerHandler
 from cleveragents.resource.handlers.fs_directory import FsDirectoryHandler
 from cleveragents.resource.handlers.git_checkout import GitCheckoutHandler
@@ -37,6 +38,7 @@ from cleveragents.resource.handlers.resolver import (
 )
 
 __all__ = [
+    "DatabaseResourceHandler",
     "DevcontainerHandler",
     "FsDirectoryHandler",
     "GitCheckoutHandler",

src/cleveragents/resource/handlers/database.py

@@ -0,0 +1,415 @@
+"""Database resource handler for CleverAgents.
+
+Resolves database resources (``postgres``, ``mysql``, ``sqlite``,
+``duckdb``) into sandbox-backed :class:`BoundResource` instances
+using the ``transaction_rollback`` sandbox strategy.
+
+Each database type defines its own connection arguments (host, port,
+database, user, password for networked; path for file-based).
+Authentication can be supplied directly in config or via environment
+variables.
+
+Connection validation tests connectivity and returns safe error
+messages that mask credentials using the shared
+:mod:`cleveragents.shared.redaction` module.
+
+Based on:
+    - implementation_plan.md group M7.post-resource-db
+    - Built-in type definitions for database resources
+"""
+
+from __future__ import annotations
+
+import logging
+import os
+import sqlite3
+from typing import Any
+
+from cleveragents.domain.models.core.resource import SandboxStrategy
+from cleveragents.resource.handlers._base import BaseResourceHandler
+from cleveragents.shared.redaction import mask_database_url, redact_dict
+
+logger = logging.getLogger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Built-in database resource type definitions
+# ---------------------------------------------------------------------------
+
+POSTGRES_TYPE_DEF: dict[str, Any] = {
+    "name": "postgres",
+    "description": "PostgreSQL database connection.",
+    "resource_kind": "physical",
+    "sandbox_strategy": "transaction_rollback",
+    "user_addable": True,
+    "built_in": True,
+    "cli_args": [
+        {
+            "name": "connection-string",
+            "type": "string",
+            "required": False,
+            "description": (
+                "Full connection string "
+                "(e.g. postgresql://user:pass@host:5432/db). "
+                "Overrides individual args."
+            ),
+            "default": None,
+        },
+        {
+            "name": "host",
+            "type": "string",
+            "required": False,
+            "description": "Database server hostname.",
+            "default": "localhost",
+        },
+        {
+            "name": "port",
+            "type": "integer",
+            "required": False,
+            "description": "Database server port.",
+            "default": 5432,
+        },
+        {
+            "name": "dbname",
+            "type": "string",
+            "required": False,
+            "description": "Database name.",
+            "default": None,
+        },
+        {
+            "name": "user",
+            "type": "string",
+            "required": False,
+            "description": (
+                "Database user. Can also be set via CLEVERAGENTS_DB_USER env var."
+            ),
+            "default": None,
+        },
+        {
+            "name": "password",
+            "type": "string",
+            "required": False,
+            "description": (
+                "Database password. Can also be set via "
+                "CLEVERAGENTS_DB_PASSWORD env var."
+            ),
+            "default": None,
+        },
+    ],
+    "parent_types": [],
+    "child_types": [],
+    "handler": "cleveragents.resource.handlers.database:DatabaseResourceHandler",
+    "capabilities": {
+        "read": True,
+        "write": True,
+        "sandbox": True,
+        "checkpoint": False,
+    },
+}
+
+MYSQL_TYPE_DEF: dict[str, Any] = {
+    "name": "mysql",
+    "description": "MySQL database connection.",
+    "resource_kind": "physical",
+    "sandbox_strategy": "transaction_rollback",
+    "user_addable": True,
+    "built_in": True,
+    "cli_args": [
+        {
+            "name": "connection-string",
+            "type": "string",
+            "required": False,
+            "description": (
+                "Full connection string "
+                "(e.g. mysql://user:pass@host:3306/db). "
+                "Overrides individual args."
+            ),
+            "default": None,
+        },
+        {
+            "name": "host",
+            "type": "string",
+            "required": False,
+            "description": "Database server hostname.",
+            "default": "localhost",
+        },
+        {
+            "name": "port",
+            "type": "integer",
+            "required": False,
+            "description": "Database server port.",
+            "default": 3306,
+        },
+        {
+            "name": "dbname",
+            "type": "string",
+            "required": False,
+            "description": "Database name.",
+            "default": None,
+        },
+        {
+            "name": "user",
+            "type": "string",
+            "required": False,
+            "description": (
+                "Database user. Can also be set via CLEVERAGENTS_DB_USER env var."
+            ),
+            "default": None,
+        },
+        {
+            "name": "password",
+            "type": "string",
+            "required": False,
+            "description": (
+                "Database password. Can also be set via "
+                "CLEVERAGENTS_DB_PASSWORD env var."
+            ),
+            "default": None,
+        },
+    ],
+    "parent_types": [],
+    "child_types": [],
+    "handler": "cleveragents.resource.handlers.database:DatabaseResourceHandler",
+    "capabilities": {
+        "read": True,
+        "write": True,
+        "sandbox": True,
+        "checkpoint": False,
+    },
+}
+
+SQLITE_TYPE_DEF: dict[str, Any] = {
+    "name": "sqlite",
+    "description": "SQLite database file.",
+    "resource_kind": "physical",
+    "sandbox_strategy": "transaction_rollback",
+    "user_addable": True,
+    "built_in": True,
+    "cli_args": [
+        {
+            "name": "path",
+            "type": "path",
+            "required": True,
+            "description": (
+                "Path to the SQLite database file (use :memory: for in-memory)."
+            ),
+        },
+    ],
+    "parent_types": ["fs-directory", "git-checkout"],
+    "child_types": [],
+    "handler": "cleveragents.resource.handlers.database:DatabaseResourceHandler",
+    "capabilities": {
+        "read": True,
+        "write": True,
+        "sandbox": True,
+        "checkpoint": False,
+    },
+}
+
+DUCKDB_TYPE_DEF: dict[str, Any] = {
+    "name": "duckdb",
+    "description": "DuckDB database (file-based or in-memory).",
+    "resource_kind": "physical",
+    "sandbox_strategy": "transaction_rollback",
+    "user_addable": True,
+    "built_in": True,
+    "cli_args": [
+        {
+            "name": "path",
+            "type": "path",
+            "required": False,
+            "description": (
+                "Path to the DuckDB database file. Omit or use :memory: for in-memory."
+            ),
+            "default": ":memory:",
+        },
+    ],
+    "parent_types": ["fs-directory", "git-checkout"],
+    "child_types": [],
+    "handler": "cleveragents.resource.handlers.database:DatabaseResourceHandler",
+    "capabilities": {
+        "read": True,
+        "write": True,
+        "sandbox": True,
+        "checkpoint": False,
+    },
+}
+
+#: All database type definitions in registration order.
+DATABASE_TYPE_DEFS: list[dict[str, Any]] = [
+    POSTGRES_TYPE_DEF,
+    MYSQL_TYPE_DEF,
+    SQLITE_TYPE_DEF,
+    DUCKDB_TYPE_DEF,
+]
+
+#: Database type names.
+DATABASE_TYPE_NAMES: frozenset[str] = frozenset(
+    {"postgres", "mysql", "sqlite", "duckdb"}
+)
+
+
+# ---------------------------------------------------------------------------
+# Connection argument resolution
+# ---------------------------------------------------------------------------
+
+
+def resolve_connection_args(
+    properties: dict[str, str | int | float | bool | None],
+    resource_type: str,
+) -> dict[str, str | int | None]:
+    """Resolve connection arguments from properties and env vars.
+
+    For networked databases (postgres, mysql), credentials can come
+    from environment variables ``CLEVERAGENTS_DB_USER`` and
+    ``CLEVERAGENTS_DB_PASSWORD`` when not set in properties.
+
+    Args:
+        properties: Resource properties dict.
+        resource_type: The database resource type name.
+
+    Returns:
+        Resolved connection arguments.
+    """
+    args: dict[str, str | int | None] = {}
+
+    if resource_type in ("postgres", "mysql"):
+        conn_str = properties.get("connection-string")
+        if conn_str is not None:
+            args["connection-string"] = str(conn_str)
+            return args
+
+        args["host"] = str(properties.get("host", "localhost") or "localhost")
+        default_port = 5432 if resource_type == "postgres" else 3306
+        raw_port = properties.get("port", default_port)
+        args["port"] = int(raw_port) if raw_port is not None else default_port
+        args["dbname"] = str(properties["dbname"]) if properties.get("dbname") else None
+
+        # User from properties or env
+        user = properties.get("user")
+        if user is None:
+            user = os.environ.get("CLEVERAGENTS_DB_USER")
+        args["user"] = str(user) if user else None
+
+        # Password from properties or env
+        password = properties.get("password")
+        if password is None:
+            password = os.environ.get("CLEVERAGENTS_DB_PASSWORD")
+        args["password"] = str(password) if password else None
+
+    elif resource_type in ("sqlite", "duckdb"):
+        path = properties.get("path", ":memory:")
+        args["path"] = str(path) if path is not None else ":memory:"
+
+    return args
+
+
+# ---------------------------------------------------------------------------
+# Connection validation
+# ---------------------------------------------------------------------------
+
+
+def validate_connection(
+    resource_type: str,
+    connection_args: dict[str, str | int | None],
+) -> tuple[bool, str]:
+    """Test database connectivity and return a safe status message.
+
+    Only SQLite is fully implemented (uses stdlib ``sqlite3``).
+    PostgreSQL, MySQL, and DuckDB return stub messages indicating
+    the driver is not available.
+
+    Error messages mask credentials using the shared redaction module.
+
+    Args:
+        resource_type: Database type name.
+        connection_args: Resolved connection arguments.
+
+    Returns:
+        Tuple of ``(success, message)``.
+    """
+    if resource_type == "sqlite":
+        return _validate_sqlite(connection_args)
+    if resource_type == "duckdb":
+        return _validate_duckdb(connection_args)
+    if resource_type in ("postgres", "mysql"):
+        return _validate_network_db(resource_type, connection_args)
+
+    return (False, f"Unknown database type: {resource_type}")
+
+
+def _validate_sqlite(
+    args: dict[str, str | int | None],
+) -> tuple[bool, str]:
+    """Validate an SQLite connection."""
+    path = str(args.get("path", ":memory:") or ":memory:")
+    try:
+        conn = sqlite3.connect(path)
+        conn.execute("SELECT 1")
+        conn.close()
+        return (True, f"SQLite connection OK: {path}")
+    except Exception as exc:
+        return (False, f"SQLite connection failed: {exc}")
+
+
+def _validate_duckdb(
+    args: dict[str, str | int | None],
+) -> tuple[bool, str]:
+    """Validate a DuckDB connection (stub -- driver not bundled)."""
+    path = str(args.get("path", ":memory:") or ":memory:")
+    try:
+        import importlib
+
+        _duckdb = importlib.import_module("duckdb")
+        conn = _duckdb.connect(path)
+        conn.execute("SELECT 1")
+        conn.close()
+        return (True, f"DuckDB connection OK: {path}")
+    except (ImportError, ModuleNotFoundError):
+        return (
+            True,
+            f"DuckDB driver not installed; config accepted for: {path}",
+        )
+    except Exception as exc:
+        return (False, f"DuckDB connection failed: {exc}")
+
+
+def _validate_network_db(
+    resource_type: str,
+    args: dict[str, str | int | None],
+) -> tuple[bool, str]:
+    """Validate a networked DB connection (stub -- drivers not bundled).
+
+    Credential values are redacted in the returned message.
+    """
+    safe_args = redact_dict(
+        {k: v for k, v in args.items() if v is not None},
+    )
+    conn_str = args.get("connection-string")
+    if conn_str is not None:
+        masked = mask_database_url(str(conn_str))
+        return (
+            True,
+            (f"{resource_type} driver not installed; config accepted for: {masked}"),
+        )
+
+    return (
+        True,
+        (f"{resource_type} driver not installed; config accepted: {safe_args}"),
+    )
+
+
+# ---------------------------------------------------------------------------
+# DatabaseResourceHandler
+# ---------------------------------------------------------------------------
+
+
+class DatabaseResourceHandler(BaseResourceHandler):
+    """Handler for database resource types.
+
+    Provisions a transaction-rollback sandbox for database resources.
+    Supports ``postgres``, ``mysql``, ``sqlite``, and ``duckdb``.
+    """
+
+    _default_strategy = SandboxStrategy.TRANSACTION_ROLLBACK
+    _type_label = "database"

vulture_whitelist.py

@@ -942,6 +942,20 @@ max_iterations  # noqa: B018, F821
 _temporal_score  # noqa: B018, F821
 _extract_node_prefix  # noqa: B018, F821
 
+# Database resource handler — public API (issue #342)
+DatabaseResourceHandler  # noqa: B018, F821
+DATABASE_TYPE_DEFS  # noqa: B018, F821
+DATABASE_TYPE_NAMES  # noqa: B018, F821
+POSTGRES_TYPE_DEF  # noqa: B018, F821
+MYSQL_TYPE_DEF  # noqa: B018, F821
+SQLITE_TYPE_DEF  # noqa: B018, F821
+DUCKDB_TYPE_DEF  # noqa: B018, F821
+resolve_connection_args  # noqa: B018, F821
+validate_connection  # noqa: B018, F821
+
+# Transaction sandbox — public API (issue #342)
+TransactionSandbox  # noqa: B018, F821
+
 # ComponentResolver — pluggable scope chain resolution (#552)
 ComponentResolver  # noqa: B018, F821
 ComponentNotFoundError  # noqa: B018, F821