CHANGELOG.md

@@ -48,6 +48,8 @@
 - Fixed failing Robot Framework integration tests related to security secrets handling.
 - Fixed style check violations across the codebase.
 - Fixed failing unit tests.
+- Added changeset persistence and diff artifact storage for tracking multi-file changes
+  across plan execution phases. (#163)
 - Expanded CONTRIBUTING.md with detailed guidance on the issue creation process, label system,
   ticket lifecycle, pull request requirements, and review/merge process.
 - Added commit scope, quality, and message format guidelines to CONTRIBUTING.md.

alembic/versions/d0_001_changeset_artifacts.py

@@ -0,0 +1,97 @@
+"""Add changeset_entries and tool_invocations tables.
+
+Provides persistent storage for ChangeSet entries and ToolInvocation
+records, replacing the in-memory stores for production use.
+
+Revision ID: d0_001_changeset_artifacts
+Revises: c0_002_merge_skill_registry
+Create Date: 2026-02-24 00:00:00
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "d0_001_changeset_artifacts"
+down_revision: str | Sequence[str] | None = "c0_002_merge_skill_registry"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Create changeset_entries and tool_invocations tables."""
+    op.create_table(
+        "changeset_entries",
+        sa.Column("entry_id", sa.String(26), primary_key=True),
+        sa.Column("changeset_id", sa.String(26), nullable=False),
+        sa.Column("plan_id", sa.String(26), nullable=False),
+        sa.Column("resource_id", sa.String(26), nullable=False),
+        sa.Column("tool_name", sa.String(255), nullable=False),
+        sa.Column("operation", sa.String(20), nullable=False),
+        sa.Column("path", sa.String(1024), nullable=False),
+        sa.Column("before_hash", sa.String(64), nullable=True),
+        sa.Column("after_hash", sa.String(64), nullable=True),
+        sa.Column("before_mode", sa.Integer, nullable=True),
+        sa.Column("after_mode", sa.Integer, nullable=True),
+        sa.Column("timestamp", sa.String(30), nullable=False),
+    )
+    op.create_index(
+        "ix_changeset_entries_changeset_id",
+        "changeset_entries",
+        ["changeset_id"],
+    )
+    op.create_index(
+        "ix_changeset_entries_plan_id",
+        "changeset_entries",
+        ["plan_id"],
+    )
+    op.create_index(
+        "ix_changeset_entries_path",
+        "changeset_entries",
+        ["path"],
+    )
+
+    op.create_table(
+        "tool_invocations",
+        sa.Column("invocation_id", sa.String(26), primary_key=True),
+        sa.Column("changeset_id", sa.String(26), nullable=True),
+        sa.Column("plan_id", sa.String(26), nullable=False),
+        sa.Column("tool_name", sa.String(255), nullable=False),
+        sa.Column("skill_name", sa.String(255), nullable=True),
+        sa.Column("arguments_json", sa.Text, nullable=True),
+        sa.Column("result_json", sa.Text, nullable=True),
+        sa.Column("error", sa.Text, nullable=True),
+        sa.Column("success", sa.Boolean, nullable=False, server_default="1"),
+        sa.Column("duration_ms", sa.Float, nullable=False, server_default="0.0"),
+        sa.Column("started_at", sa.String(30), nullable=False),
+        sa.Column("completed_at", sa.String(30), nullable=True),
+        sa.Column("change_ids_json", sa.Text, nullable=True),
+        sa.Column("sequence_number", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("sandbox_path", sa.String(1024), nullable=True),
+        sa.Column("resource_refs_json", sa.Text, nullable=True),
+        sa.Column("provider_metadata_json", sa.Text, nullable=True),
+    )
+    op.create_index(
+        "ix_tool_invocations_changeset_id",
+        "tool_invocations",
+        ["changeset_id"],
+    )
+    op.create_index(
+        "ix_tool_invocations_plan_id",
+        "tool_invocations",
+        ["plan_id"],
+    )
+    op.create_index(
+        "ix_tool_invocations_tool_name",
+        "tool_invocations",
+        ["tool_name"],
+    )
+
+
+def downgrade() -> None:
+    """Drop changeset_entries and tool_invocations tables."""
+    op.drop_table("tool_invocations")
+    op.drop_table("changeset_entries")

alembic/versions/d0_002_merge_changeset_and_locks.py

@@ -0,0 +1,32 @@
+"""Merge changeset artifacts and concurrency locks heads.
+
+Both d0_001_changeset_artifacts and m4_001_concurrency_locks branched
+from c0_002_merge_skill_registry independently, creating two Alembic
+heads.  This no-op merge migration resolves them into a single head.
+
+Revision ID: d0_002_merge_changeset_and_locks
+Revises: d0_001_changeset_artifacts, m4_001_concurrency_locks
+Create Date: 2026-02-25 00:00:00
+
+"""
+
+from collections.abc import Sequence
+
+# revision identifiers, used by Alembic.
+revision: str = "d0_002_merge_changeset_and_locks"
+down_revision: str | Sequence[str] | None = (
+    "d0_001_changeset_artifacts",
+    "m4_001_concurrency_locks",
+)
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Upgrade schema."""
+    pass
+
+
+def downgrade() -> None:
+    """Downgrade schema."""
+    pass

benchmarks/changeset_persistence_bench.py

@@ -0,0 +1,98 @@
+"""ASV benchmarks for ChangeSet persistence overhead.
+
+Measures the cost of persisting ChangeEntry records and retrieving
+them via the SqliteChangeSetStore, simulating typical plan execution
+workloads.
+"""
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.domain.models.core.change import (
+    ChangeEntry,
+    ChangeOperation,
+)
+from cleveragents.infrastructure.database.changeset_repository import (
+    ChangeSetEntryRepository,
+    SqliteChangeSetStore,
+)
+from cleveragents.infrastructure.database.models import Base
+
+
+def _make_factory() -> tuple[sessionmaker[Session], object]:
+    """Create an in-memory SQLite engine + session factory."""
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        connect_args={"check_same_thread": False},
+    )
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    return factory, engine
+
+
+def _make_entry(plan_id: str, idx: int) -> ChangeEntry:
+    return ChangeEntry(
+        plan_id=plan_id,
+        resource_id="res-001",
+        tool_name="file-write",
+        operation=ChangeOperation.CREATE,
+        path=f"src/file_{idx}.py",
+        after_hash=f"hash_{idx:06d}",
+    )
+
+
+class TimeChangeSetPersistence:
+    """Benchmark suite for ChangeSet persistence operations."""
+
+    timeout = 60.0
+
+    def setup(self) -> None:
+        self.factory, self.engine = _make_factory()
+        self.store = SqliteChangeSetStore(self.factory)
+        self.plan_id = "bench-plan-001"
+        self.changeset_id = self.store.start(self.plan_id)
+        # Pre-populate for retrieval benchmarks
+        for i in range(50):
+            entry = _make_entry(self.plan_id, i)
+            self.store.record(self.changeset_id, entry)
+        session = self.factory()
+        session.commit()
+        session.close()
+
+    def time_record_single_entry(self) -> None:
+        """Time recording a single ChangeEntry."""
+        entry = _make_entry(self.plan_id, 999)
+        self.store.record(self.changeset_id, entry)
+
+    def time_get_changeset_50_entries(self) -> None:
+        """Time retrieving a changeset with 50 entries."""
+        self.store.get(self.changeset_id)
+
+    def time_summarize_changeset(self) -> None:
+        """Time computing summary for a changeset."""
+        self.store.summarize(self.changeset_id)
+
+    def time_start_changeset(self) -> None:
+        """Time starting a new empty changeset."""
+        self.store.start("bench-plan-new")
+
+
+class TimeEntryRepositoryBulk:
+    """Benchmark bulk entry operations."""
+
+    timeout = 60.0
+
+    def setup(self) -> None:
+        self.factory, self.engine = _make_factory()
+        self.repo = ChangeSetEntryRepository(self.factory)
+
+    def time_save_100_entries(self) -> None:
+        """Time saving 100 entries sequentially."""
+        for i in range(100):
+            entry = _make_entry("bench-bulk", i)
+            self.repo.save_entry(f"cs-bulk-{i % 5}", entry)
+
+    def time_retrieve_by_plan(self) -> None:
+        """Time retrieving all entries for a plan."""
+        self.repo.get_entries_for_plan("bench-bulk")

docs/reference/changeset.md

@@ -0,0 +1,156 @@
+# ChangeSet Persistence
+
+## Overview
+
+ChangeSets track every file mutation made during the Execute phase of a plan.
+Each mutation is recorded as a `ChangeEntry` with content hashes, operation
+type, and tool provenance.  Tool executions are recorded as `ToolInvocation`
+records for full auditability.
+
+## Persistence Fields
+
+### changeset_entries table
+
+| Column        | Type         | Description                          |
+|---------------|--------------|--------------------------------------|
+| entry_id      | String(26)   | ULID primary key                     |
+| changeset_id  | String(26)   | ULID of the owning changeset         |
+| plan_id       | String(26)   | ULID of the owning plan              |
+| resource_id   | String(26)   | ULID of the affected resource        |
+| tool_name     | String(255)  | Namespaced tool name                 |
+| operation     | String(20)   | create, modify, delete, or rename    |
+| path          | String(1024) | Repo-relative file path              |
+| before_hash   | String(64)   | SHA-256 before change (nullable)     |
+| after_hash    | String(64)   | SHA-256 after change (nullable)      |
+| before_mode   | Integer      | File mode before change (nullable)   |
+| after_mode    | Integer      | File mode after change (nullable)    |
+| timestamp     | String(30)   | ISO-8601 UTC timestamp               |
+
+### tool_invocations table
+
+| Column              | Type         | Description                      |
+|---------------------|--------------|----------------------------------|
+| invocation_id       | String(26)   | ULID primary key                 |
+| changeset_id        | String(26)   | Optional owning changeset        |
+| plan_id             | String(26)   | Owning plan ULID                 |
+| tool_name           | String(255)  | Namespaced tool name             |
+| skill_name          | String(255)  | Skill that provided the tool     |
+| arguments_json      | Text         | JSON input parameters            |
+| result_json         | Text         | JSON output payload              |
+| error               | Text         | Error message on failure         |
+| success             | Boolean      | Whether execution succeeded      |
+| duration_ms         | Float        | Execution time in milliseconds   |
+| started_at          | String(30)   | ISO-8601 start timestamp         |
+| completed_at        | String(30)   | ISO-8601 completion timestamp    |
+| change_ids_json     | Text         | JSON list of ChangeEntry IDs     |
+| sequence_number     | Integer      | Ordering within a plan           |
+| sandbox_path        | String(1024) | Sandbox root path                |
+| resource_refs_json  | Text         | JSON list of resource IDs        |
+| provider_metadata_json | Text      | JSON provider info               |
+
+## Diff Output Examples
+
+### Plain format (`agents plan diff <plan_id> --format plain`)
+
+```text
+ChangeSet: 01HXYZ123456789ABCDEFG
+Plan: 01HABCDEF123456789XYZ
+Total changes: 3
+
+--- a/src/new_module.py
++++ b/src/new_module.py
+@@ new file @@
++ hash: abc123def456...
+
+--- a/src/existing.py
++++ b/src/existing.py
+@@ modified @@
+- hash: old_hash_val...
++ hash: new_hash_val...
+
+--- a/src/removed.py
++++ b/src/removed.py
+@@ deleted @@
+- hash: file_hash_be...
+```
+
+### JSON format (`agents plan diff <plan_id> --format json`)
+
+```json
+{
+  "changeset_id": "01HXYZ123456789ABCDEFG",
+  "plan_id": "01HABCDEF123456789XYZ",
+  "total_changes": 3,
+  "summary": {
+    "total": 3,
+    "creates": 1,
+    "modifies": 1,
+    "deletes": 1,
+    "renames": 0,
+    "paths_changed": 3,
+    "resources_involved": 1
+  },
+  "entries": [
+    {
+      "path": "src/new_module.py",
+      "operation": "create",
+      "before_hash": null,
+      "after_hash": "abc123def456",
+      "resource_id": "01HRES001",
+      "tool_name": "file-write",
+      "timestamp": "2026-02-24T12:00:00+00:00"
+    }
+  ]
+}
+```
+
+### Artifacts output (`agents plan artifacts <plan_id>`)
+
+```json
+{
+  "plan_id": "01HABCDEF123456789XYZ",
+  "phase": "execute",
+  "processing_state": "complete",
+  "changeset_id": "01HXYZ123456789ABCDEFG",
+  "sandbox_refs": ["sandbox-001"],
+  "changeset_summary": {
+    "total": 3,
+    "creates": 1,
+    "modifies": 1,
+    "deletes": 1,
+    "renames": 0
+  },
+  "files_changed": [
+    {"path": "src/new_module.py", "operation": "create"},
+    {"path": "src/existing.py", "operation": "modify"},
+    {"path": "src/removed.py", "operation": "delete"}
+  ]
+}
+```
+
+## Cleanup Behaviour
+
+Changeset artifacts are automatically cleaned up when:
+
+- **Plan cancel**: `cleanup_changeset()` deletes all persisted entries and
+  invocations for the cancelled plan.
+- **Apply failure**: `cleanup_changeset()` is called after
+  `handle_merge_failure()` to free storage for failed plans.
+
+## API
+
+### SqliteChangeSetStore
+
+Implements the `ChangeSetStore` protocol with SQLite persistence.
+
+```python
+from cleveragents.infrastructure.database.changeset_repository import (
+    SqliteChangeSetStore,
+)
+
+store = SqliteChangeSetStore(session_factory)
+changeset_id = store.start(plan_id)
+store.record(changeset_id, entry)
+changeset = store.get(changeset_id)
+store.delete_for_plan(plan_id)
+```

features/changeset_persistence.feature

@@ -0,0 +1,105 @@
+@unit
+Feature: ChangeSet Persistence with SQLite
+  As a developer using CleverAgents
+  I want changesets and tool invocations persisted to SQLite
+  So that diffs, artifacts, and audit trails survive restarts
+
+  # ---- SqliteChangeSetStore start/record/get ----
+
+  Scenario: SqliteChangeSetStore start creates a new changeset ID
+    Given I have a sqlite-backed changeset store
+    When I persist-start a changeset for plan "plan-persist-1"
+    Then the persisted changeset ID should be a valid ULID
+
+  Scenario: SqliteChangeSetStore record and get round-trip
+    Given I have a sqlite-backed changeset store
+    When I persist-start a changeset for plan "plan-persist-2"
+    And I persist-record a create entry in the changeset
+    And I persist-record a modify entry in the changeset
+    Then persisted changeset get should return 2 entries
+    And the persisted first entry operation should be "create"
+    And the persisted second entry operation should be "modify"
+
+  Scenario: SqliteChangeSetStore get returns None for unknown ID
+    Given I have a sqlite-backed changeset store
+    Then persisted changeset "nonexistent-id" returns None on get
+
+  Scenario: SqliteChangeSetStore get_for_plan returns entries
+    Given I have a sqlite-backed changeset store
+    When I persist-start and populate a changeset for plan "plan-fp-1"
+    Then persisted get_for_plan "plan-fp-1" returns at least 1 changeset
+
+  Scenario: SqliteChangeSetStore summarize returns counts
+    Given I have a sqlite-backed changeset store
+    When I persist-start a changeset for plan "plan-sum-1"
+    And I persist-record a create entry in the changeset
+    And I persist-record a delete entry in the changeset
+    Then persisted summarize should show 2 total
+
+  Scenario: SqliteChangeSetStore summarize returns empty for unknown
+    Given I have a sqlite-backed changeset store
+    Then persisted summarize of changeset "no-such-id" returns empty dict
+
+  # ---- ChangeSetEntryRepository CRUD ----
+
+  Scenario: ChangeSetEntryRepository saves and retrieves entries
+    Given I have a persisted ChangeSetEntryRepository
+    When I persist-save an entry for changeset "cs-repo-1" plan "plan-repo-1"
+    Then persisted entries for changeset "cs-repo-1" count is 1
+
+  Scenario: ChangeSetEntryRepository deletes for plan
+    Given I have a persisted ChangeSetEntryRepository
+    When I persist-save entries for plan "plan-del-1"
+    And I persist-delete entries for plan "plan-del-1"
+    Then persisted entries for plan "plan-del-1" count is 0
+
+  Scenario: ChangeSetEntryRepository deletes for changeset
+    Given I have a persisted ChangeSetEntryRepository
+    When I persist-save an entry for changeset "cs-del-1" plan "plan-del-2"
+    And I persist-delete entries for changeset "cs-del-1"
+    Then persisted entries for changeset "cs-del-1" count is 0
+
+  # ---- ToolInvocationRepository CRUD ----
+
+  Scenario: ToolInvocationRepository saves and retrieves invocations
+    Given I have a persisted ToolInvocationRepository
+    When I persist-save an invocation for plan "plan-inv-1"
+    Then persisted invocations for plan "plan-inv-1" count is 1
+
+  Scenario: ToolInvocationRepository deletes for plan
+    Given I have a persisted ToolInvocationRepository
+    When I persist-save an invocation for plan "plan-inv-del"
+    And I persist-delete invocations for plan "plan-inv-del"
+    Then persisted invocations for plan "plan-inv-del" count is 0
+
+  # ---- Cleanup on cancel ----
+
+  Scenario: Cleanup changeset deletes artifacts for plan
+    Given I have a PlanApplyService backed by sqlite changeset store
+    When I persist-start and populate a changeset for the test plan
+    And I call cleanup_changeset on the persisted test plan
+    Then the persisted changeset entries for the test plan are empty
+
+  # ---- Validation ----
+
+  Scenario: ChangeSetEntryRepository rejects None session_factory
+    When I try creating a persisted ChangeSetEntryRepository with None
+    Then a persist ValueError should be raised
+
+  Scenario: ToolInvocationRepository rejects None session_factory
+    When I try creating a persisted ToolInvocationRepository with None
+    Then a persist ValueError should be raised
+
+  Scenario: SqliteChangeSetStore rejects None session_factory
+    When I try creating a persisted SqliteChangeSetStore with None
+    Then a persist ValueError should be raised
+
+  Scenario: SqliteChangeSetStore start rejects empty plan_id
+    Given I have a sqlite-backed changeset store
+    When I try to persist-start a changeset with empty plan_id
+    Then a persist ValueError should be raised
+
+  Scenario: Cleanup changeset rejects empty plan_id
+    Given I have a PlanApplyService backed by sqlite changeset store
+    When I try to persist-cleanup changeset with empty plan_id
+    Then a persist validation error should be raised

features/steps/changeset_persistence_steps.py

@@ -0,0 +1,346 @@
+"""Step definitions for changeset persistence feature tests."""
+
+from __future__ import annotations
+
+import re
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context  # type: ignore[import-untyped]
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from cleveragents.application.services.plan_apply_service import (
+    PlanApplyService,
+)
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.core.exceptions import ValidationError
+from cleveragents.domain.models.core.change import (
+    ChangeEntry,
+    ChangeOperation,
+    ToolInvocation,
+)
+from cleveragents.infrastructure.database.changeset_repository import (
+    ChangeSetEntryRepository,
+    SqliteChangeSetStore,
+    ToolInvocationRepository,
+)
+from cleveragents.infrastructure.database.models import Base
+
+
+def _make_in_memory_session_factory() -> tuple[sessionmaker[Session], object]:
+    """Create an in-memory SQLite engine + session factory with schema."""
+    engine = create_engine(
+        "sqlite:///:memory:",
+        echo=False,
+        connect_args={"check_same_thread": False},
+    )
+    Base.metadata.create_all(engine)
+    factory: sessionmaker[Session] = sessionmaker(bind=engine)
+    return factory, engine
+
+
+def _make_entry(
+    plan_id: str,
+    operation: ChangeOperation = ChangeOperation.CREATE,
+    path: str = "src/test.py",
+    resource_id: str = "res-001",
+) -> ChangeEntry:
+    """Create a test ChangeEntry."""
+    before_hash = "aaa111" if operation != ChangeOperation.CREATE else None
+    after_hash = "bbb222" if operation != ChangeOperation.DELETE else None
+    return ChangeEntry(
+        plan_id=plan_id,
+        resource_id=resource_id,
+        tool_name="file-write",
+        operation=operation,
+        path=path,
+        before_hash=before_hash,
+        after_hash=after_hash,
+    )
+
+
+# ---- SqliteChangeSetStore steps ----
+
+
+@given("I have a sqlite-backed changeset store")
+def step_sqlite_store(context: Context) -> None:
+    factory, engine = _make_in_memory_session_factory()
+    context.session_factory = factory
+    context.engine = engine
+    context.sqlite_store = SqliteChangeSetStore(factory)
+
+
+@when('I persist-start a changeset for plan "{plan_id}"')
+def step_start_changeset(context: Context, plan_id: str) -> None:
+    context.changeset_id = context.sqlite_store.start(plan_id)
+    context.plan_id = plan_id
+
+
+@then("the persisted changeset ID should be a valid ULID")
+def step_valid_ulid(context: Context) -> None:
+    assert context.changeset_id is not None
+    assert len(context.changeset_id) == 26
+    assert re.match(r"^[0-9A-Z]{26}$", context.changeset_id)
+
+
+@when("I persist-record a create entry in the changeset")
+def step_record_create(context: Context) -> None:
+    entry = _make_entry(context.plan_id, ChangeOperation.CREATE, "src/new.py")
+    context.sqlite_store.record(context.changeset_id, entry)
+
+
+@when("I persist-record a modify entry in the changeset")
+def step_record_modify(context: Context) -> None:
+    entry = _make_entry(context.plan_id, ChangeOperation.MODIFY, "src/existing.py")
+    context.sqlite_store.record(context.changeset_id, entry)
+
+
+@when("I persist-record a delete entry in the changeset")
+def step_record_delete(context: Context) -> None:
+    entry = _make_entry(context.plan_id, ChangeOperation.DELETE, "src/old.py")
+    context.sqlite_store.record(context.changeset_id, entry)
+
+
+@then("persisted changeset get should return {count:d} entries")
+def step_get_entries(context: Context, count: int) -> None:
+    cs = context.sqlite_store.get(context.changeset_id)
+    assert cs is not None
+    assert len(cs.entries) == count
+
+
+@then('the persisted first entry operation should be "{op}"')
+def step_first_op(context: Context, op: str) -> None:
+    cs = context.sqlite_store.get(context.changeset_id)
+    assert cs is not None
+    assert cs.entries[0].operation == op
+
+
+@then('the persisted second entry operation should be "{op}"')
+def step_second_op(context: Context, op: str) -> None:
+    cs = context.sqlite_store.get(context.changeset_id)
+    assert cs is not None
+    assert cs.entries[1].operation == op
+
+
+@then('persisted changeset "{cid}" returns None on get')
+def step_get_none(context: Context, cid: str) -> None:
+    result = context.sqlite_store.get(cid)
+    assert result is None
+
+
+@when('I persist-start and populate a changeset for plan "{plan_id}"')
+def step_start_and_populate(context: Context, plan_id: str) -> None:
+    context.plan_id = plan_id
+    context.changeset_id = context.sqlite_store.start(plan_id)
+    entry = _make_entry(plan_id, ChangeOperation.CREATE, "src/file.py")
+    context.sqlite_store.record(context.changeset_id, entry)
+
+
+@then('persisted get_for_plan "{plan_id}" returns at least {count:d} changeset')
+def step_get_for_plan(context: Context, plan_id: str, count: int) -> None:
+    result = context.sqlite_store.get_for_plan(plan_id)
+    assert len(result) >= count
+
+
+@then("persisted summarize should show {count:d} total")
+def step_summarize(context: Context, count: int) -> None:
+    summary = context.sqlite_store.summarize(context.changeset_id)
+    assert summary.get("total") == count
+
+
+@then('persisted summarize of changeset "{cid}" returns empty dict')
+def step_summarize_empty(context: Context, cid: str) -> None:
+    result = context.sqlite_store.summarize(cid)
+    assert result == {}
+
+
+# ---- ChangeSetEntryRepository steps ----
+
+
+@given("I have a persisted ChangeSetEntryRepository")
+def step_entry_repo(context: Context) -> None:
+    factory, engine = _make_in_memory_session_factory()
+    context.session_factory = factory
+    context.engine = engine
+    context.entry_repo = ChangeSetEntryRepository(factory)
+
+
+@when('I persist-save an entry for changeset "{cs_id}" plan "{plan_id}"')
+def step_save_entry(context: Context, cs_id: str, plan_id: str) -> None:
+    entry = _make_entry(plan_id, ChangeOperation.CREATE, "src/test.py")
+    context.entry_repo.save_entry(cs_id, entry)
+
+
+@then('persisted entries for changeset "{cs_id}" count is {count:d}')
+def step_retrieve_for_changeset(context: Context, cs_id: str, count: int) -> None:
+    entries = context.entry_repo.get_entries_for_changeset(cs_id)
+    assert len(entries) == count
+
+
+@when('I persist-save entries for plan "{plan_id}"')
+def step_save_entries_for_plan(context: Context, plan_id: str) -> None:
+    context.plan_id = plan_id
+    for i in range(3):
+        entry = _make_entry(plan_id, ChangeOperation.CREATE, f"src/f{i}.py")
+        context.entry_repo.save_entry(f"cs-{i}", entry)
+
+
+@when('I persist-delete entries for plan "{plan_id}"')
+def step_delete_for_plan(context: Context, plan_id: str) -> None:
+    context.entry_repo.delete_for_plan(plan_id)
+
+
+@then('persisted entries for plan "{plan_id}" count is {count:d}')
+def step_retrieve_for_plan(context: Context, plan_id: str, count: int) -> None:
+    entries = context.entry_repo.get_entries_for_plan(plan_id)
+    assert len(entries) == count
+
+
+@when('I persist-delete entries for changeset "{cs_id}"')
+def step_delete_for_changeset(context: Context, cs_id: str) -> None:
+    context.entry_repo.delete_for_changeset(cs_id)
+
+
+# ---- ToolInvocationRepository steps ----
+
+
+@given("I have a persisted ToolInvocationRepository")
+def step_inv_repo(context: Context) -> None:
+    factory, engine = _make_in_memory_session_factory()
+    context.session_factory = factory
+    context.engine = engine
+    context.inv_repo = ToolInvocationRepository(factory)
+
+
+@when('I persist-save an invocation for plan "{plan_id}"')
+def step_save_invocation(context: Context, plan_id: str) -> None:
+    context.plan_id = plan_id
+    inv = ToolInvocation(
+        plan_id=plan_id,
+        tool_name="file-write",
+        arguments={"path": "test.py"},
+        success=True,
+    )
+    context.inv_repo.save_invocation(inv, changeset_id="cs-inv")
+
+
+@then('persisted invocations for plan "{plan_id}" count is {count:d}')
+def step_retrieve_invocations(context: Context, plan_id: str, count: int) -> None:
+    invocations = context.inv_repo.get_invocations_for_plan(plan_id)
+    assert len(invocations) == count
+
+
+@when('I persist-delete invocations for plan "{plan_id}"')
+def step_delete_invocations(context: Context, plan_id: str) -> None:
+    context.inv_repo.delete_for_plan(plan_id)
+
+
+# ---- PlanApplyService cleanup steps ----
+
+
+@given("I have a PlanApplyService backed by sqlite changeset store")
+def step_apply_service_with_store(context: Context) -> None:
+    factory, engine = _make_in_memory_session_factory()
+    context.session_factory = factory
+    context.engine = engine
+    context.sqlite_store = SqliteChangeSetStore(factory)
+
+    settings = Settings()
+    lifecycle = PlanLifecycleService(settings=settings)
+    context.apply_service = PlanApplyService(
+        lifecycle_service=lifecycle,
+        changeset_store=context.sqlite_store,
+    )
+
+    lifecycle.create_action(
+        name="local/cleanup-test",
+        description="Cleanup test action",
+        definition_of_done="Tests pass",
+        strategy_actor="local/planner",
+        execution_actor="local/executor",
+    )
+    plan = lifecycle.use_action(action_name="local/cleanup-test")
+    context.test_plan_id = plan.identity.plan_id
+
+
+@when("I persist-start and populate a changeset for the test plan")
+def step_populate_test_changeset(context: Context) -> None:
+    cs_id = context.sqlite_store.start(context.test_plan_id)
+    entry = _make_entry(context.test_plan_id, ChangeOperation.CREATE, "src/test.py")
+    context.sqlite_store.record(cs_id, entry)
+    context.test_changeset_id = cs_id
+
+
+@when("I call cleanup_changeset on the persisted test plan")
+def step_cleanup(context: Context) -> None:
+    context.cleanup_count = context.apply_service.cleanup_changeset(
+        context.test_plan_id,
+    )
+
+
+@then("the persisted changeset entries for the test plan are empty")
+def step_entries_empty(context: Context) -> None:
+    entries = context.sqlite_store.get_for_plan(context.test_plan_id)
+    total_entries = sum(len(cs.entries) for cs in entries)
+    assert total_entries == 0
+
+
+# ---- Validation steps ----
+
+
+@when("I try creating a persisted ChangeSetEntryRepository with None")
+def step_entry_repo_none(context: Context) -> None:
+    try:
+        ChangeSetEntryRepository(None)  # type: ignore[arg-type]
+        context.persist_raised = None
+    except ValueError as exc:
+        context.persist_raised = exc
+
+
+@when("I try creating a persisted ToolInvocationRepository with None")
+def step_inv_repo_none(context: Context) -> None:
+    try:
+        ToolInvocationRepository(None)  # type: ignore[arg-type]
+        context.persist_raised = None
+    except ValueError as exc:
+        context.persist_raised = exc
+
+
+@when("I try creating a persisted SqliteChangeSetStore with None")
+def step_store_none(context: Context) -> None:
+    try:
+        SqliteChangeSetStore(None)  # type: ignore[arg-type]
+        context.persist_raised = None
+    except ValueError as exc:
+        context.persist_raised = exc
+
+
+@then("a persist ValueError should be raised")
+def step_value_error(context: Context) -> None:
+    assert isinstance(context.persist_raised, ValueError)
+
+
+@when("I try to persist-start a changeset with empty plan_id")
+def step_start_empty(context: Context) -> None:
+    try:
+        context.sqlite_store.start("")
+        context.persist_raised = None
+    except ValueError as exc:
+        context.persist_raised = exc
+
+
+@when("I try to persist-cleanup changeset with empty plan_id")
+def step_cleanup_empty(context: Context) -> None:
+    try:
+        context.apply_service.cleanup_changeset("")
+        context.persist_raised = None
+    except (ValidationError, ValueError) as exc:
+        context.persist_raised = exc
+
+
+@then("a persist validation error should be raised")
+def step_validation_error(context: Context) -> None:
+    assert isinstance(context.persist_raised, (ValidationError, ValueError))

robot/changeset_persistence.robot

@@ -0,0 +1,86 @@
+*** Settings ***
+Documentation    Integration tests for ChangeSet persistence via CLI plan artifacts output.
+Library          OperatingSystem
+Library          Process
+Library          Collections
+
+Suite Setup      Set Suite Variables
+Force Tags       changeset    persistence
+
+*** Variables ***
+${PYTHON}        python
+${TIMEOUT}       30s
+
+*** Keywords ***
+Set Suite Variables
+    Set Suite Variable    ${AGENTS}    ${PYTHON} -m cleveragents.cli.main
+
+Run Agents Command
+    [Arguments]    @{args}
+    ${result}=    Run Process    ${PYTHON}    -m    cleveragents.cli.main    @{args}
+    ...    timeout=${TIMEOUT}
+    ...    env:CLEVERAGENTS_TESTING_USE_MOCK_AI=true
+    ...    env:CLEVERAGENTS_AUTO_APPLY_MIGRATIONS=true
+    ...    env:NO_COLOR=1
+    RETURN    ${result}
+
+*** Test Cases ***
+Plan Artifacts Shows Changeset ID Field
+    [Documentation]    Verify plan artifacts output contains changeset_id field.
+    ${result}=    Run Agents Command    plan    artifacts    --help
+    Should Contain    ${result.stdout}    plan_id
+    ...    msg=Plan artifacts help should mention plan_id
+
+Plan Diff Help Shows Format Options
+    [Documentation]    Verify plan diff command exposes format options.
+    ${result}=    Run Agents Command    plan    diff    --help
+    Should Contain    ${result.stdout}    format
+    ...    msg=Plan diff help should mention format option
+
+Plan Artifacts Help Contains Expected Text
+    [Documentation]    Verify plan artifacts help text is present.
+    ${result}=    Run Agents Command    plan    artifacts    --help
+    Should Contain    ${result.stdout}    artifacts
+    ...    msg=Help text should mention artifacts
+
+Changeset Persistence Module Is Importable
+    [Documentation]    Verify the changeset_repository module can be imported.
+    ${result}=    Run Process    ${PYTHON}    -c
+    ...    from cleveragents.infrastructure.database.changeset_repository import SqliteChangeSetStore; print("OK")
+    ...    timeout=${TIMEOUT}
+    ...    env:PYTHONPATH=src
+    Should Be Equal As Strings    ${result.stdout.strip()}    OK
+
+SqliteChangeSetStore Round Trip Via CLI Script
+    [Documentation]    Run a Python script that exercises SqliteChangeSetStore round-trip.
+    ${script}=    Catenate    SEPARATOR=\n
+    ...    import sys
+    ...    from sqlalchemy import create_engine
+    ...    from sqlalchemy.orm import sessionmaker
+    ...    from cleveragents.infrastructure.database.models import Base
+    ...    from cleveragents.infrastructure.database.changeset_repository import SqliteChangeSetStore
+    ...    from cleveragents.domain.models.core.change import ChangeEntry, ChangeOperation
+    ...    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+    ...    Base.metadata.create_all(engine)
+    ...    factory = sessionmaker(bind=engine)
+    ...    store = SqliteChangeSetStore(factory)
+    ...    cs_id = store.start("plan-robot-1")
+    ...    entry = ChangeEntry(plan_id="plan-robot-1", resource_id="res-1", tool_name="file-write", operation=ChangeOperation.CREATE, path="src/test.py", after_hash="abc123")
+    ...    store.record(cs_id, entry)
+    ...    factory().commit()
+    ...    cs = store.get(cs_id)
+    ...    assert cs is not None
+    ...    assert len(cs.entries) == 1
+    ...    summary = store.summarize(cs_id)
+    ...    assert summary["total"] == 1
+    ...    assert summary["creates"] == 1
+    ...    store.delete_for_plan("plan-robot-1")
+    ...    factory().commit()
+    ...    cs2 = store.get(cs_id)
+    ...    assert cs2 is None or len(cs2.entries) == 0
+    ...    print("PASS")
+    ${result}=    Run Process    ${PYTHON}    -c    ${script}
+    ...    timeout=${TIMEOUT}
+    ...    env:PYTHONPATH=src
+    Should Contain    ${result.stdout}    PASS
+    ...    msg=SqliteChangeSetStore round-trip failed: ${result.stderr}

src/cleveragents/application/services/plan_apply_service.py

@@ -666,3 +666,32 @@ class PlanApplyService:
             changeset_id=plan.changeset_id,
             plan_id=plan.identity.plan_id,
         )
+
+    # -- ChangeSet cleanup --------------------------------------------------
+
+    def cleanup_changeset(self, plan_id: str) -> int:
+        """Delete persisted changeset entries and invocations for a plan.
+
+        Called on plan cancel and apply failure to free storage used by
+        artifacts that are no longer needed.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            Number of artifact records deleted.
+        """
+        if not plan_id:
+            raise ValidationError("plan_id must not be empty")
+
+        deleted = 0
+        store = self._changeset_store
+        if store is not None and hasattr(store, "delete_for_plan"):
+            deleted += store.delete_for_plan(plan_id)
+
+        self._logger.info(
+            "Changeset artifacts cleaned up",
+            plan_id=plan_id,
+            records_deleted=deleted,
+        )
+        return deleted

src/cleveragents/infrastructure/database/changeset_repository.py

@@ -0,0 +1,433 @@
+"""ChangeSet persistence repository and SQLite-backed store.
+
+Provides ``ChangeSetEntryRepository`` and ``ToolInvocationRepository``
+for CRUD on the ``changeset_entries`` and ``tool_invocations`` tables,
+plus ``SqliteChangeSetStore`` which implements the
+``ChangeSetStore`` protocol using these repositories.
+
+All repositories follow the session-factory pattern (ADR-007).
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from datetime import datetime
+from typing import Any, cast
+
+from sqlalchemy.exc import DatabaseError as SQLAlchemyDatabaseError
+from sqlalchemy.exc import OperationalError
+from sqlalchemy.orm import Session
+from ulid import ULID
+
+from cleveragents.core.exceptions import DatabaseError
+from cleveragents.core.retry_patterns import (
+    retry_database_operation as database_retry,
+)
+from cleveragents.domain.models.core.change import (
+    ChangeEntry,
+    ChangeOperation,
+    SpecChangeSet,
+    ToolInvocation,
+)
+from cleveragents.infrastructure.database.models import (
+    ChangeSetEntryModel,
+    ToolInvocationModel,
+)
+
+
+class ChangeSetEntryRepository:
+    """Repository for persisting ChangeEntry records.
+
+    Uses the session-factory pattern: each public method obtains
+    a session from the factory.  Callers are responsible for commit.
+    """
+
+    def __init__(
+        self,
+        session_factory: Callable[[], Session],
+    ) -> None:
+        """Initialise with a callable returning a SQLAlchemy Session."""
+        if session_factory is None:
+            raise ValueError("session_factory must not be None")
+        self._sf = session_factory
+
+    def _session(self) -> Session:
+        return self._sf()
+
+    @database_retry
+    def save_entry(
+        self,
+        changeset_id: str,
+        entry: ChangeEntry,
+    ) -> None:
+        """Persist a single ChangeEntry row."""
+        if not changeset_id:
+            raise ValueError("changeset_id must not be empty")
+        if not isinstance(entry, ChangeEntry):
+            raise TypeError("entry must be a ChangeEntry instance")
+
+        session = self._session()
+        try:
+            model = ChangeSetEntryModel(
+                entry_id=entry.entry_id,
+                changeset_id=changeset_id,
+                plan_id=entry.plan_id,
+                resource_id=entry.resource_id,
+                tool_name=entry.tool_name,
+                operation=entry.operation,
+                path=entry.path,
+                before_hash=entry.before_hash,
+                after_hash=entry.after_hash,
+                before_mode=entry.before_mode,
+                after_mode=entry.after_mode,
+                timestamp=entry.timestamp.isoformat(),
+            )
+            session.add(model)
+            session.flush()
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to save changeset entry: {exc}") from exc
+
+    @database_retry
+    def get_entries_for_changeset(
+        self,
+        changeset_id: str,
+    ) -> list[ChangeEntry]:
+        """Return all ChangeEntry records for a changeset."""
+        if not changeset_id:
+            raise ValueError("changeset_id must not be empty")
+
+        session = self._session()
+        try:
+            rows = (
+                session.query(ChangeSetEntryModel)
+                .filter_by(changeset_id=changeset_id)
+                .order_by(ChangeSetEntryModel.timestamp)
+                .all()
+            )
+            return [self._to_domain(r) for r in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to get entries: {exc}") from exc
+
+    @database_retry
+    def get_entries_for_plan(
+        self,
+        plan_id: str,
+    ) -> list[ChangeEntry]:
+        """Return all ChangeEntry records for a plan."""
+        if not plan_id:
+            raise ValueError("plan_id must not be empty")
+
+        session = self._session()
+        try:
+            rows = (
+                session.query(ChangeSetEntryModel)
+                .filter_by(plan_id=plan_id)
+                .order_by(ChangeSetEntryModel.timestamp)
+                .all()
+            )
+            return [self._to_domain(r) for r in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to get entries for plan: {exc}") from exc
+
+    @database_retry
+    def delete_for_changeset(self, changeset_id: str) -> int:
+        """Delete all entries for a changeset."""
+        if not changeset_id:
+            raise ValueError("changeset_id must not be empty")
+
+        session = self._session()
+        try:
+            count: int = (
+                session.query(ChangeSetEntryModel)
+                .filter_by(changeset_id=changeset_id)
+                .delete()
+            )
+            session.flush()
+            return count
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to delete entries: {exc}") from exc
+
+    @database_retry
+    def delete_for_plan(self, plan_id: str) -> int:
+        """Delete all entries for a plan."""
+        if not plan_id:
+            raise ValueError("plan_id must not be empty")
+
+        session = self._session()
+        try:
+            count: int = (
+                session.query(ChangeSetEntryModel).filter_by(plan_id=plan_id).delete()
+            )
+            session.flush()
+            return count
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to delete entries: {exc}") from exc
+
+    @staticmethod
+    def _to_domain(row: ChangeSetEntryModel) -> ChangeEntry:
+        """Convert a DB row to a ChangeEntry domain object."""
+        bh = cast("str | None", row.before_hash)
+        ah = cast("str | None", row.after_hash)
+        bm_raw = cast("int | None", row.before_mode)
+        am_raw = cast("int | None", row.after_mode)
+        return ChangeEntry(
+            entry_id=cast(str, row.entry_id),
+            plan_id=cast(str, row.plan_id),
+            resource_id=cast(str, row.resource_id),
+            tool_name=cast(str, row.tool_name),
+            operation=ChangeOperation(cast(str, row.operation)),
+            path=cast(str, row.path),
+            before_hash=bh if bh else None,
+            after_hash=ah if ah else None,
+            before_mode=bm_raw,
+            after_mode=am_raw,
+            timestamp=datetime.fromisoformat(
+                cast(str, row.timestamp),
+            ),
+        )
+
+
+class ToolInvocationRepository:
+    """Repository for persisting ToolInvocation records."""
+
+    def __init__(
+        self,
+        session_factory: Callable[[], Session],
+    ) -> None:
+        if session_factory is None:
+            raise ValueError("session_factory must not be None")
+        self._sf = session_factory
+
+    def _session(self) -> Session:
+        return self._sf()
+
+    @database_retry
+    def save_invocation(
+        self,
+        invocation: ToolInvocation,
+        changeset_id: str | None = None,
+    ) -> None:
+        """Persist a single ToolInvocation row."""
+        if not isinstance(invocation, ToolInvocation):
+            raise TypeError("invocation must be a ToolInvocation instance")
+
+        session = self._session()
+        try:
+            result_json: str | None = None
+            if invocation.result is not None:
+                result_json = json.dumps(
+                    invocation.result,
+                    default=str,
+                )
+
+            completed_iso: str | None = None
+            if invocation.completed_at is not None:
+                completed_iso = invocation.completed_at.isoformat()
+
+            pm_json: str | None = None
+            if invocation.provider_metadata is not None:
+                pm_json = json.dumps(
+                    invocation.provider_metadata,
+                    default=str,
+                )
+
+            model = ToolInvocationModel(
+                invocation_id=invocation.invocation_id,
+                changeset_id=changeset_id,
+                plan_id=invocation.plan_id,
+                tool_name=invocation.tool_name,
+                skill_name=invocation.skill_name,
+                arguments_json=json.dumps(invocation.arguments),
+                result_json=result_json,
+                error=invocation.error,
+                success=invocation.success,
+                duration_ms=invocation.duration_ms,
+                started_at=invocation.started_at.isoformat(),
+                completed_at=completed_iso,
+                change_ids_json=json.dumps(invocation.change_ids),
+                sequence_number=invocation.sequence_number,
+                sandbox_path=invocation.sandbox_path,
+                resource_refs_json=json.dumps(
+                    invocation.resource_refs,
+                ),
+                provider_metadata_json=pm_json,
+            )
+            session.add(model)
+            session.flush()
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to save tool invocation: {exc}") from exc
+
+    @database_retry
+    def get_invocations_for_plan(
+        self,
+        plan_id: str,
+    ) -> list[ToolInvocation]:
+        """Return all ToolInvocation records for a plan."""
+        if not plan_id:
+            raise ValueError("plan_id must not be empty")
+
+        session = self._session()
+        try:
+            rows = (
+                session.query(ToolInvocationModel)
+                .filter_by(plan_id=plan_id)
+                .order_by(ToolInvocationModel.sequence_number)
+                .all()
+            )
+            return [self._to_domain(r) for r in rows]
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            raise DatabaseError(f"Failed to get invocations: {exc}") from exc
+
+    @database_retry
+    def delete_for_plan(self, plan_id: str) -> int:
+        """Delete all invocations for a plan."""
+        if not plan_id:
+            raise ValueError("plan_id must not be empty")
+
+        session = self._session()
+        try:
+            count: int = (
+                session.query(ToolInvocationModel).filter_by(plan_id=plan_id).delete()
+            )
+            session.flush()
+            return count
+        except (OperationalError, SQLAlchemyDatabaseError) as exc:
+            session.rollback()
+            raise DatabaseError(f"Failed to delete invocations: {exc}") from exc
+
+    @staticmethod
+    def _to_domain(row: ToolInvocationModel) -> ToolInvocation:
+        """Convert a DB row to a ToolInvocation domain object."""
+        args_raw = cast("str | None", row.arguments_json)
+        args: dict[str, Any] = json.loads(args_raw) if args_raw else {}
+        res_raw = cast("str | None", row.result_json)
+        result: dict[str, Any] | None = json.loads(res_raw) if res_raw else None
+        cids_raw = cast("str | None", row.change_ids_json)
+        change_ids: list[str] = json.loads(cids_raw) if cids_raw else []
+        rr_raw = cast("str | None", row.resource_refs_json)
+        resource_refs: list[str] = json.loads(rr_raw) if rr_raw else []
+        pm_raw = cast("str | None", row.provider_metadata_json)
+        provider_meta: dict[str, Any] | None = json.loads(pm_raw) if pm_raw else None
+
+        dur_raw = cast("float | None", row.duration_ms)
+        seq_raw = cast("int | None", row.sequence_number)
+
+        completed_raw = cast("str | None", row.completed_at)
+        completed_dt = datetime.fromisoformat(completed_raw) if completed_raw else None
+
+        return ToolInvocation(
+            invocation_id=cast(str, row.invocation_id),
+            plan_id=cast(str, row.plan_id),
+            tool_name=cast(str, row.tool_name),
+            skill_name=cast("str | None", row.skill_name),
+            arguments=args,
+            result=result,
+            error=cast("str | None", row.error),
+            success=cast(bool, row.success),
+            duration_ms=float(dur_raw) if dur_raw else 0.0,
+            started_at=datetime.fromisoformat(
+                cast(str, row.started_at),
+            ),
+            completed_at=completed_dt,
+            change_ids=change_ids,
+            sequence_number=int(seq_raw) if seq_raw else 0,
+            sandbox_path=cast("str | None", row.sandbox_path),
+            resource_refs=resource_refs,
+            provider_metadata=provider_meta,
+        )
+
+
+class SqliteChangeSetStore:
+    """SQLite-backed ``ChangeSetStore`` implementation.
+
+    Satisfies the ``ChangeSetStore`` protocol from
+    ``cleveragents.domain.models.core.change``.
+    """
+
+    def __init__(
+        self,
+        session_factory: Callable[[], Session],
+    ) -> None:
+        if session_factory is None:
+            raise ValueError("session_factory must not be None")
+        self._entry_repo = ChangeSetEntryRepository(session_factory)
+        self._plan_map: dict[str, str] = {}
+
+    def start(self, plan_id: str) -> str:
+        """Create a new empty ChangeSet for *plan_id*."""
+        if not plan_id:
+            raise ValueError("plan_id must not be empty")
+        changeset_id = str(ULID())
+        self._plan_map[changeset_id] = plan_id
+        return changeset_id
+
+    def record(
+        self,
+        changeset_id: str,
+        entry: ChangeEntry,
+    ) -> None:
+        """Persist *entry* under the identified changeset."""
+        if not changeset_id:
+            raise ValueError("changeset_id must not be empty")
+        self._entry_repo.save_entry(changeset_id, entry)
+
+    def get(self, changeset_id: str) -> SpecChangeSet | None:
+        """Retrieve a ChangeSet by ID, or ``None``."""
+        if not changeset_id:
+            return None
+
+        entries = self._entry_repo.get_entries_for_changeset(
+            changeset_id,
+        )
+        if not entries:
+            plan_id = self._plan_map.get(changeset_id, "")
+            if not plan_id:
+                return None
+            return SpecChangeSet(
+                changeset_id=changeset_id,
+                plan_id=plan_id,
+            )
+
+        plan_id = entries[0].plan_id
+        return SpecChangeSet(
+            changeset_id=changeset_id,
+            plan_id=plan_id,
+            entries=entries,
+        )
+
+    def get_for_plan(
+        self,
+        plan_id: str,
+    ) -> list[SpecChangeSet]:
+        """Return all ChangeSets associated with *plan_id*."""
+        if not plan_id:
+            return []
+
+        entries = self._entry_repo.get_entries_for_plan(plan_id)
+        if not entries:
+            return []
+
+        return [
+            SpecChangeSet(plan_id=plan_id, entries=entries),
+        ]
+
+    def summarize(
+        self,
+        changeset_id: str,
+    ) -> dict[str, Any]:
+        """Return summary counts for a changeset."""
+        cs = self.get(changeset_id)
+        if cs is None:
+            return {}
+        return cs.summary()
+
+    def delete_for_plan(self, plan_id: str) -> int:
+        """Delete all changeset entries for a plan."""
+        if not plan_id:
+            raise ValueError("plan_id must not be empty")
+        return self._entry_repo.delete_for_plan(plan_id)

src/cleveragents/infrastructure/database/models.py

@@ -2464,3 +2464,81 @@ def get_session(engine: Any) -> Any:
     """
     Session = sessionmaker(bind=engine)
     return Session()
+
+
+# ---------------------------------------------------------------------------
+# ChangeSet Artifact Models (Stage D0 - migration d0_001)
+# ---------------------------------------------------------------------------
+
+
+class ChangeSetEntryModel(Base):  # type: ignore[misc]
+    """Database model for persisted ChangeEntry records.
+
+    Each row represents a single file mutation captured during the
+    Execute phase of a plan.  Rows are keyed by a ULID ``entry_id``
+    and grouped by ``changeset_id`` / ``plan_id`` for efficient
+    retrieval during diff and artifacts output.
+
+    Table: ``changeset_entries``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "changeset_entries"
+
+    entry_id = Column(String(26), primary_key=True)
+    changeset_id = Column(String(26), nullable=False)
+    plan_id = Column(String(26), nullable=False)
+    resource_id = Column(String(26), nullable=False)
+    tool_name = Column(String(255), nullable=False)
+    operation = Column(String(20), nullable=False)
+    path = Column(String(1024), nullable=False)
+    before_hash = Column(String(64), nullable=True)
+    after_hash = Column(String(64), nullable=True)
+    before_mode = Column(Integer, nullable=True)
+    after_mode = Column(Integer, nullable=True)
+    timestamp = Column(String(30), nullable=False)
+
+    __table_args__ = (
+        Index("ix_changeset_entries_changeset_id", "changeset_id"),
+        Index("ix_changeset_entries_plan_id", "plan_id"),
+        Index("ix_changeset_entries_path", "path"),
+    )
+
+
+class ToolInvocationModel(Base):  # type: ignore[misc]
+    """Database model for persisted ToolInvocation records.
+
+    Each row records a single tool execution for auditability.
+    Linked to a plan and optionally to a changeset.  Arguments,
+    results, change IDs, resource refs, and provider metadata
+    are stored as JSON text fields.
+
+    Table: ``tool_invocations``
+    """
+
+    __allow_unmapped__ = True
+    __tablename__ = "tool_invocations"
+
+    invocation_id = Column(String(26), primary_key=True)
+    changeset_id = Column(String(26), nullable=True)
+    plan_id = Column(String(26), nullable=False)
+    tool_name = Column(String(255), nullable=False)
+    skill_name = Column(String(255), nullable=True)
+    arguments_json = Column(Text, nullable=True)
+    result_json = Column(Text, nullable=True)
+    error = Column(Text, nullable=True)
+    success = Column(Boolean, nullable=False, default=True)
+    duration_ms = Column(Float, nullable=False, default=0.0)
+    started_at = Column(String(30), nullable=False)
+    completed_at = Column(String(30), nullable=True)
+    change_ids_json = Column(Text, nullable=True)
+    sequence_number = Column(Integer, nullable=False, default=0)
+    sandbox_path = Column(String(1024), nullable=True)
+    resource_refs_json = Column(Text, nullable=True)
+    provider_metadata_json = Column(Text, nullable=True)
+
+    __table_args__ = (
+        Index("ix_tool_invocations_changeset_id", "changeset_id"),
+        Index("ix_tool_invocations_plan_id", "plan_id"),
+        Index("ix_tool_invocations_tool_name", "tool_name"),
+    )