From e3fcce413b481fc4a8ece047fc7e411ae5b83056 Mon Sep 17 00:00:00 2001 From: Jeffrey Phillips Freeman Date: Wed, 25 Feb 2026 01:50:37 +0000 Subject: [PATCH] feat(changeset): persist changesets and diff artifacts Add SQLite persistence for ChangeSet entries and ToolInvocation records via new changeset_repository module implementing the ChangeSetStore protocol. - Alembic migration d0_001 creates changeset_entries and tool_invocations tables - SqliteChangeSetStore, ChangeSetEntryRepository, ToolInvocationRepository - PlanApplyService.cleanup_changeset() for cancel/failure cleanup - Behave tests (17 scenarios), Robot tests (5 cases), ASV benchmarks - Reference documentation in docs/reference/changeset.md Closes #163 --- CHANGELOG.md | 2 + .../versions/d0_001_changeset_artifacts.py | 97 ++++ .../d0_002_merge_changeset_and_locks.py | 32 ++ benchmarks/changeset_persistence_bench.py | 98 ++++ docs/reference/changeset.md | 156 +++++++ features/changeset_persistence.feature | 105 +++++ features/steps/changeset_persistence_steps.py | 346 ++++++++++++++ robot/changeset_persistence.robot | 86 ++++ .../services/plan_apply_service.py | 29 ++ .../database/changeset_repository.py | 433 ++++++++++++++++++ .../infrastructure/database/models.py | 78 ++++ 11 files changed, 1462 insertions(+) create mode 100644 alembic/versions/d0_001_changeset_artifacts.py create mode 100644 alembic/versions/d0_002_merge_changeset_and_locks.py create mode 100644 benchmarks/changeset_persistence_bench.py create mode 100644 docs/reference/changeset.md create mode 100644 features/changeset_persistence.feature create mode 100644 features/steps/changeset_persistence_steps.py create mode 100644 robot/changeset_persistence.robot create mode 100644 src/cleveragents/infrastructure/database/changeset_repository.py diff --git a/CHANGELOG.md b/CHANGELOG.md index daaf90225..11fa47fbe 100644 --- a/CHANGELOG.md +++ b/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. diff --git a/alembic/versions/d0_001_changeset_artifacts.py b/alembic/versions/d0_001_changeset_artifacts.py new file mode 100644 index 000000000..a02face04 --- /dev/null +++ b/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") diff --git a/alembic/versions/d0_002_merge_changeset_and_locks.py b/alembic/versions/d0_002_merge_changeset_and_locks.py new file mode 100644 index 000000000..1f4db8447 --- /dev/null +++ b/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 diff --git a/benchmarks/changeset_persistence_bench.py b/benchmarks/changeset_persistence_bench.py new file mode 100644 index 000000000..6a8728d9d --- /dev/null +++ b/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") diff --git a/docs/reference/changeset.md b/docs/reference/changeset.md new file mode 100644 index 000000000..924057a6f --- /dev/null +++ b/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 --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 --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 `) + +```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) +``` diff --git a/features/changeset_persistence.feature b/features/changeset_persistence.feature new file mode 100644 index 000000000..868d9553d --- /dev/null +++ b/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 diff --git a/features/steps/changeset_persistence_steps.py b/features/steps/changeset_persistence_steps.py new file mode 100644 index 000000000..a0fe739d7 --- /dev/null +++ b/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)) diff --git a/robot/changeset_persistence.robot b/robot/changeset_persistence.robot new file mode 100644 index 000000000..0ab56f754 --- /dev/null +++ b/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} diff --git a/src/cleveragents/application/services/plan_apply_service.py b/src/cleveragents/application/services/plan_apply_service.py index 5c156774d..ced1a6e66 100644 --- a/src/cleveragents/application/services/plan_apply_service.py +++ b/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 diff --git a/src/cleveragents/infrastructure/database/changeset_repository.py b/src/cleveragents/infrastructure/database/changeset_repository.py new file mode 100644 index 000000000..8ce860010 --- /dev/null +++ b/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) diff --git a/src/cleveragents/infrastructure/database/models.py b/src/cleveragents/infrastructure/database/models.py index dbc97450c..08c10b160 100644 --- a/src/cleveragents/infrastructure/database/models.py +++ b/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"), + )