From f678d611bb2e96043f8b80b93127b8bc1623765e Mon Sep 17 00:00:00 2001 From: Luis Mendes Date: Tue, 24 Mar 2026 03:33:08 +0000 Subject: [PATCH] feat(db): add correction_attempts table per specification DDL Added CorrectionAttemptModel SQLAlchemy model with all spec-defined columns (correction_attempt_id, plan_id, original_decision_id, new_decision_id, mode, guidance, archived_artifacts_path, state, created_at, completed_at). Added FK constraints to v3_plans and decisions tables. Created Alembic migration and idx_corrections_plan index. Added repository layer for CRUD operations. Addressed code review feedback (rounds 1-12): - Replaced Any type annotations with typed CorrectionAttemptRecord signatures using TYPE_CHECKING imports. - Replaced fragile time.sleep with deterministic timestamps. - Added cascade deletion test with PRAGMA foreign_keys=ON. - Changed update_state() to accept typed enum and datetime params. - Added guidance non-empty validator with max_length=10_000. - Added spec-aligned server_default for created_at column. - Fixed timezone handling in to_domain() and from_domain(). - Added spec-defined lifecycle state transition validation via validate_correction_state_transition() domain function. - Improved FK-violation error messages in create() and update_state(). - Normalised timestamp to millisecond precision matching SQLite server_default strftime('%f') output. - Added auto-set completed_at on terminal transitions. - Added CorrectionAttemptRecord field validators (strip, non-empty). - Changed original_decision_id FK from CASCADE to RESTRICT matching spec DDL default, preserving correction audit trail. - Added input validation in update_state() for new_decision_id and archived_artifacts_path (empty/whitespace rejection). - Fixed dirty-session bug by moving validation before ORM mutations. - Extracted _SQLITE_TIMESTAMP_MS_LEN constant for timestamp truncation. Addressed thirteenth code review feedback: - Changed InvalidCorrectionStateTransitionError base class from DatabaseError to BusinessRuleViolation per CONTRIBUTING.md exception semantics (state transition is a business rule, not a database error; prevents incorrect retries by @database_retry decorator). - Changed new_decision_id FK from SET NULL to RESTRICT matching the spec DDL default (no ON DELETE clause) and consistent with the RESTRICT approach used for original_decision_id. - Changed update_state() input validation for new_decision_id and archived_artifacts_path from DatabaseError to ValueError per CONTRIBUTING.md argument validation guidelines. - Defensive to_domain() coercion now defaults corrupted state to 'failed' (terminal) instead of 'pending', preventing re-execution of completed/failed corrections with corrupted DB values. - Extracted format_sqlite_timestamp() helper and SQLITE_TIMESTAMP_MS_LEN public constant, removing duplicated timestamp formatting logic between from_domain() and update_state(). - Added code comment explaining CASCADE on plan_id FK as a codebase convention deviation from spec DDL default. - Added BDD scenario verifying RESTRICT FK on original_decision_id blocks decision deletion. - Replaced weak cross-plan isolation test with stronger two-plan scenario verifying list_by_plan returns only each plan's attempts. - Fixed hardcoded assertion in step_check_archived_path to use context variable. - 45 BDD scenarios and 5 Robot integration tests. Addressed fourteenth code review feedback: - Added ORM-level relationship(cascade="all, delete-orphan") on LifecyclePlanModel for CorrectionAttemptModel, consistent with all other v3_plans child tables, ensuring ORM-level cascade deletes work even when SQLite FK enforcement is disabled. - Added defensive to_domain() coercion for corrupted guidance column (defaults to "[corrupted]" with warning log), consistent with existing mode/state coercion pattern. - Added ValueError guard in format_sqlite_timestamp() rejecting naive datetimes per CONTRIBUTING.md fail-fast argument validation. - Fixed stale spec DDL line reference in CorrectionAttemptModel docstring. - Fixed duplicated docstring on SQLITE_TIMESTAMP_MS_LEN constant. Addressed fifteenth code review feedback: - Fixed update_state() to defensively handle corrupted DB state values via try/except ValueError coercion to FAILED terminal state with warning log, consistent with to_domain() defensive coercion pattern. - Strengthened RESTRICT FK BDD assertion to verify exception type (IntegrityError/DatabaseError) instead of only checking presence. - Split multi-When/Then cross-plan isolation BDD scenario into idiomatic single-When/Then scenarios per Gherkin best practice. - 53 BDD scenarios (was 45) and 5 Robot integration tests. ISSUES CLOSED: #920 --- CHANGELOG.md | 117 ++ .../m8_001_correction_attempts_table.py | 86 ++ .../correction_attempt_persistence.feature | 401 ++++++ .../correction_attempt_persistence_steps.py | 1263 +++++++++++++++++ robot/correction_attempt_persistence.robot | 44 + .../helper_correction_attempt_persistence.py | 317 +++++ .../domain/models/core/__init__.py | 10 + .../domain/models/core/correction.py | 164 +++ .../infrastructure/database/models.py | 237 ++++ .../infrastructure/database/repositories.py | 316 +++++ .../infrastructure/database/unit_of_work.py | 14 + 11 files changed, 2969 insertions(+) create mode 100644 alembic/versions/m8_001_correction_attempts_table.py create mode 100644 features/correction_attempt_persistence.feature create mode 100644 features/steps/correction_attempt_persistence_steps.py create mode 100644 robot/correction_attempt_persistence.robot create mode 100644 robot/helper_correction_attempt_persistence.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e192b76a0..3141033b1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,123 @@ ## Unreleased +- Added `correction_attempts` table per specification DDL with + `CorrectionAttemptModel` ORM, `CorrectionAttemptRecord` domain model, + `CorrectionAttemptRepository` CRUD layer, Alembic migration, and + `CorrectionAttemptState` enum. Repository `update_state()` accepts + typed `CorrectionAttemptState` enum and `datetime` parameters and + enforces the spec lifecycle (`pending → executing → complete|failed`) + via `InvalidCorrectionStateTransitionError`. + `CorrectionAttemptRecord.guidance` validates non-empty with + `max_length=10_000`. `created_at` column includes spec-aligned + `server_default`; `to_domain()` normalises naive timestamps to UTC. + State transition validation extracted to domain-level + `validate_correction_state_transition()` function with + `CORRECTION_ATTEMPT_VALID_TRANSITIONS` and + `CORRECTION_ATTEMPT_TERMINAL_STATES` constants. + `update_state()` rejects `completed_at` on non-terminal transitions. + Improved FK-violation error messages in `create()` and `update_state()`. + Normalised timestamp format in `from_domain()` to millisecond precision + (`SS.mmm`) matching SQLite `server_default` `strftime('%f')` output + for consistent string-based ordering. + `CORRECTION_ATTEMPT_VALID_TRANSITIONS` and + `CORRECTION_ATTEMPT_TERMINAL_STATES` now use typed + `CorrectionAttemptState` enum keys/values. + Updated repository module docstring tables. + `from_domain()` normalises timestamps to UTC via `astimezone(UTC)` + before formatting, preventing silent data loss for non-UTC datetimes. + `update_state()` auto-sets `completed_at` when transitioning to + terminal states if not explicitly provided. + `CorrectionAttemptRecord` `plan_id` and `original_decision_id` + validators now return stripped values, preventing whitespace-padded + IDs from causing FK lookup failures. + Added new domain exports to `__init__.py` `__all__`. + Aligned `update_state()` `completed_at` timestamp to millisecond + precision for consistency. + Improved FK-violation error message in `update_state()` to avoid + misleading reference when `new_decision_id` is `None`. + Removed unnecessary `session.rollback()` in read-only repository + methods (`get()`, `list_by_plan()`) for consistency with other repos. + Added `created_at` and `completed_at` Pydantic validators on + `CorrectionAttemptRecord` to normalise naive datetimes to UTC, + preventing `ValueError` in `from_domain()` `astimezone()` calls. + Added defensive enum coercion in `CorrectionAttemptModel.to_domain()` + with warning-level logging for invalid `mode`/`state` DB values, + consistent with `LifecyclePlanModel.to_domain()` pattern. + Moved `CorrectionAttemptState` from `TYPE_CHECKING`-only to runtime + import in the repository module, removing redundant in-method import. + Changed `original_decision_id` FK from `CASCADE` to `RESTRICT` + matching the spec DDL default and the codebase convention for + non-dependency FK references to decisions, preserving correction + audit trail when decisions are cleaned up. + Added `new_decision_id` strip-and-validate field validator matching + the pattern used for `plan_id` and `original_decision_id`. + Added input validation for `new_decision_id` in `update_state()` + rejecting empty and whitespace-only values per CONTRIBUTING.md + argument validation guidelines. + Fixed BDD mode-validation scenario to use dedicated `Then` step + with field-level assertion instead of reusing guidance error step. + Moved `new_decision_id` and `archived_artifacts_path` argument + validation in `update_state()` before any ORM row mutations per + CONTRIBUTING.md early-validation guidelines, preventing dirty + session state on validation failure. + Added `archived_artifacts_path` empty/whitespace-only rejection + in `update_state()` matching the `new_decision_id` validation + pattern per CONTRIBUTING.md argument validation guidelines. + 43 BDD scenarios and 5 Robot integration tests including cascade + deletion, terminal-state rejection, failed-path transition, guidance + validation, max-length boundary, min-length boundary, not-found + update, FK-violation update, completed_at guard, timezone + normalization, archived_artifacts_path round-trip, delete-in-complete- + state, cross-plan list isolation, auto-set completed_at on terminal + transition, FK violation on create, invalid mode rejection, + whitespace/empty `new_decision_id` rejection, combined field + update, self-transition rejection, and empty/whitespace + `archived_artifacts_path` rejection. + Fixed `update_state()` bug where `archived_artifacts_path` was + stored without stripping leading/trailing whitespace, unlike + `new_decision_id` which correctly used the stripped value. + Extracted `SQLITE_TIMESTAMP_MS_LEN` constant and + `format_sqlite_timestamp()` helper for millisecond-precision + timestamp formatting, used by both `from_domain()` and + `update_state()`. + Changed `InvalidCorrectionStateTransitionError` base class from + `DatabaseError` to `BusinessRuleViolation` per CONTRIBUTING.md + exception semantics (state transition is a business rule, not a + database error). + Changed `new_decision_id` FK from `SET NULL` to `RESTRICT` matching + the spec DDL default (no ON DELETE clause) and consistent with + `original_decision_id`. + Changed `update_state()` input validation for `new_decision_id` and + `archived_artifacts_path` from `DatabaseError` to `ValueError` per + CONTRIBUTING.md argument validation guidelines. + Defensive `to_domain()` coercion now defaults corrupted state to + `failed` (terminal) instead of `pending`, preventing re-execution + of completed/failed corrections with corrupted DB values. + 45 BDD scenarios (was 43) with new RESTRICT FK test for + `original_decision_id` and stronger cross-plan isolation test. + Added ORM-level `relationship(cascade="all, delete-orphan")` on + `LifecyclePlanModel` for `CorrectionAttemptModel`, consistent with + all other `v3_plans` child tables, ensuring ORM-level cascade + deletes work even when SQLite FK enforcement is disabled. + Added defensive `to_domain()` coercion for corrupted `guidance` + column (defaults to `"[corrupted]"` with warning log), consistent + with existing mode/state coercion pattern. + Added `ValueError` guard in `format_sqlite_timestamp()` rejecting + naive datetimes per CONTRIBUTING.md fail-fast argument validation. + Fixed `update_state()` to defensively handle corrupted DB state + values (coerces to `failed` terminal with warning log), consistent + with `to_domain()` defensive coercion pattern. + Strengthened RESTRICT FK BDD assertion to verify exception type + (`IntegrityError`/`DatabaseError`) instead of only checking + presence. + Split multi-When/Then cross-plan isolation BDD scenario into + idiomatic single-When/Then scenarios. + 53 BDD scenarios (was 45) with new defensive `to_domain()` coercion + tests (corrupted mode/state/guidance), `format_sqlite_timestamp()` + naive datetime rejection, domain model naive datetime normalisation, + and corrupted DB state handling in `update_state()`. + (#920) - Added TDD bug-capture tests for bug #1141 — session create does not persist into subsequent session list output. Added a Behave scenario and Robot E2E test with required tags (`@tdd_bug`, `@tdd_bug_1141`, `@tdd_expected_fail` / diff --git a/alembic/versions/m8_001_correction_attempts_table.py b/alembic/versions/m8_001_correction_attempts_table.py new file mode 100644 index 000000000..62a42a1d6 --- /dev/null +++ b/alembic/versions/m8_001_correction_attempts_table.py @@ -0,0 +1,86 @@ +"""Create correction_attempts table. + +Adds the ``correction_attempts`` table for tracking decision correction +workflows (revert/append modes) as defined in the specification DDL. +Each row records one correction attempt against a plan's decision tree. + +Revision ID: m8_001_correction_attempts +Revises: m4_003_plan_env_columns +Create Date: 2026-03-24 00:00:00 + +""" + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +# revision identifiers, used by Alembic. +revision: str = "m8_001_correction_attempts" +down_revision: str | Sequence[str] | None = "m4_003_plan_env_columns" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + """Create correction_attempts table with spec-aligned schema.""" + + op.create_table( + "correction_attempts", + sa.Column("correction_attempt_id", sa.String(26), primary_key=True), + sa.Column( + "plan_id", + sa.String(26), + sa.ForeignKey("v3_plans.plan_id", ondelete="CASCADE"), + nullable=False, + ), + sa.Column( + "original_decision_id", + sa.String(26), + sa.ForeignKey("decisions.decision_id", ondelete="RESTRICT"), + nullable=False, + ), + sa.Column( + "new_decision_id", + sa.String(26), + sa.ForeignKey("decisions.decision_id", ondelete="RESTRICT"), + nullable=True, + ), + sa.Column("mode", sa.String(20), nullable=False), + sa.Column("guidance", sa.Text(), nullable=False), + sa.Column("archived_artifacts_path", sa.Text(), nullable=True), + sa.Column( + "state", + sa.String(20), + nullable=False, + server_default="pending", + ), + sa.Column( + "created_at", + sa.String(30), + nullable=False, + server_default=sa.text("(strftime('%Y-%m-%dT%H:%M:%f', 'now'))"), + ), + sa.Column("completed_at", sa.String(30), nullable=True), + sa.CheckConstraint( + "mode IN ('revert', 'append')", + name="ck_correction_attempts_mode", + ), + sa.CheckConstraint( + "state IN ('pending', 'executing', 'complete', 'failed')", + name="ck_correction_attempts_state", + ), + ) + + op.create_index( + "idx_corrections_plan", + "correction_attempts", + ["plan_id"], + ) + + +def downgrade() -> None: + """Drop correction_attempts table.""" + + op.drop_index("idx_corrections_plan", table_name="correction_attempts") + op.drop_table("correction_attempts") diff --git a/features/correction_attempt_persistence.feature b/features/correction_attempt_persistence.feature new file mode 100644 index 000000000..97a3b2682 --- /dev/null +++ b/features/correction_attempt_persistence.feature @@ -0,0 +1,401 @@ +Feature: Correction attempt persistence via CorrectionAttemptRepository + As a developer + I want correction attempt records to persist correctly + So that decision correction workflows are durable across operations + + Background: + Given a fresh in-memory correction attempt database + And a prerequisite action "local/correction-action" exists for correction attempts + And a prerequisite plan exists for correction attempts + And a prerequisite root decision exists for correction attempts + + # ------------------------------------------------------------------ + # Create and retrieve + # ------------------------------------------------------------------ + + Scenario: Create a correction attempt and retrieve it by ID + Given a new correction attempt in revert mode + When I persist the correction attempt via the repository + Then I can retrieve the correction attempt by its ID + And the persisted correction attempt mode should be "revert" + And the persisted correction attempt state should be "pending" + + Scenario: Create an append-mode correction attempt + Given a new correction attempt in append mode + When I persist the correction attempt via the repository + Then I can retrieve the correction attempt by its ID + And the persisted correction attempt mode should be "append" + + Scenario: Persisted correction attempt preserves all fields + Given a new correction attempt with all fields populated + When I persist the correction attempt via the repository + Then I can retrieve the correction attempt by its ID + And the persisted correction attempt guidance should match + And the persisted correction attempt plan_id should match + And the persisted correction attempt original_decision_id should match + + # ------------------------------------------------------------------ + # List by plan + # ------------------------------------------------------------------ + + Scenario: List correction attempts by plan returns ordered results + Given 3 persisted correction attempts for the same plan + When I list correction attempts by plan ID + Then I should get 3 correction attempts in creation order + + # ------------------------------------------------------------------ + # Update state + # ------------------------------------------------------------------ + + Scenario: Update correction attempt state to executing + Given a persisted correction attempt in pending state + When I update the correction attempt state to "executing" + Then the correction attempt state should be "executing" + + Scenario: Update correction attempt state to complete with timestamp + Given a persisted correction attempt in executing state + When I update the correction attempt to complete with timestamp + Then the correction attempt state should be "complete" + And the correction attempt completed_at should be set + + Scenario: Invalid state transition from pending to complete is rejected + Given a persisted correction attempt in pending state + When I try to update the correction attempt state to "complete" + Then an InvalidCorrectionStateTransitionError should be raised + + Scenario: Update correction attempt with new decision ID + Given a persisted correction attempt in pending state + When I update the correction attempt with a new decision ID + Then the correction attempt new_decision_id should be set + + Scenario: Update correction attempt with archived artifacts path + Given a persisted correction attempt in pending state + When I update the correction attempt with archived artifacts path + Then the correction attempt archived_artifacts_path should be set + + # ------------------------------------------------------------------ + # Delete + # ------------------------------------------------------------------ + + Scenario: Delete a correction attempt by ID + Given a persisted correction attempt in pending state + When I delete the correction attempt + Then the correction attempt should no longer exist + + Scenario: Delete a non-existent correction attempt returns false + When I try to delete a non-existent correction attempt + Then the correction attempt delete result should be false + + # ------------------------------------------------------------------ + # Duplicate detection + # ------------------------------------------------------------------ + + Scenario: Creating a duplicate correction attempt raises error + Given a persisted correction attempt in pending state + When I try to persist the same correction attempt again + Then a DuplicateCorrectionAttemptError should be raised + + # ------------------------------------------------------------------ + # Cascade deletion + # ------------------------------------------------------------------ + + Scenario: Deleting a plan cascades to its correction attempts + Given a persisted correction attempt in pending state + When the parent plan is deleted + Then the correction attempt should have been cascade deleted + + # ------------------------------------------------------------------ + # Not found + # ------------------------------------------------------------------ + + Scenario: Getting a non-existent correction attempt raises error + When I try to get a non-existent correction attempt + Then a CorrectionAttemptNotFoundError should be raised + + # ------------------------------------------------------------------ + # Edge cases + # ------------------------------------------------------------------ + + Scenario: List correction attempts for a plan with no attempts + When I list correction attempts for a plan with no correction attempts + Then I should get 0 correction attempts + + # ------------------------------------------------------------------ + # Additional state transition coverage + # ------------------------------------------------------------------ + + Scenario: Update correction attempt state to failed with timestamp + Given a persisted correction attempt in executing state + When I update the correction attempt to failed with timestamp + Then the correction attempt state should be "failed" + And the correction attempt completed_at should be set + + Scenario: Transition from complete state is rejected + Given a persisted correction attempt in complete state + When I try to update the correction attempt state to "executing" + Then an InvalidCorrectionStateTransitionError should be raised + + Scenario: Transition from failed state is rejected + Given a persisted correction attempt in failed state + When I try to update the correction attempt state to "pending" + Then an InvalidCorrectionStateTransitionError should be raised + + Scenario: Setting completed_at on non-terminal transition is rejected + Given a persisted correction attempt in pending state + When I try to update the correction attempt to executing with completed_at + Then an InvalidCorrectionStateTransitionError should be raised + + # ------------------------------------------------------------------ + # Domain model guidance validation + # ------------------------------------------------------------------ + + Scenario: Empty guidance is rejected at domain model level + When I try to create a correction attempt with empty guidance + Then a guidance validation error should be raised + + Scenario: Whitespace-only guidance is rejected at domain model level + When I try to create a correction attempt with whitespace-only guidance + Then a guidance validation error should be raised + + # ------------------------------------------------------------------ + # Additional test coverage + # ------------------------------------------------------------------ + + Scenario: Updating state of a non-existent correction attempt raises error + When I try to update the state of a non-existent correction attempt + Then a CorrectionAttemptNotFoundError should be raised + + Scenario: Updating with non-existent new_decision_id raises FK error + Given a persisted correction attempt in pending state + When I try to update the correction attempt with a non-existent decision ID + Then a DatabaseError should be raised + + Scenario: Guidance at max length is accepted + Given a new correction attempt with guidance at max length + When I persist the correction attempt via the repository + Then I can retrieve the correction attempt by its ID + + Scenario: Guidance exceeding max length is rejected at domain model level + When I try to create a correction attempt with guidance exceeding max length + Then a guidance validation error should be raised + + # ------------------------------------------------------------------ + # Timezone normalization coverage (M3) + # ------------------------------------------------------------------ + + Scenario: Non-UTC timezone datetimes are normalised to UTC on persist + Given a new correction attempt with non-UTC timezone timestamps + When I persist the correction attempt via the repository + Then I can retrieve the correction attempt by its ID + And the persisted correction attempt created_at should be in UTC + + # ------------------------------------------------------------------ + # archived_artifacts_path round-trip on create (M4) + # ------------------------------------------------------------------ + + Scenario: Correction attempt with archived_artifacts_path round-trips on create + Given a new correction attempt with archived_artifacts_path set + When I persist the correction attempt via the repository + Then I can retrieve the correction attempt by its ID + And the persisted correction attempt archived_artifacts_path should match + + # ------------------------------------------------------------------ + # Minimum-boundary guidance (L7) + # ------------------------------------------------------------------ + + Scenario: Single-character guidance is accepted + Given a new correction attempt with single-character guidance + When I persist the correction attempt via the repository + Then I can retrieve the correction attempt by its ID + And the persisted correction attempt guidance should be a single character + + # ------------------------------------------------------------------ + # Delete in non-pending state (L4) + # ------------------------------------------------------------------ + + Scenario: Delete a correction attempt in complete state + Given a persisted correction attempt in complete state + When I delete the correction attempt + Then the correction attempt should no longer exist + + # ------------------------------------------------------------------ + # Cross-plan isolation for list_by_plan (L5) + # ------------------------------------------------------------------ + + Scenario: list_by_plan returns only attempts for the specified plan + Given a persisted correction attempt in pending state + When I list correction attempts for a plan with no correction attempts + Then I should get 0 correction attempts + + # ------------------------------------------------------------------ + # Auto-set completed_at for terminal states (T-1) + # ------------------------------------------------------------------ + + Scenario: Terminal transition auto-sets completed_at when not provided + Given a persisted correction attempt in executing state + When I update the correction attempt state to "complete" without completed_at + Then the correction attempt state should be "complete" + And the correction attempt completed_at should be set + + Scenario: Terminal transition to failed auto-sets completed_at when not provided + Given a persisted correction attempt in executing state + When I update the correction attempt state to "failed" without completed_at + Then the correction attempt state should be "failed" + And the correction attempt completed_at should be set + + # ------------------------------------------------------------------ + # FK violation on create (T-2) + # ------------------------------------------------------------------ + + Scenario: Creating a correction attempt with non-existent plan_id raises error + When I try to create a correction attempt with a non-existent plan_id + Then a DatabaseError should be raised + + Scenario: Creating a correction attempt with non-existent original_decision_id raises error + When I try to create a correction attempt with a non-existent original_decision_id + Then a DatabaseError should be raised + + # ------------------------------------------------------------------ + # Invalid mode at domain model level (T-3) + # ------------------------------------------------------------------ + + Scenario: Invalid mode value is rejected at domain model level + When I try to create a correction attempt with an invalid mode + Then a mode validation error should be raised + + # ------------------------------------------------------------------ + # Empty/whitespace new_decision_id via update_state (M-1) + # ------------------------------------------------------------------ + + Scenario: Whitespace-only new_decision_id is rejected in update_state + Given a persisted correction attempt in pending state + When I try to update the correction attempt with whitespace-only new_decision_id + Then a correction attempt ValueError should be raised + + Scenario: Empty new_decision_id is rejected in update_state + Given a persisted correction attempt in pending state + When I try to update the correction attempt with empty new_decision_id + Then a correction attempt ValueError should be raised + + # ------------------------------------------------------------------ + # Combined field update (L-5) + # ------------------------------------------------------------------ + + Scenario: Update with both new_decision_id and archived_artifacts_path + Given a persisted correction attempt in pending state + When I update the correction attempt with new_decision_id and archived_artifacts_path + Then the correction attempt new_decision_id should be set + And the correction attempt archived_artifacts_path should be set + + # ------------------------------------------------------------------ + # Self-transition rejection (L-3) + # ------------------------------------------------------------------ + + Scenario: Self-transition from pending to pending is rejected + Given a persisted correction attempt in pending state + When I try to update the correction attempt state to "pending" + Then an InvalidCorrectionStateTransitionError should be raised + + Scenario: Self-transition from executing to executing is rejected + Given a persisted correction attempt in executing state + When I try to update the correction attempt state to "executing" + Then an InvalidCorrectionStateTransitionError should be raised + + # ------------------------------------------------------------------ + # Empty/whitespace archived_artifacts_path via update_state (L-2) + # ------------------------------------------------------------------ + + Scenario: Empty archived_artifacts_path is rejected in update_state + Given a persisted correction attempt in pending state + When I try to update the correction attempt with empty archived_artifacts_path + Then a correction attempt ValueError should be raised + + Scenario: Whitespace-only archived_artifacts_path is rejected in update_state + Given a persisted correction attempt in pending state + When I try to update the correction attempt with whitespace-only archived_artifacts_path + Then a correction attempt ValueError should be raised + + # ------------------------------------------------------------------ + # Whitespace-padded archived_artifacts_path stripping (L-6) + # ------------------------------------------------------------------ + + Scenario: Whitespace-padded archived_artifacts_path is stripped on update + Given a persisted correction attempt in pending state + When I update the correction attempt with whitespace-padded archived_artifacts_path + Then the correction attempt archived_artifacts_path should be stripped + + # ------------------------------------------------------------------ + # RESTRICT FK on original_decision_id (R-1) + # ------------------------------------------------------------------ + + Scenario: Deleting a decision referenced by original_decision_id is blocked + Given a persisted correction attempt in pending state + When I try to delete the original decision + Then a database integrity error should be raised for the decision + And the correction attempt should still exist + + # ------------------------------------------------------------------ + # Stronger cross-plan isolation (I-1) + # ------------------------------------------------------------------ + + Scenario: list_by_plan returns only first plan's attempts + Given a second prerequisite plan exists for correction attempts + And a persisted correction attempt for the first plan + And a persisted correction attempt for the second plan + When I list correction attempts for the first plan + Then I should get exactly 1 correction attempt for the first plan + + Scenario: list_by_plan returns only second plan's attempts + Given a second prerequisite plan exists for correction attempts + And a persisted correction attempt for the first plan + And a persisted correction attempt for the second plan + When I list correction attempts for the second plan + Then I should get exactly 1 correction attempt for the second plan + + # ------------------------------------------------------------------ + # Defensive to_domain() coercion for corrupted DB data (DC-1) + # ------------------------------------------------------------------ + + Scenario: to_domain coerces unknown mode to revert + Given a raw correction attempt row with corrupted mode "unknown_mode" + When I retrieve the corrupted correction attempt by ID + Then the retrieved correction attempt mode should be "revert" + + Scenario: to_domain coerces unknown state to failed + Given a raw correction attempt row with corrupted state "unknown_state" + When I retrieve the corrupted correction attempt by ID + Then the retrieved correction attempt state should be "failed" + + Scenario: to_domain coerces empty guidance to corrupted marker + Given a raw correction attempt row with empty guidance + When I retrieve the corrupted correction attempt by ID + Then the retrieved correction attempt guidance should be "[corrupted]" + + # ------------------------------------------------------------------ + # format_sqlite_timestamp naive datetime rejection (TS-1) + # ------------------------------------------------------------------ + + Scenario: format_sqlite_timestamp rejects naive datetimes + When I call format_sqlite_timestamp with a naive datetime + Then a ValueError should be raised for naive datetime + + # ------------------------------------------------------------------ + # Domain model naive datetime normalisation (TS-2) + # ------------------------------------------------------------------ + + Scenario: CorrectionAttemptRecord normalizes naive created_at to UTC + When I create a CorrectionAttemptRecord with naive created_at + Then the record created_at should have UTC timezone + + Scenario: CorrectionAttemptRecord normalizes naive completed_at to UTC + When I create a CorrectionAttemptRecord with naive completed_at + Then the record completed_at should have UTC timezone + + # ------------------------------------------------------------------ + # Corrupted DB state in update_state (DC-2) + # ------------------------------------------------------------------ + + Scenario: update_state treats corrupted DB state as terminal failed + Given a raw correction attempt row with corrupted state "garbage_state" + When I try to update the corrupted correction attempt state to "executing" + Then an InvalidCorrectionStateTransitionError should be raised diff --git a/features/steps/correction_attempt_persistence_steps.py b/features/steps/correction_attempt_persistence_steps.py new file mode 100644 index 000000000..c7d2051ac --- /dev/null +++ b/features/steps/correction_attempt_persistence_steps.py @@ -0,0 +1,1263 @@ +"""Step definitions for correction_attempt_persistence.feature. + +Tests the CorrectionAttemptRepository CRUD operations, state updates, +list queries, and constraint enforcement. +""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta +from typing import Any + +from behave import given, then, when # type: ignore[import-untyped] +from behave.runner import Context +from pydantic import ValidationError +from sqlalchemy import create_engine, event +from sqlalchemy.exc import IntegrityError as SAIntegrityError +from sqlalchemy.orm import sessionmaker +from ulid import ULID + +from cleveragents.core.exceptions import DatabaseError +from cleveragents.domain.models.core.action import Action, ActionState +from cleveragents.domain.models.core.correction import ( + CorrectionAttemptRecord, + CorrectionAttemptState, + CorrectionMode, +) +from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + Decision, + DecisionType, +) +from cleveragents.domain.models.core.plan import ( + NamespacedName, + Plan, + PlanIdentity, + PlanPhase, + PlanTimestamps, + ProcessingState, +) +from cleveragents.infrastructure.database.models import Base +from cleveragents.infrastructure.database.repositories import ( + ActionRepository, + CorrectionAttemptNotFoundError, + CorrectionAttemptRepository, + DecisionRepository, + DuplicateCorrectionAttemptError, + InvalidCorrectionStateTransitionError, + LifecyclePlanRepository, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_PLAN_ID = "01HV000000000000000000CA01" +_DECISION_ID = "01HV000000000000000000CD01" + + +def _setup_db(context: Context) -> None: + """Create an in-memory SQLite DB and attach repos.""" + engine = create_engine("sqlite:///:memory:", echo=False) + + @event.listens_for(engine, "connect") + def _enable_fk(dbapi_conn: Any, _rec: Any) -> None: + cursor = dbapi_conn.cursor() + cursor.execute("PRAGMA foreign_keys=ON") + cursor.close() + + Base.metadata.create_all(engine) + sm = sessionmaker(bind=engine) + session = sm() + context._ca_engine = engine + context._ca_session = session + context._ca_factory = lambda: session + context._ca_repo = CorrectionAttemptRepository( + session_factory=context._ca_factory, + ) + context._ca_action_repo = ActionRepository( + session_factory=context._ca_factory, + ) + context._ca_plan_repo = LifecyclePlanRepository( + session_factory=context._ca_factory, + ) + context._ca_decision_repo = DecisionRepository( + session_factory=context._ca_factory, + ) + + +def _make_correction_attempt( + plan_id: str = _PLAN_ID, + original_decision_id: str = _DECISION_ID, + mode: CorrectionMode = CorrectionMode.REVERT, + guidance: str = "Fix the broken implementation", + state: CorrectionAttemptState = CorrectionAttemptState.PENDING, + new_decision_id: str | None = None, + archived_artifacts_path: str | None = None, + completed_at: datetime | None = None, +) -> CorrectionAttemptRecord: + return CorrectionAttemptRecord( + plan_id=plan_id, + original_decision_id=original_decision_id, + mode=mode, + guidance=guidance, + state=state, + new_decision_id=new_decision_id, + archived_artifacts_path=archived_artifacts_path, + completed_at=completed_at, + ) + + +def _create_prerequisite_action(context: Context) -> None: + ns = NamespacedName.parse("local/correction-action") + action = Action( + namespaced_name=ns, + description="Prerequisite action for correction attempt tests", + definition_of_done="Action is complete", + strategy_actor="strategy-actor", + execution_actor="execution-actor", + state=ActionState.AVAILABLE, + ) + context._ca_action_repo.create(action) + context._ca_session.commit() + + +def _create_prerequisite_plan(context: Context) -> None: + now = datetime.now(UTC) + plan = Plan( + identity=PlanIdentity(plan_id=_PLAN_ID, attempt=1), + namespaced_name=NamespacedName.parse("local/test-plan"), + action_name="local/correction-action", + description="Test plan for correction attempt persistence", + phase=PlanPhase.EXECUTE, + processing_state=ProcessingState.PROCESSING, + timestamps=PlanTimestamps(created_at=now, updated_at=now), + ) + context._ca_plan_repo.create(plan) + context._ca_session.commit() + + +def _create_prerequisite_decision(context: Context) -> None: + decision = Decision( + decision_id=_DECISION_ID, + plan_id=_PLAN_ID, + sequence_number=0, + decision_type=DecisionType.PROMPT_DEFINITION, + question="What approach?", + chosen_option="Build a REST API", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:test", + hot_context_ref="ref:test", + relevant_resources=[], + actor_state_ref="", + ), + ) + context._ca_decision_repo.create(decision) + context._ca_session.commit() + + +# --------------------------------------------------------------------------- +# Background steps +# --------------------------------------------------------------------------- + + +@given("a fresh in-memory correction attempt database") +def step_fresh_db(context: Context) -> None: + _setup_db(context) + + +@given('a prerequisite action "local/correction-action" exists for correction attempts') +def step_prerequisite_action(context: Context) -> None: + _create_prerequisite_action(context) + + +@given("a prerequisite plan exists for correction attempts") +def step_prerequisite_plan(context: Context) -> None: + _create_prerequisite_plan(context) + + +@given("a prerequisite root decision exists for correction attempts") +def step_prerequisite_decision(context: Context) -> None: + _create_prerequisite_decision(context) + + +# --------------------------------------------------------------------------- +# Create scenarios +# --------------------------------------------------------------------------- + + +@given("a new correction attempt in revert mode") +def step_new_revert_attempt(context: Context) -> None: + context._ca_record = _make_correction_attempt(mode=CorrectionMode.REVERT) + + +@given("a new correction attempt in append mode") +def step_new_append_attempt(context: Context) -> None: + context._ca_record = _make_correction_attempt(mode=CorrectionMode.APPEND) + + +@given("a new correction attempt with all fields populated") +def step_new_full_attempt(context: Context) -> None: + context._ca_record = _make_correction_attempt( + guidance="Detailed correction guidance text", + archived_artifacts_path="/tmp/archived/artifacts", + ) + + +@when("I persist the correction attempt via the repository") +def step_persist_attempt(context: Context) -> None: + result = context._ca_repo.create(context._ca_record) + context._ca_session.commit() + context._ca_persisted = result + + +@then("I can retrieve the correction attempt by its ID") +def step_retrieve_by_id(context: Context) -> None: + attempt_id = context._ca_persisted.correction_attempt_id + result = context._ca_repo.get(attempt_id) + context._ca_retrieved = result + assert result is not None + assert result.correction_attempt_id == attempt_id + + +@then('the persisted correction attempt mode should be "{mode}"') +def step_check_mode(context: Context, mode: str) -> None: + assert context._ca_retrieved.mode.value == mode + + +@then('the persisted correction attempt state should be "{state}"') +def step_check_state(context: Context, state: str) -> None: + assert context._ca_retrieved.state.value == state + + +@then("the persisted correction attempt guidance should match") +def step_check_guidance(context: Context) -> None: + assert context._ca_retrieved.guidance == context._ca_record.guidance + + +@then("the persisted correction attempt plan_id should match") +def step_check_plan_id(context: Context) -> None: + assert context._ca_retrieved.plan_id == context._ca_record.plan_id + + +@then("the persisted correction attempt original_decision_id should match") +def step_check_original_decision_id(context: Context) -> None: + assert ( + context._ca_retrieved.original_decision_id + == context._ca_record.original_decision_id + ) + + +# --------------------------------------------------------------------------- +# List by plan +# --------------------------------------------------------------------------- + + +@given("3 persisted correction attempts for the same plan") +def step_three_attempts(context: Context) -> None: + context._ca_attempts = [] + base_time = datetime(2026, 1, 1, 0, 0, 0, tzinfo=UTC) + for i in range(3): + record = _make_correction_attempt( + guidance=f"Correction guidance {i}", + ) + # Use deterministic timestamps to guarantee ordering + record.created_at = base_time + timedelta(seconds=i) + result = context._ca_repo.create(record) + context._ca_session.commit() + context._ca_attempts.append(result) + + +@when("I list correction attempts by plan ID") +def step_list_by_plan(context: Context) -> None: + context._ca_list_result = context._ca_repo.list_by_plan(_PLAN_ID) + + +@then("I should get 3 correction attempts in creation order") +def step_check_list_count(context: Context) -> None: + assert len(context._ca_list_result) == 3 + # Check ordering + for i in range(len(context._ca_list_result) - 1): + assert ( + context._ca_list_result[i].created_at + <= context._ca_list_result[i + 1].created_at + ) + + +# --------------------------------------------------------------------------- +# Update state +# --------------------------------------------------------------------------- + + +@given("a persisted correction attempt in pending state") +def step_persisted_pending(context: Context) -> None: + record = _make_correction_attempt() + result = context._ca_repo.create(record) + context._ca_session.commit() + context._ca_persisted = result + context._ca_record = record + + +@given("a persisted correction attempt in executing state") +def step_persisted_executing(context: Context) -> None: + record = _make_correction_attempt() + result = context._ca_repo.create(record) + context._ca_session.commit() + # Transition to executing first (valid: pending → executing) + result = context._ca_repo.update_state( + result.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + ) + context._ca_session.commit() + context._ca_persisted = result + context._ca_record = record + + +@when('I update the correction attempt state to "{state}"') +def step_update_state(context: Context, state: str) -> None: + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState(state), + ) + context._ca_session.commit() + + +@when('I try to update the correction attempt state to "{state}"') +def step_try_update_state(context: Context, state: str) -> None: + try: + context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState(state), + ) + context._ca_session.commit() + context._ca_invalid_transition_error = None + except InvalidCorrectionStateTransitionError as exc: + context._ca_session.rollback() + context._ca_invalid_transition_error = exc + + +@then('the correction attempt state should be "{state}"') +def step_verify_state(context: Context, state: str) -> None: + refreshed = context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + assert refreshed.state.value == state + + +@when("I update the correction attempt to complete with timestamp") +def step_update_complete_with_timestamp(context: Context) -> None: + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.COMPLETE, + completed_at=datetime.now(UTC), + ) + context._ca_session.commit() + + +@then("the correction attempt completed_at should be set") +def step_check_completed_at(context: Context) -> None: + refreshed = context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + assert refreshed.completed_at is not None + + +@when("I update the correction attempt with a new decision ID") +def step_update_new_decision(context: Context) -> None: + new_dec_id = str(ULID()) + # Create the new decision first + new_decision = Decision( + decision_id=new_dec_id, + plan_id=_PLAN_ID, + sequence_number=1, + decision_type=DecisionType.STRATEGY_CHOICE, + question="New approach?", + chosen_option="Use a new strategy", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:new", + hot_context_ref="ref:new", + relevant_resources=[], + actor_state_ref="", + ), + ) + context._ca_decision_repo.create(new_decision) + context._ca_session.commit() + + context._ca_new_decision_id = new_dec_id + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + new_decision_id=new_dec_id, + ) + context._ca_session.commit() + + +@then("the correction attempt new_decision_id should be set") +def step_check_new_decision_id(context: Context) -> None: + refreshed = context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + assert refreshed.new_decision_id == context._ca_new_decision_id + + +@when("I update the correction attempt with archived artifacts path") +def step_update_archived_path(context: Context) -> None: + context._ca_archived_path = "/tmp/archived/correction" + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + archived_artifacts_path=context._ca_archived_path, + ) + context._ca_session.commit() + + +@then("the correction attempt archived_artifacts_path should be set") +def step_check_archived_path(context: Context) -> None: + refreshed = context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + assert refreshed.archived_artifacts_path == context._ca_archived_path + + +# --------------------------------------------------------------------------- +# Delete +# --------------------------------------------------------------------------- + + +@when("I delete the correction attempt") +def step_delete_attempt(context: Context) -> None: + context._ca_delete_result = context._ca_repo.delete( + context._ca_persisted.correction_attempt_id, + ) + context._ca_session.commit() + + +@then("the correction attempt should no longer exist") +def step_check_deleted(context: Context) -> None: + assert context._ca_delete_result is True + try: + context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + msg = "Should have raised CorrectionAttemptNotFoundError" + raise AssertionError(msg) + except CorrectionAttemptNotFoundError: + pass + + +@when("I try to delete a non-existent correction attempt") +def step_delete_nonexistent(context: Context) -> None: + context._ca_delete_result = context._ca_repo.delete( + "01HV000000000000000NONEXIST", + ) + + +@then("the correction attempt delete result should be false") +def step_check_delete_false(context: Context) -> None: + assert context._ca_delete_result is False + + +# --------------------------------------------------------------------------- +# Cascade deletion +# --------------------------------------------------------------------------- + + +@when("the parent plan is deleted") +def step_delete_parent_plan(context: Context) -> None: + context._ca_plan_repo.delete(_PLAN_ID) + context._ca_session.commit() + + +@then("the correction attempt should have been cascade deleted") +def step_check_cascade_deleted(context: Context) -> None: + try: + context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + msg = "Should have raised CorrectionAttemptNotFoundError after cascade delete" + raise AssertionError(msg) + except CorrectionAttemptNotFoundError: + pass + + +# --------------------------------------------------------------------------- +# Duplicate detection +# --------------------------------------------------------------------------- + + +@when("I try to persist the same correction attempt again") +def step_persist_duplicate(context: Context) -> None: + try: + context._ca_repo.create(context._ca_record) + context._ca_session.commit() + context._ca_duplicate_error = None + except DuplicateCorrectionAttemptError as exc: + context._ca_session.rollback() + context._ca_duplicate_error = exc + + +@then("a DuplicateCorrectionAttemptError should be raised") +def step_check_duplicate_error(context: Context) -> None: + assert context._ca_duplicate_error is not None + assert isinstance( + context._ca_duplicate_error, + DuplicateCorrectionAttemptError, + ) + + +# --------------------------------------------------------------------------- +# Not found +# --------------------------------------------------------------------------- + + +@when("I try to get a non-existent correction attempt") +def step_get_nonexistent(context: Context) -> None: + try: + context._ca_repo.get("01HV000000000000000NONEXIST") + context._ca_not_found_error = None + except CorrectionAttemptNotFoundError as exc: + context._ca_not_found_error = exc + + +@then("a CorrectionAttemptNotFoundError should be raised") +def step_check_not_found_error(context: Context) -> None: + assert context._ca_not_found_error is not None + assert isinstance( + context._ca_not_found_error, + CorrectionAttemptNotFoundError, + ) + + +# --------------------------------------------------------------------------- +# Invalid state transitions +# --------------------------------------------------------------------------- + + +@then("an InvalidCorrectionStateTransitionError should be raised") +def step_check_invalid_transition_error(context: Context) -> None: + assert context._ca_invalid_transition_error is not None + assert isinstance( + context._ca_invalid_transition_error, + InvalidCorrectionStateTransitionError, + ) + + +# --------------------------------------------------------------------------- +# Edge cases +# --------------------------------------------------------------------------- + + +@when("I list correction attempts for a plan with no correction attempts") +def step_list_empty(context: Context) -> None: + context._ca_list_result = context._ca_repo.list_by_plan( + "01HV000000000000000NOATTEMPT", + ) + + +@then("I should get 0 correction attempts") +def step_check_empty_list(context: Context) -> None: + assert len(context._ca_list_result) == 0 + + +# --------------------------------------------------------------------------- +# Additional state transition coverage +# --------------------------------------------------------------------------- + + +@when("I update the correction attempt to failed with timestamp") +def step_update_failed_with_timestamp(context: Context) -> None: + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.FAILED, + completed_at=datetime.now(UTC), + ) + context._ca_session.commit() + + +@given("a persisted correction attempt in complete state") +def step_persisted_complete(context: Context) -> None: + record = _make_correction_attempt() + result = context._ca_repo.create(record) + context._ca_session.commit() + # pending → executing + result = context._ca_repo.update_state( + result.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + ) + context._ca_session.commit() + # executing → complete + result = context._ca_repo.update_state( + result.correction_attempt_id, + state=CorrectionAttemptState.COMPLETE, + completed_at=datetime.now(UTC), + ) + context._ca_session.commit() + context._ca_persisted = result + context._ca_record = record + + +@given("a persisted correction attempt in failed state") +def step_persisted_failed(context: Context) -> None: + record = _make_correction_attempt() + result = context._ca_repo.create(record) + context._ca_session.commit() + # pending → executing + result = context._ca_repo.update_state( + result.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + ) + context._ca_session.commit() + # executing → failed + result = context._ca_repo.update_state( + result.correction_attempt_id, + state=CorrectionAttemptState.FAILED, + completed_at=datetime.now(UTC), + ) + context._ca_session.commit() + context._ca_persisted = result + context._ca_record = record + + +@when("I try to update the correction attempt to executing with completed_at") +def step_try_update_executing_with_completed_at(context: Context) -> None: + try: + context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + completed_at=datetime.now(UTC), + ) + context._ca_session.commit() + context._ca_invalid_transition_error = None + except InvalidCorrectionStateTransitionError as exc: + context._ca_session.rollback() + context._ca_invalid_transition_error = exc + + +# --------------------------------------------------------------------------- +# Domain model guidance validation +# --------------------------------------------------------------------------- + + +@when("I try to create a correction attempt with empty guidance") +def step_create_empty_guidance(context: Context) -> None: + try: + _make_correction_attempt(guidance="") + context._ca_guidance_error = None + except ValidationError as exc: + context._ca_guidance_error = exc + + +@when("I try to create a correction attempt with whitespace-only guidance") +def step_create_whitespace_guidance(context: Context) -> None: + try: + _make_correction_attempt(guidance=" \t\n ") + context._ca_guidance_error = None + except ValidationError as exc: + context._ca_guidance_error = exc + + +@then("a guidance validation error should be raised") +def step_check_guidance_error(context: Context) -> None: + assert context._ca_guidance_error is not None + assert isinstance(context._ca_guidance_error, ValidationError) + + +# --------------------------------------------------------------------------- +# Additional test coverage +# --------------------------------------------------------------------------- + + +@when("I try to update the state of a non-existent correction attempt") +def step_try_update_nonexistent(context: Context) -> None: + try: + context._ca_repo.update_state( + "01HV000000000000000NONEXIST", + state=CorrectionAttemptState.EXECUTING, + ) + context._ca_not_found_error = None + except CorrectionAttemptNotFoundError as exc: + context._ca_not_found_error = exc + + +@when("I try to update the correction attempt with a non-existent decision ID") +def step_try_update_bad_fk(context: Context) -> None: + try: + context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + new_decision_id="01HV000000000000000BADFK01", + ) + context._ca_session.commit() + context._ca_db_error = None + except DatabaseError as exc: + context._ca_session.rollback() + context._ca_db_error = exc + + +@then("a DatabaseError should be raised") +def step_check_db_error(context: Context) -> None: + assert context._ca_db_error is not None + assert isinstance(context._ca_db_error, DatabaseError) + + +@given("a new correction attempt with guidance at max length") +def step_new_max_guidance(context: Context) -> None: + context._ca_record = _make_correction_attempt( + guidance="x" * 10_000, + ) + + +@when("I try to create a correction attempt with guidance exceeding max length") +def step_create_over_max_guidance(context: Context) -> None: + try: + _make_correction_attempt(guidance="x" * 10_001) + context._ca_guidance_error = None + except ValidationError as exc: + context._ca_guidance_error = exc + + +# --------------------------------------------------------------------------- +# Timezone normalization coverage (M3) +# --------------------------------------------------------------------------- + + +@given("a new correction attempt with non-UTC timezone timestamps") +def step_new_non_utc_attempt(context: Context) -> None: + from datetime import timezone as _tz + + eastern = _tz(timedelta(hours=-5)) + context._ca_record = _make_correction_attempt( + guidance="Timezone normalization test", + ) + # Override created_at with a non-UTC timezone-aware datetime + context._ca_record.created_at = datetime(2026, 6, 15, 12, 0, 0, tzinfo=eastern) + + +@then("the persisted correction attempt created_at should be in UTC") +def step_check_created_at_utc(context: Context) -> None: + retrieved = context._ca_retrieved + assert retrieved.created_at.tzinfo is not None + assert retrieved.created_at.utcoffset() == timedelta(0) + # Original was 12:00 EST (-5h) -> should be 17:00 UTC + assert retrieved.created_at.hour == 17 + + +# --------------------------------------------------------------------------- +# archived_artifacts_path round-trip on create (M4) +# --------------------------------------------------------------------------- + + +@given("a new correction attempt with archived_artifacts_path set") +def step_new_attempt_with_artifacts_path(context: Context) -> None: + context._ca_record = _make_correction_attempt( + guidance="Artifacts path round-trip test", + archived_artifacts_path="/archive/correction/2026-06-15", + ) + + +@then("the persisted correction attempt archived_artifacts_path should match") +def step_check_archived_path_match(context: Context) -> None: + assert ( + context._ca_retrieved.archived_artifacts_path + == context._ca_record.archived_artifacts_path + ) + + +# --------------------------------------------------------------------------- +# Minimum-boundary guidance (L7) +# --------------------------------------------------------------------------- + + +@given("a new correction attempt with single-character guidance") +def step_new_single_char_guidance(context: Context) -> None: + context._ca_record = _make_correction_attempt(guidance="X") + + +@then("the persisted correction attempt guidance should be a single character") +def step_check_single_char_guidance(context: Context) -> None: + assert context._ca_retrieved.guidance == "X" + assert len(context._ca_retrieved.guidance) == 1 + + +# --------------------------------------------------------------------------- +# Auto-set completed_at for terminal states (T-1) +# --------------------------------------------------------------------------- + + +@when('I update the correction attempt state to "complete" without completed_at') +def step_update_complete_no_timestamp(context: Context) -> None: + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.COMPLETE, + ) + context._ca_session.commit() + + +@when('I update the correction attempt state to "failed" without completed_at') +def step_update_failed_no_timestamp(context: Context) -> None: + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.FAILED, + ) + context._ca_session.commit() + + +# --------------------------------------------------------------------------- +# FK violation on create (T-2) +# --------------------------------------------------------------------------- + + +@when("I try to create a correction attempt with a non-existent plan_id") +def step_create_bad_plan_fk(context: Context) -> None: + try: + record = _make_correction_attempt( + plan_id="01HV000000000000000BADPLAN", + ) + context._ca_repo.create(record) + context._ca_session.commit() + context._ca_db_error = None + except DatabaseError as exc: + context._ca_session.rollback() + context._ca_db_error = exc + + +@when("I try to create a correction attempt with a non-existent original_decision_id") +def step_create_bad_decision_fk(context: Context) -> None: + try: + record = _make_correction_attempt( + original_decision_id="01HV000000000000000BADDEC1", + ) + context._ca_repo.create(record) + context._ca_session.commit() + context._ca_db_error = None + except DatabaseError as exc: + context._ca_session.rollback() + context._ca_db_error = exc + + +# --------------------------------------------------------------------------- +# Invalid mode at domain model level (T-3) +# --------------------------------------------------------------------------- + + +@when("I try to create a correction attempt with an invalid mode") +def step_create_invalid_mode(context: Context) -> None: + try: + CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode="invalid_mode", # type: ignore[arg-type] + guidance="Some guidance", + ) + context._ca_mode_error = None + except ValidationError as exc: + context._ca_mode_error = exc + + +@then("a mode validation error should be raised") +def step_check_mode_error(context: Context) -> None: + assert context._ca_mode_error is not None + assert isinstance(context._ca_mode_error, ValidationError) + # Verify the error pertains to the mode field specifically + error_fields = {str(e["loc"]) for e in context._ca_mode_error.errors()} + assert any("mode" in field for field in error_fields) + + +# --------------------------------------------------------------------------- +# Empty/whitespace new_decision_id via update_state (M-1) +# --------------------------------------------------------------------------- + + +@when("I try to update the correction attempt with whitespace-only new_decision_id") +def step_try_update_whitespace_decision_id(context: Context) -> None: + try: + context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + new_decision_id=" ", + ) + context._ca_session.commit() + context._ca_value_error = None + except ValueError as exc: + context._ca_value_error = exc + + +@when("I try to update the correction attempt with empty new_decision_id") +def step_try_update_empty_decision_id(context: Context) -> None: + try: + context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + new_decision_id="", + ) + context._ca_session.commit() + context._ca_value_error = None + except ValueError as exc: + context._ca_value_error = exc + + +# --------------------------------------------------------------------------- +# Combined field update (L-5) +# --------------------------------------------------------------------------- + + +@when( + "I update the correction attempt with new_decision_id and archived_artifacts_path" +) +def step_update_combined_fields(context: Context) -> None: + new_dec_id = str(ULID()) + # Create the new decision first + new_decision = Decision( + decision_id=new_dec_id, + plan_id=_PLAN_ID, + sequence_number=2, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Combined update approach?", + chosen_option="Use both fields", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:combined", + hot_context_ref="ref:combined", + relevant_resources=[], + actor_state_ref="", + ), + ) + context._ca_decision_repo.create(new_decision) + context._ca_session.commit() + + context._ca_new_decision_id = new_dec_id + context._ca_archived_path = "/tmp/archived/correction" + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + new_decision_id=new_dec_id, + archived_artifacts_path=context._ca_archived_path, + ) + context._ca_session.commit() + + +# --------------------------------------------------------------------------- +# Empty/whitespace archived_artifacts_path via update_state (L-2) +# --------------------------------------------------------------------------- + + +@when("I try to update the correction attempt with empty archived_artifacts_path") +def step_try_update_empty_artifacts_path(context: Context) -> None: + try: + context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + archived_artifacts_path="", + ) + context._ca_session.commit() + context._ca_value_error = None + except ValueError as exc: + context._ca_value_error = exc + + +@when( + "I try to update the correction attempt with whitespace-only archived_artifacts_path" +) +def step_try_update_whitespace_artifacts_path(context: Context) -> None: + try: + context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + archived_artifacts_path=" ", + ) + context._ca_session.commit() + context._ca_value_error = None + except ValueError as exc: + context._ca_value_error = exc + + +# --------------------------------------------------------------------------- +# Whitespace-padded archived_artifacts_path stripping (L-6) +# --------------------------------------------------------------------------- + + +@when("I update the correction attempt with whitespace-padded archived_artifacts_path") +def step_update_whitespace_padded_artifacts_path(context: Context) -> None: + context._ca_updated = context._ca_repo.update_state( + context._ca_persisted.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + archived_artifacts_path=" /tmp/archived/padded ", + ) + context._ca_session.commit() + + +@then("the correction attempt archived_artifacts_path should be stripped") +def step_check_stripped_artifacts_path(context: Context) -> None: + refreshed = context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + assert refreshed.archived_artifacts_path == "/tmp/archived/padded" + + +@then("a correction attempt ValueError should be raised") +def step_check_value_error(context: Context) -> None: + assert context._ca_value_error is not None + assert isinstance(context._ca_value_error, ValueError) + + +# --------------------------------------------------------------------------- +# RESTRICT FK on original_decision_id (R-1) +# --------------------------------------------------------------------------- + +_SECOND_PLAN_ID = "01HV000000000000000000CA02" + + +@when("I try to delete the original decision") +def step_try_delete_original_decision(context: Context) -> None: + try: + context._ca_decision_repo.delete(_DECISION_ID) + context._ca_session.commit() + context._ca_integrity_error = None + except Exception as exc: + context._ca_session.rollback() + context._ca_integrity_error = exc + + +@then("a database integrity error should be raised for the decision") +def step_check_integrity_error(context: Context) -> None: + assert context._ca_integrity_error is not None + assert isinstance(context._ca_integrity_error, (SAIntegrityError, DatabaseError)) + + +@then("the correction attempt should still exist") +def step_check_attempt_still_exists(context: Context) -> None: + result = context._ca_repo.get( + context._ca_persisted.correction_attempt_id, + ) + assert result is not None + + +# --------------------------------------------------------------------------- +# Stronger cross-plan isolation (I-1) +# --------------------------------------------------------------------------- + + +@given("a second prerequisite plan exists for correction attempts") +def step_second_prerequisite_plan(context: Context) -> None: + now = datetime.now(UTC) + plan = Plan( + identity=PlanIdentity(plan_id=_SECOND_PLAN_ID, attempt=1), + namespaced_name=NamespacedName.parse("local/test-plan-2"), + action_name="local/correction-action", + description="Second test plan for cross-plan isolation", + phase=PlanPhase.EXECUTE, + processing_state=ProcessingState.PROCESSING, + timestamps=PlanTimestamps(created_at=now, updated_at=now), + ) + context._ca_plan_repo.create(plan) + context._ca_session.commit() + + +@given("a persisted correction attempt for the first plan") +def step_persisted_first_plan(context: Context) -> None: + record = _make_correction_attempt( + plan_id=_PLAN_ID, + guidance="Attempt for first plan", + ) + result = context._ca_repo.create(record) + context._ca_session.commit() + context._ca_first_plan_attempt = result + + +@given("a persisted correction attempt for the second plan") +def step_persisted_second_plan(context: Context) -> None: + record = _make_correction_attempt( + plan_id=_SECOND_PLAN_ID, + guidance="Attempt for second plan", + ) + result = context._ca_repo.create(record) + context._ca_session.commit() + context._ca_second_plan_attempt = result + + +@when("I list correction attempts for the first plan") +def step_list_first_plan(context: Context) -> None: + context._ca_first_list_result = context._ca_repo.list_by_plan(_PLAN_ID) + + +@then("I should get exactly 1 correction attempt for the first plan") +def step_check_first_plan_count(context: Context) -> None: + assert len(context._ca_first_list_result) == 1 + assert context._ca_first_list_result[0].plan_id == _PLAN_ID + + +@when("I list correction attempts for the second plan") +def step_list_second_plan(context: Context) -> None: + context._ca_second_list_result = context._ca_repo.list_by_plan(_SECOND_PLAN_ID) + + +@then("I should get exactly 1 correction attempt for the second plan") +def step_check_second_plan_count(context: Context) -> None: + assert len(context._ca_second_list_result) == 1 + assert context._ca_second_list_result[0].plan_id == _SECOND_PLAN_ID + + +# --------------------------------------------------------------------------- +# Defensive to_domain() coercion for corrupted DB data (DC-1) +# --------------------------------------------------------------------------- + + +def _insert_corrupted_row( + context: Context, + attempt_id: str, + mode: str = "revert", + guidance: str = "test guidance", + state: str = "pending", +) -> None: + """Insert a raw row bypassing ORM and CHECK constraints. + + Temporarily disables SQLite CHECK constraints via PRAGMA so + corrupted values can be inserted for defensive-coercion tests. + """ + raw_conn = context._ca_engine.raw_connection() + cursor = raw_conn.cursor() + cursor.execute("PRAGMA ignore_check_constraints = ON") + cursor.execute( + "INSERT INTO correction_attempts " + "(correction_attempt_id, plan_id, original_decision_id, " + "mode, guidance, state, created_at) " + "VALUES (?, ?, ?, ?, ?, ?, strftime('%Y-%m-%dT%H:%M:%f', 'now'))", + (attempt_id, _PLAN_ID, _DECISION_ID, mode, guidance, state), + ) + cursor.execute("PRAGMA ignore_check_constraints = OFF") + raw_conn.commit() + cursor.close() + raw_conn.close() + + +@given('a raw correction attempt row with corrupted mode "{mode}"') +def step_insert_corrupted_mode(context: Context, mode: str) -> None: + """Insert a raw row with a corrupted mode value.""" + attempt_id = str(ULID()) + context._ca_corrupted_id = attempt_id + _insert_corrupted_row(context, attempt_id, mode=mode) + + +@given('a raw correction attempt row with corrupted state "{state}"') +def step_insert_corrupted_state(context: Context, state: str) -> None: + """Insert a raw row with a corrupted state value.""" + attempt_id = str(ULID()) + context._ca_corrupted_id = attempt_id + _insert_corrupted_row(context, attempt_id, state=state) + + +@given("a raw correction attempt row with empty guidance") +def step_insert_empty_guidance(context: Context) -> None: + """Insert a raw row with empty guidance.""" + attempt_id = str(ULID()) + context._ca_corrupted_id = attempt_id + _insert_corrupted_row(context, attempt_id, guidance="") + + +@when("I retrieve the corrupted correction attempt by ID") +def step_retrieve_corrupted(context: Context) -> None: + # Expire session cache so the ORM re-reads from the database, + # picking up data inserted via the raw connection. + context._ca_session.expire_all() + context._ca_corrupted_record = context._ca_repo.get( + context._ca_corrupted_id, + ) + + +@then('the retrieved correction attempt mode should be "{mode}"') +def step_check_corrupted_mode(context: Context, mode: str) -> None: + assert context._ca_corrupted_record.mode.value == mode + + +@then('the retrieved correction attempt state should be "{state}"') +def step_check_corrupted_state(context: Context, state: str) -> None: + assert context._ca_corrupted_record.state.value == state + + +@then('the retrieved correction attempt guidance should be "{guidance}"') +def step_check_corrupted_guidance(context: Context, guidance: str) -> None: + assert context._ca_corrupted_record.guidance == guidance + + +# --------------------------------------------------------------------------- +# format_sqlite_timestamp naive datetime rejection (TS-1) +# --------------------------------------------------------------------------- + + +@when("I call format_sqlite_timestamp with a naive datetime") +def step_call_format_sqlite_naive(context: Context) -> None: + from cleveragents.infrastructure.database.models import ( + format_sqlite_timestamp, + ) + + try: + format_sqlite_timestamp(datetime(2026, 1, 1, 12, 0, 0)) + context._ca_naive_dt_error = None + except ValueError as exc: + context._ca_naive_dt_error = exc + + +@then("a ValueError should be raised for naive datetime") +def step_check_naive_dt_error(context: Context) -> None: + assert context._ca_naive_dt_error is not None + assert isinstance(context._ca_naive_dt_error, ValueError) + + +# --------------------------------------------------------------------------- +# Domain model naive datetime normalisation (TS-2) +# --------------------------------------------------------------------------- + + +@when("I create a CorrectionAttemptRecord with naive created_at") +def step_create_record_naive_created_at(context: Context) -> None: + context._ca_naive_record = CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode=CorrectionMode.REVERT, + guidance="Naive datetime test", + created_at=datetime(2026, 6, 15, 12, 0, 0), + ) + + +@then("the record created_at should have UTC timezone") +def step_check_record_created_at_utc(context: Context) -> None: + assert context._ca_naive_record.created_at.tzinfo is not None + assert context._ca_naive_record.created_at.utcoffset() == timedelta(0) + + +@when("I create a CorrectionAttemptRecord with naive completed_at") +def step_create_record_naive_completed_at(context: Context) -> None: + context._ca_naive_record = CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode=CorrectionMode.REVERT, + guidance="Naive datetime test", + completed_at=datetime(2026, 6, 15, 12, 0, 0), + ) + + +@then("the record completed_at should have UTC timezone") +def step_check_record_completed_at_utc(context: Context) -> None: + assert context._ca_naive_record.completed_at is not None + assert context._ca_naive_record.completed_at.tzinfo is not None + assert context._ca_naive_record.completed_at.utcoffset() == timedelta(0) + + +# --------------------------------------------------------------------------- +# Corrupted DB state in update_state (DC-2) +# --------------------------------------------------------------------------- + + +@when('I try to update the corrupted correction attempt state to "{state}"') +def step_try_update_corrupted_state(context: Context, state: str) -> None: + # Expire session cache so the ORM re-reads the corrupted row. + context._ca_session.expire_all() + try: + context._ca_repo.update_state( + context._ca_corrupted_id, + state=CorrectionAttemptState(state), + ) + context._ca_session.commit() + context._ca_invalid_transition_error = None + except InvalidCorrectionStateTransitionError as exc: + context._ca_session.rollback() + context._ca_invalid_transition_error = exc diff --git a/robot/correction_attempt_persistence.robot b/robot/correction_attempt_persistence.robot new file mode 100644 index 000000000..5b6058401 --- /dev/null +++ b/robot/correction_attempt_persistence.robot @@ -0,0 +1,44 @@ +*** Settings *** +Documentation Smoke tests for correction attempt persistence via CorrectionAttemptRepository +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_correction_attempt_persistence.py + +*** Test Cases *** +Create And Retrieve Correction Attempt + [Documentation] Create a revert correction attempt, persist, retrieve + [Tags] database correction persistence + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} create-retrieve cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} create-retrieve-ok + +List Correction Attempts By Plan + [Documentation] Create multiple correction attempts and list by plan + [Tags] database correction query persistence + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} list-by-plan cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} list-by-plan-ok + +Update Correction Attempt State + [Documentation] Update state from pending to executing to complete + [Tags] database correction state persistence + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} update-state cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} update-state-ok + +Delete Correction Attempt + [Documentation] Delete a correction attempt and verify it is gone + [Tags] database correction persistence + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} delete cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} delete-ok + +Domain Model Round-Trip + [Documentation] Verify all spec DDL columns survive domain-model round-trip + [Tags] database correction persistence + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} domain-roundtrip cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} domain-roundtrip-ok diff --git a/robot/helper_correction_attempt_persistence.py b/robot/helper_correction_attempt_persistence.py new file mode 100644 index 000000000..9348d425e --- /dev/null +++ b/robot/helper_correction_attempt_persistence.py @@ -0,0 +1,317 @@ +"""Helper script for Robot Framework correction attempt persistence tests. + +Usage: + python robot/helper_correction_attempt_persistence.py + +Subcommands: + create-retrieve Create + retrieve a correction attempt + list-by-plan Create multiple, list by plan ID + update-state Update state transitions + delete Delete a correction attempt + domain-roundtrip Verify all spec DDL columns round-trip +""" + +from __future__ import annotations + +import sys +from datetime import UTC, datetime, timedelta +from pathlib import Path + +# Ensure src is importable when run from workspace root +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + +from collections.abc import Callable +from typing import Any + +from sqlalchemy import create_engine, event +from sqlalchemy.orm import Session, sessionmaker + +from cleveragents.domain.models.core.action import ( + Action, + ActionState, +) +from cleveragents.domain.models.core.correction import ( + CorrectionAttemptRecord, + CorrectionAttemptState, + CorrectionMode, +) +from cleveragents.domain.models.core.decision import ( + ContextSnapshot, + Decision, + DecisionType, +) +from cleveragents.domain.models.core.plan import ( + NamespacedName, + Plan, + PlanIdentity, + PlanPhase, + PlanTimestamps, + ProcessingState, +) +from cleveragents.infrastructure.database.models import Base +from cleveragents.infrastructure.database.repositories import ( + ActionRepository, + CorrectionAttemptNotFoundError, + CorrectionAttemptRepository, + DecisionRepository, + LifecyclePlanRepository, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_PLAN_ID = "01HV000000000000000000RC01" +_DECISION_ID = "01HV000000000000000000RD01" + + +def _setup() -> tuple[Session, Callable[[], Session]]: + """Create an in-memory SQLite DB with all tables and FK enforcement.""" + engine = create_engine("sqlite:///:memory:", echo=False) + + @event.listens_for(engine, "connect") + def _enable_fk(dbapi_conn: Any, _rec: Any) -> None: + cursor = dbapi_conn.cursor() + cursor.execute("PRAGMA foreign_keys=ON") + cursor.close() + + Base.metadata.create_all(engine) + sm = sessionmaker(bind=engine) + session = sm() + factory: Callable[[], Session] = lambda: session # noqa: E731 + return session, factory + + +def _create_prerequisites(session: Session, factory: Callable[[], Session]) -> None: + """Create action + plan + decision needed to satisfy FK constraints.""" + now = datetime.now(UTC) + + action_repo = ActionRepository(session_factory=factory) + action = Action( + namespaced_name=NamespacedName.parse("local/robot-correction-action"), + description="Robot correction test action", + definition_of_done="Correction is applied", + strategy_actor="strategy-actor", + execution_actor="execution-actor", + state=ActionState.AVAILABLE, + ) + action_repo.create(action) + session.commit() + + plan_repo = LifecyclePlanRepository(session_factory=factory) + plan = Plan( + identity=PlanIdentity(plan_id=_PLAN_ID, attempt=1), + namespaced_name=NamespacedName.parse("local/robot-test-plan"), + action_name="local/robot-correction-action", + description="Robot test plan for correction persistence", + phase=PlanPhase.EXECUTE, + processing_state=ProcessingState.PROCESSING, + timestamps=PlanTimestamps(created_at=now, updated_at=now), + ) + plan_repo.create(plan) + session.commit() + + dec_repo = DecisionRepository(session_factory=factory) + decision = Decision( + decision_id=_DECISION_ID, + plan_id=_PLAN_ID, + sequence_number=0, + decision_type=DecisionType.PROMPT_DEFINITION, + question="What approach?", + chosen_option="Build REST API", + context_snapshot=ContextSnapshot( + hot_context_hash="sha256:robot", + hot_context_ref="ref:robot", + relevant_resources=[], + actor_state_ref="", + ), + ) + dec_repo.create(decision) + session.commit() + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def cmd_create_retrieve() -> None: + """Create a correction attempt and retrieve it.""" + session, factory = _setup() + _create_prerequisites(session, factory) + repo = CorrectionAttemptRepository(session_factory=factory) + + record = CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode=CorrectionMode.REVERT, + guidance="Fix the broken implementation", + ) + created = repo.create(record) + session.commit() + + retrieved = repo.get(created.correction_attempt_id) + assert retrieved.correction_attempt_id == created.correction_attempt_id + assert retrieved.mode == CorrectionMode.REVERT + assert retrieved.state == CorrectionAttemptState.PENDING + assert retrieved.plan_id == _PLAN_ID + assert retrieved.original_decision_id == _DECISION_ID + print("create-retrieve-ok") + + +def cmd_list_by_plan() -> None: + """Create multiple correction attempts and list by plan.""" + session, factory = _setup() + _create_prerequisites(session, factory) + repo = CorrectionAttemptRepository(session_factory=factory) + + base_time = datetime(2026, 1, 1, 0, 0, 0, tzinfo=UTC) + for i in range(3): + record = CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode=CorrectionMode.REVERT, + guidance=f"Guidance {i}", + created_at=base_time + timedelta(seconds=i), + ) + repo.create(record) + session.commit() + + results = repo.list_by_plan(_PLAN_ID) + assert len(results) == 3 + # Check ordering + for i in range(len(results) - 1): + assert results[i].created_at <= results[i + 1].created_at + print("list-by-plan-ok") + + +def cmd_update_state() -> None: + """Update state transitions.""" + session, factory = _setup() + _create_prerequisites(session, factory) + repo = CorrectionAttemptRepository(session_factory=factory) + + record = CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode=CorrectionMode.REVERT, + guidance="Fix it", + ) + created = repo.create(record) + session.commit() + + # Update to executing + updated = repo.update_state( + created.correction_attempt_id, + state=CorrectionAttemptState.EXECUTING, + ) + session.commit() + assert updated.state == CorrectionAttemptState.EXECUTING + + # Update to complete with timestamp + updated = repo.update_state( + created.correction_attempt_id, + state=CorrectionAttemptState.COMPLETE, + completed_at=datetime.now(UTC), + ) + session.commit() + assert updated.state == CorrectionAttemptState.COMPLETE + assert updated.completed_at is not None + print("update-state-ok") + + +def cmd_delete() -> None: + """Delete a correction attempt.""" + session, factory = _setup() + _create_prerequisites(session, factory) + repo = CorrectionAttemptRepository(session_factory=factory) + + record = CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode=CorrectionMode.REVERT, + guidance="Delete me", + ) + created = repo.create(record) + session.commit() + + result = repo.delete(created.correction_attempt_id) + session.commit() + assert result is True + + # Verify deleted + try: + repo.get(created.correction_attempt_id) + msg = "Should have raised CorrectionAttemptNotFoundError" + raise AssertionError(msg) + except CorrectionAttemptNotFoundError: + pass + + # Delete non-existent + result = repo.delete("01HV000000000000000NONEXIST") + assert result is False + print("delete-ok") + + +def cmd_domain_roundtrip() -> None: + """Verify all spec DDL columns survive domain-model round-trip.""" + session, factory = _setup() + _create_prerequisites(session, factory) + repo = CorrectionAttemptRepository(session_factory=factory) + + now = datetime.now(UTC) + record = CorrectionAttemptRecord( + plan_id=_PLAN_ID, + original_decision_id=_DECISION_ID, + mode=CorrectionMode.APPEND, + guidance="Detailed guidance text for correction", + archived_artifacts_path="/tmp/archived/artifacts", + state=CorrectionAttemptState.PENDING, + created_at=now, + ) + created = repo.create(record) + session.commit() + + retrieved = repo.get(created.correction_attempt_id) + + # Verify all columns + assert retrieved.correction_attempt_id == created.correction_attempt_id + assert retrieved.plan_id == _PLAN_ID + assert retrieved.original_decision_id == _DECISION_ID + assert retrieved.new_decision_id is None + assert retrieved.mode == CorrectionMode.APPEND + assert retrieved.guidance == "Detailed guidance text for correction" + assert retrieved.archived_artifacts_path == "/tmp/archived/artifacts" + assert retrieved.state == CorrectionAttemptState.PENDING + assert retrieved.created_at is not None + # Verify created_at value survives the round-trip within 10ms + # (from_domain truncates to millisecond precision) + delta = abs((retrieved.created_at - now).total_seconds()) + assert delta < 0.01, f"created_at drift {delta}s exceeds 10ms tolerance" + assert retrieved.completed_at is None + print("domain-roundtrip-ok") + + +# --------------------------------------------------------------------------- +# Dispatch +# --------------------------------------------------------------------------- + +_COMMANDS = { + "create-retrieve": cmd_create_retrieve, + "list-by-plan": cmd_list_by_plan, + "update-state": cmd_update_state, + "delete": cmd_delete, + "domain-roundtrip": cmd_domain_roundtrip, +} + + +def main() -> None: + """Dispatch to the requested subcommand.""" + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>", file=sys.stderr) + sys.exit(1) + _COMMANDS[sys.argv[1]]() + + +if __name__ == "__main__": + main() diff --git a/src/cleveragents/domain/models/core/__init__.py b/src/cleveragents/domain/models/core/__init__.py index 05f36f7df..c00739293 100644 --- a/src/cleveragents/domain/models/core/__init__.py +++ b/src/cleveragents/domain/models/core/__init__.py @@ -76,13 +76,18 @@ from cleveragents.domain.models.core.context_policy import ( enforce_size_budget, ) from cleveragents.domain.models.core.correction import ( + CORRECTION_ATTEMPT_TERMINAL_STATES, + CORRECTION_ATTEMPT_VALID_TRANSITIONS, CorrectionAttempt, + CorrectionAttemptRecord, + CorrectionAttemptState, CorrectionDryRunReport, CorrectionImpact, CorrectionMode, CorrectionRequest, CorrectionResult, CorrectionStatus, + validate_correction_state_transition, ) from cleveragents.domain.models.core.cost_budget import ( BudgetCheckResult, @@ -312,6 +317,8 @@ __all__ = [ "ASYNC_TERMINAL_STATUSES", "BUILTIN_PROFILES", "CATEGORY_DEFAULTS", + "CORRECTION_ATTEMPT_TERMINAL_STATES", + "CORRECTION_ATTEMPT_VALID_TRANSITIONS", "DEFAULT_CIRCUIT_BREAKER", "DEFAULT_DATABASE_RETRY", "DEFAULT_FILE_RETRY", @@ -360,6 +367,8 @@ __all__ = [ "ContextUpdateResult", "ContextView", "CorrectionAttempt", + "CorrectionAttemptRecord", + "CorrectionAttemptState", "CorrectionDryRunReport", "CorrectionImpact", "CorrectionMode", @@ -536,4 +545,5 @@ __all__ = [ "render_dod_template", "resolve_safety_profile", "serialize_job_payload", + "validate_correction_state_transition", ] diff --git a/src/cleveragents/domain/models/core/correction.py b/src/cleveragents/domain/models/core/correction.py index 1d703bd39..d12b7f283 100644 --- a/src/cleveragents/domain/models/core/correction.py +++ b/src/cleveragents/domain/models/core/correction.py @@ -372,6 +372,165 @@ class CascadeAction(BaseModel): return v +class CorrectionAttemptState(StrEnum): + """Lifecycle states for a correction attempt record (spec DDL). + + Spec lifecycle: ``pending → executing → complete|failed``. + Terminal states (``complete``, ``failed``) allow no further transitions. + """ + + PENDING = "pending" + EXECUTING = "executing" + COMPLETE = "complete" + FAILED = "failed" + + +#: Spec-defined lifecycle transitions for correction attempts. +#: ``pending`` may only move to ``executing``; +#: ``executing`` may only move to ``complete`` or ``failed``. +#: Terminal states have no outgoing transitions. +CORRECTION_ATTEMPT_VALID_TRANSITIONS: dict[ + CorrectionAttemptState, frozenset[CorrectionAttemptState] +] = { + CorrectionAttemptState.PENDING: frozenset({CorrectionAttemptState.EXECUTING}), + CorrectionAttemptState.EXECUTING: frozenset( + {CorrectionAttemptState.COMPLETE, CorrectionAttemptState.FAILED} + ), +} + +#: Terminal states from which no further transitions are allowed. +CORRECTION_ATTEMPT_TERMINAL_STATES: frozenset[CorrectionAttemptState] = frozenset( + {CorrectionAttemptState.COMPLETE, CorrectionAttemptState.FAILED} +) + + +def validate_correction_state_transition( + current: CorrectionAttemptState, + target: CorrectionAttemptState, +) -> None: + """Validate a correction attempt state transition per the spec lifecycle. + + Args: + current: The current state. + target: The desired target state. + + Raises: + ValueError: If the transition is not allowed by the spec + lifecycle (``pending → executing → complete|failed``). + """ + allowed = CORRECTION_ATTEMPT_VALID_TRANSITIONS.get( + current, + frozenset(), + ) + if target not in allowed: + msg = f"Invalid state transition from '{current.value}' to '{target.value}'" + raise ValueError(msg) + + +class CorrectionAttemptRecord(BaseModel): + """Spec-aligned record of a correction attempt workflow. + + Maps to the ``correction_attempts`` table defined in the specification + DDL. Tracks the full lifecycle of a decision correction including the + plan, original and new decisions, mode (revert/append), guidance text, + archived artifacts path, state, and timestamps. + """ + + model_config = ConfigDict(frozen=False, populate_by_name=True) + + correction_attempt_id: str = Field( + default_factory=lambda: str(ULID()), + description="Unique identifier (ULID) for the correction attempt.", + ) + plan_id: str = Field( + ..., + description="Plan that owns the targeted decision tree.", + ) + original_decision_id: str = Field( + ..., + description="Decision node being corrected.", + ) + new_decision_id: str | None = Field( + default=None, + description="New decision created by the correction (if any).", + ) + mode: CorrectionMode = Field( + ..., + description="Correction strategy: revert or append.", + ) + guidance: str = Field( + ..., + max_length=10_000, + description="Human-supplied guidance for the correction.", + ) + archived_artifacts_path: str | None = Field( + default=None, + description="Path to archived artifacts (revert mode).", + ) + state: CorrectionAttemptState = Field( + default=CorrectionAttemptState.PENDING, + description="Current lifecycle state.", + ) + created_at: datetime = Field( + default_factory=lambda: datetime.now(UTC), + description="Timestamp when the record was created.", + ) + completed_at: datetime | None = Field( + default=None, + description="Timestamp when the attempt completed (None if still running).", + ) + + @field_validator("plan_id") + @classmethod + def _plan_id_not_empty(cls, v: str) -> str: + stripped = v.strip() + if not stripped: + raise ValueError("plan_id must not be empty") + return stripped + + @field_validator("original_decision_id") + @classmethod + def _original_decision_id_not_empty(cls, v: str) -> str: + stripped = v.strip() + if not stripped: + raise ValueError("original_decision_id must not be empty") + return stripped + + @field_validator("new_decision_id") + @classmethod + def _new_decision_id_not_empty(cls, v: str | None) -> str | None: + if v is None: + return v + stripped = v.strip() + if not stripped: + raise ValueError("new_decision_id must not be empty when set") + return stripped + + @field_validator("guidance") + @classmethod + def _guidance_not_empty(cls, v: str) -> str: + stripped = v.strip() + if not stripped: + raise ValueError("guidance must not be empty") + return stripped + + @field_validator("created_at") + @classmethod + def _created_at_tz_aware(cls, v: datetime) -> datetime: + """Ensure ``created_at`` is timezone-aware (default to UTC if naive).""" + if v.tzinfo is None: + return v.replace(tzinfo=UTC) + return v + + @field_validator("completed_at") + @classmethod + def _completed_at_tz_aware(cls, v: datetime | None) -> datetime | None: + """Ensure ``completed_at`` is timezone-aware when set.""" + if v is not None and v.tzinfo is None: + return v.replace(tzinfo=UTC) + return v + + class CorrectionRejection(BaseModel): """Result when a correction is rejected due to already-applied child plans. @@ -436,10 +595,14 @@ class CascadeResult(BaseModel): __all__ = [ + "CORRECTION_ATTEMPT_TERMINAL_STATES", + "CORRECTION_ATTEMPT_VALID_TRANSITIONS", "CascadeAction", "CascadeResult", "ChildPlanState", "CorrectionAttempt", + "CorrectionAttemptRecord", + "CorrectionAttemptState", "CorrectionDryRunReport", "CorrectionImpact", "CorrectionMode", @@ -447,4 +610,5 @@ __all__ = [ "CorrectionRequest", "CorrectionResult", "CorrectionStatus", + "validate_correction_state_transition", ] diff --git a/src/cleveragents/infrastructure/database/models.py b/src/cleveragents/infrastructure/database/models.py index fac86667b..ae88f7924 100644 --- a/src/cleveragents/infrastructure/database/models.py +++ b/src/cleveragents/infrastructure/database/models.py @@ -24,6 +24,7 @@ Alembic migrations. | ``checkpoint_metadata`` | ``CheckpointModel`` | Plan checkpoints | | ``repo_indexes`` | ``RepoIndexModel`` | Repo index metadata| | ``indexed_files`` | ``IndexedFileModel`` | Per-file records | +| ``correction_attempts`` | ``CorrectionAttemptModel`` | Correction records | Based on ADR-007 (Repository Pattern) and Phase 0 discovery. Includes spec-aligned lifecycle models per Stage A5 @@ -40,6 +41,7 @@ from typing import TYPE_CHECKING, Any, cast if TYPE_CHECKING: from cleveragents.domain.models.core.checkpoint import Checkpoint + from cleveragents.domain.models.core.correction import CorrectionAttemptRecord from sqlalchemy import ( JSON, @@ -56,6 +58,7 @@ from sqlalchemy import ( Text, UniqueConstraint, create_engine, + text, ) from sqlalchemy.orm import ( Mapped, @@ -690,6 +693,12 @@ class LifecyclePlanModel(Base): # type: ignore[misc] cascade="all, delete-orphan", order_by="PlanInvariantModel.position", ) + correction_attempts_rel = relationship( + "CorrectionAttemptModel", + back_populates="plan", + cascade="all, delete-orphan", + order_by="CorrectionAttemptModel.created_at", + ) __table_args__ = ( CheckConstraint( @@ -2949,6 +2958,234 @@ class CheckpointModel(Base): # type: ignore[misc] ) +# --------------------------------------------------------------------------- +# Correction Attempt Models (spec DDL — correction_attempts table) +# --------------------------------------------------------------------------- + +#: Length of a SQLite-compatible ISO-8601 timestamp truncated to millisecond +#: Length of the millisecond-precision ISO-8601 timestamp string: +#: ``YYYY-MM-DDTHH:MM:SS.mmm`` (23 characters). Matches the output +#: of SQLite ``strftime('%Y-%m-%dT%H:%M:%f', 'now')``. +SQLITE_TIMESTAMP_MS_LEN: int = 23 + + +def format_sqlite_timestamp(dt: datetime) -> str: + """Format a datetime to millisecond-precision ISO-8601 for SQLite. + + Normalises to UTC via ``astimezone(UTC)`` and truncates to the + ``YYYY-MM-DDTHH:MM:SS.mmm`` format matching the SQLite + ``server_default`` ``strftime('%Y-%m-%dT%H:%M:%f', 'now')``. + + Args: + dt: A timezone-aware datetime to format. + + Returns: + A 23-character ISO-8601 timestamp string in UTC. + + Raises: + ValueError: If *dt* is a naive datetime (no ``tzinfo``). + """ + if dt.tzinfo is None: + raise ValueError( + "format_sqlite_timestamp requires a timezone-aware datetime, " + f"got naive datetime: {dt!r}" + ) + utc_dt = dt.astimezone(UTC) + return utc_dt.strftime("%Y-%m-%dT%H:%M:%S.%f")[:SQLITE_TIMESTAMP_MS_LEN] + + +class CorrectionAttemptModel(Base): # type: ignore[misc] + """Database model for correction attempt records. + + Tracks decision correction workflows (revert/append modes) as defined + in the specification DDL (``correction_attempts`` table). Each row records one + correction attempt against a plan's decision tree, including the + original decision, optionally the new replacement decision, the mode, + human guidance, archived artifacts path, lifecycle state, and + timestamps. + + Table: ``correction_attempts`` + """ + + __allow_unmapped__ = True + __tablename__ = "correction_attempts" + + # PK: ULID (26-char string) + correction_attempt_id = Column(String(26), primary_key=True) + + # FK to v3_plans. Uses CASCADE (not the spec DDL default RESTRICT) + # for consistency with all other v3_plans child tables in the codebase + # (PlanProjectModel, PlanArgumentModel, PlanInvariantModel, etc.). + plan_id = Column( + String(26), + ForeignKey("v3_plans.plan_id", ondelete="CASCADE"), + nullable=False, + ) + + # FK to decisions — the decision being corrected. + # Uses RESTRICT (spec DDL default) to preserve correction audit trail + # when decisions are cleaned up — consistent with non-dependency FK + # references to decisions elsewhere in the codebase. + original_decision_id = Column( + String(26), + ForeignKey("decisions.decision_id", ondelete="RESTRICT"), + nullable=False, + ) + + # FK to decisions — the new decision created by the correction (nullable). + # Uses RESTRICT (spec DDL default: no ON DELETE clause) to preserve + # the correction audit trail, consistent with original_decision_id. + new_decision_id = Column( + String(26), + ForeignKey("decisions.decision_id", ondelete="RESTRICT"), + nullable=True, + ) + + # Correction strategy: 'revert' | 'append' + mode = Column(String(20), nullable=False) + + # Human-supplied guidance text + guidance = Column(Text, nullable=False) + + # Path to archived artifacts (revert mode) + archived_artifacts_path = Column(Text, nullable=True) + + # Lifecycle state: 'pending' | 'executing' | 'complete' | 'failed' + state = Column( + String(20), + nullable=False, + default="pending", + server_default="pending", + ) + + # Timestamps (ISO-8601 strings) + created_at = Column( + String(30), + nullable=False, + server_default=text("(strftime('%Y-%m-%dT%H:%M:%f', 'now'))"), + ) + completed_at = Column(String(30), nullable=True) + + # Relationships + plan = relationship( + "LifecyclePlanModel", + back_populates="correction_attempts_rel", + foreign_keys=[plan_id], + ) + + __table_args__ = ( + CheckConstraint( + "mode IN ('revert', 'append')", + name="ck_correction_attempts_mode", + ), + CheckConstraint( + "state IN ('pending', 'executing', 'complete', 'failed')", + name="ck_correction_attempts_state", + ), + Index("idx_corrections_plan", "plan_id"), + ) + + # -- Domain conversion helpers ------------------------------------------ + + def to_domain(self) -> CorrectionAttemptRecord: + """Convert to ``CorrectionAttemptRecord`` domain model. + + Returns: + A ``CorrectionAttemptRecord`` domain instance. + """ + from cleveragents.domain.models.core.correction import ( + CorrectionAttemptRecord, + CorrectionAttemptState, + CorrectionMode, + ) + + completed_at_dt = None + raw_completed = cast("str | None", self.completed_at) + if raw_completed: + completed_at_dt = datetime.fromisoformat(raw_completed) + if completed_at_dt.tzinfo is None: + completed_at_dt = completed_at_dt.replace(tzinfo=UTC) + + created_at_dt = datetime.fromisoformat(cast(str, self.created_at)) + if created_at_dt.tzinfo is None: + created_at_dt = created_at_dt.replace(tzinfo=UTC) + + raw_mode = cast(str, self.mode) + raw_state = cast(str, self.state) + raw_guidance = cast(str, self.guidance) + try: + mode_enum = CorrectionMode(raw_mode) + except ValueError: + _logger.warning( + "Unknown correction mode '%s' for attempt %s; defaulting to 'revert'", + raw_mode, + self.correction_attempt_id, + ) + mode_enum = CorrectionMode.REVERT + try: + state_enum = CorrectionAttemptState(raw_state) + except ValueError: + _logger.warning( + "Unknown correction state '%s' for attempt %s; defaulting to 'failed'", + raw_state, + self.correction_attempt_id, + ) + state_enum = CorrectionAttemptState.FAILED + if not raw_guidance or not raw_guidance.strip(): + _logger.warning( + "Empty/whitespace guidance for attempt %s; defaulting to '[corrupted]'", + self.correction_attempt_id, + ) + raw_guidance = "[corrupted]" + + return CorrectionAttemptRecord( + correction_attempt_id=cast(str, self.correction_attempt_id), + plan_id=cast(str, self.plan_id), + original_decision_id=cast(str, self.original_decision_id), + new_decision_id=cast("str | None", self.new_decision_id), + mode=mode_enum, + guidance=raw_guidance, + archived_artifacts_path=cast("str | None", self.archived_artifacts_path), + state=state_enum, + created_at=created_at_dt, + completed_at=completed_at_dt, + ) + + @classmethod + def from_domain(cls, record: CorrectionAttemptRecord) -> CorrectionAttemptModel: + """Create from ``CorrectionAttemptRecord`` domain model. + + Timestamps are normalised to UTC before formatting to prevent + silent data loss when non-UTC timezone-aware datetimes are provided. + The output format ``YYYY-MM-DDTHH:MM:SS.mmm`` (no timezone offset, + millisecond precision) matches the SQLite ``server_default`` + ``strftime('%Y-%m-%dT%H:%M:%f', 'now')`` for consistent + string-based ordering. + + Args: + record: A ``CorrectionAttemptRecord`` domain instance. + + Returns: + A ``CorrectionAttemptModel`` ready for persistence. + """ + completed_at_str: str | None = None + if record.completed_at is not None: + completed_at_str = format_sqlite_timestamp(record.completed_at) + + return cls( + correction_attempt_id=record.correction_attempt_id, + plan_id=record.plan_id, + original_decision_id=record.original_decision_id, + new_decision_id=record.new_decision_id, + mode=record.mode.value, + guidance=record.guidance, + archived_artifacts_path=record.archived_artifacts_path, + state=record.state.value, + created_at=format_sqlite_timestamp(record.created_at), + completed_at=completed_at_str, + ) + + # Database initialization functions def init_database(database_url: str = "sqlite:///.cleveragents/db.sqlite") -> Any: """Initialize the database. diff --git a/src/cleveragents/infrastructure/database/repositories.py b/src/cleveragents/infrastructure/database/repositories.py index cf17c1819..c9f6e7cc3 100644 --- a/src/cleveragents/infrastructure/database/repositories.py +++ b/src/cleveragents/infrastructure/database/repositories.py @@ -12,6 +12,7 @@ Provides database-backed repositories following the session-factory pattern | ``NamespacedProjectRepository``| CRUD for namespaced projects | | ``ProjectResourceLinkRepository``| Project-resource link management | | ``DecisionRepository`` | CRUD for decision tree nodes | +| ``CorrectionAttemptRepository``| CRUD for correction attempt records | ## Session-Factory Pattern @@ -46,6 +47,9 @@ with uow: | ``DuplicateLinkError`` | Project-resource link already exists | | ``DuplicateDecisionError``| Decision ID already exists | | ``DecisionNotFoundError``| Decision ID not found | +| ``CorrectionAttemptNotFoundError``| Correction attempt not found | +| ``DuplicateCorrectionAttemptError``| Correction attempt ID already exists | +| ``InvalidCorrectionStateTransitionError``| Invalid lifecycle transition | Based on ADR-007 (Repository Pattern) and ADR-033 (Retry Patterns). """ @@ -91,6 +95,7 @@ from cleveragents.infrastructure.database.models import ( ChangeModel, CheckpointModel, ContextModel, + CorrectionAttemptModel, DebugAttemptModel, DecisionModel, LifecycleActionModel, @@ -113,11 +118,21 @@ from cleveragents.infrastructure.database.models import ( ToolModel, ToolResourceBindingModel, ValidationAttachmentModel, + format_sqlite_timestamp, ) if TYPE_CHECKING: + from cleveragents.domain.models.core.correction import ( + CorrectionAttemptRecord, + ) from cleveragents.domain.models.core.decision import Decision +from cleveragents.domain.models.core.correction import ( + CORRECTION_ATTEMPT_TERMINAL_STATES, + CorrectionAttemptState, + validate_correction_state_transition, +) + _log = structlog.get_logger(__name__) @@ -5663,3 +5678,304 @@ class CheckpointRepository: raise DatabaseError( f"Failed to prune checkpoints for plan {plan_id}: {exc}" ) from exc + + +# --------------------------------------------------------------------------- +# Correction Attempt Repository +# --------------------------------------------------------------------------- + + +class CorrectionAttemptNotFoundError(DatabaseError): + """Raised when a correction attempt record is not found.""" + + +class DuplicateCorrectionAttemptError(DatabaseError): + """Raised when a correction attempt ID already exists.""" + + +class InvalidCorrectionStateTransitionError(BusinessRuleViolation): + """Raised when an invalid state transition is attempted.""" + + +class CorrectionAttemptRepository: + """CRUD operations for the ``correction_attempts`` table. + + Follows the session-factory pattern (ADR-007) consistent with + other repositories in this module. State transition validation + is delegated to the domain-level + ``validate_correction_state_transition`` function. + """ + + def __init__(self, session_factory: Callable[[], Session]) -> None: + """Initialise with a callable that returns a SQLAlchemy Session.""" + self._session_factory = session_factory + + def _session(self) -> Session: + """Convenience helper to obtain a session.""" + return self._session_factory() + + # --- CREATE ------------------------------------------------------------ + + @database_retry + def create(self, record: CorrectionAttemptRecord) -> CorrectionAttemptRecord: + """Persist a new correction attempt record. + + Args: + record: A ``CorrectionAttemptRecord`` domain instance. + + Returns: + The persisted ``CorrectionAttemptRecord`` domain object. + + Raises: + DuplicateCorrectionAttemptError: If the ID already exists. + DatabaseError: On transient or unexpected DB errors. + """ + session = self._session() + try: + model = CorrectionAttemptModel.from_domain(record) + session.add(model) + session.flush() + return model.to_domain() + except IntegrityError as exc: + session.rollback() + exc_str = str(exc).upper() + if "UNIQUE" in exc_str: + raise DuplicateCorrectionAttemptError( + f"Correction attempt already exists: {record.correction_attempt_id}" + ) from exc + if "FOREIGN KEY" in exc_str or "FOREIGN_KEY" in exc_str: + raise DatabaseError( + f"Foreign key constraint violated creating correction " + f"attempt {record.correction_attempt_id}: verify that " + f"plan_id '{record.plan_id}' and decision IDs exist" + ) from exc + raise DatabaseError(f"Failed to create correction attempt: {exc}") from exc + except (OperationalError, SQLAlchemyDatabaseError) as exc: + session.rollback() + raise DatabaseError(f"Failed to create correction attempt: {exc}") from exc + + # --- GET --------------------------------------------------------------- + + @database_retry + def get(self, correction_attempt_id: str) -> CorrectionAttemptRecord: + """Retrieve a correction attempt by ID. + + Args: + correction_attempt_id: ULID of the correction attempt. + + Returns: + A ``CorrectionAttemptRecord`` domain instance. + + Raises: + CorrectionAttemptNotFoundError: If not found. + DatabaseError: On transient or unexpected DB errors. + """ + session = self._session() + try: + row = ( + session.query(CorrectionAttemptModel) + .filter_by(correction_attempt_id=correction_attempt_id) + .first() + ) + if row is None: + raise CorrectionAttemptNotFoundError( + f"Correction attempt not found: {correction_attempt_id}" + ) + return row.to_domain() + except CorrectionAttemptNotFoundError: + raise + except (OperationalError, SQLAlchemyDatabaseError) as exc: + raise DatabaseError( + f"Failed to get correction attempt {correction_attempt_id}: {exc}" + ) from exc + + # --- LIST BY PLAN ------------------------------------------------------ + + @database_retry + def list_by_plan(self, plan_id: str) -> list[CorrectionAttemptRecord]: + """List all correction attempts for a plan, ordered by creation time. + + Args: + plan_id: Plan ULID. + + Returns: + List of ``CorrectionAttemptRecord`` domain objects. + + Raises: + DatabaseError: On transient or unexpected DB errors. + """ + session = self._session() + try: + rows = ( + session.query(CorrectionAttemptModel) + .filter_by(plan_id=plan_id) + .order_by(CorrectionAttemptModel.created_at) + .all() + ) + return [row.to_domain() for row in rows] + except (OperationalError, SQLAlchemyDatabaseError) as exc: + raise DatabaseError( + f"Failed to list correction attempts for plan {plan_id}: {exc}" + ) from exc + + # --- UPDATE STATE ------------------------------------------------------ + + @database_retry + def update_state( + self, + correction_attempt_id: str, + state: CorrectionAttemptState, + completed_at: datetime | None = None, + new_decision_id: str | None = None, + archived_artifacts_path: str | None = None, + ) -> CorrectionAttemptRecord: + """Update the state and optional fields of a correction attempt. + + Args: + correction_attempt_id: ULID of the correction attempt. + state: New lifecycle state (``CorrectionAttemptState`` enum). + completed_at: Optional completion timestamp. + new_decision_id: Optional new decision ID (for append mode). + archived_artifacts_path: Optional archived artifacts path. + + Returns: + The updated ``CorrectionAttemptRecord`` domain instance. + + Raises: + CorrectionAttemptNotFoundError: If not found. + InvalidCorrectionStateTransitionError: If the state transition + is not allowed or ``completed_at`` is set on a non-terminal + transition. + DatabaseError: On input validation failure, FK violation, or + transient/unexpected DB errors. + """ + # -- Argument validation (before any session/row mutation) ---------- + # Per CONTRIBUTING.md: validate arguments as the first guard. + # Use ValueError per CONTRIBUTING.md guidelines for invalid + # values/ranges/empty inputs. + stripped_decision_id: str | None = None + if new_decision_id is not None: + stripped_decision_id = new_decision_id.strip() + if not stripped_decision_id: + raise ValueError("new_decision_id must not be empty or whitespace-only") + stripped_artifacts_path: str | None = None + if archived_artifacts_path is not None: + stripped_artifacts_path = archived_artifacts_path.strip() + if not stripped_artifacts_path: + raise ValueError( + "archived_artifacts_path must not be empty or whitespace-only" + ) + + session = self._session() + try: + row = ( + session.query(CorrectionAttemptModel) + .filter_by(correction_attempt_id=correction_attempt_id) + .first() + ) + if row is None: + raise CorrectionAttemptNotFoundError( + f"Correction attempt not found: {correction_attempt_id}" + ) + try: + current_state = CorrectionAttemptState(cast(str, row.state)) + except ValueError: + _log.warning( + "Corrupted state '%s' for correction attempt %s; " + "treating as terminal 'failed'", + row.state, + correction_attempt_id, + ) + current_state = CorrectionAttemptState.FAILED + try: + validate_correction_state_transition(current_state, state) + except ValueError as exc: + raise InvalidCorrectionStateTransitionError( + f"{exc} for correction attempt {correction_attempt_id}" + ) from exc + + if ( + completed_at is not None + and state not in CORRECTION_ATTEMPT_TERMINAL_STATES + ): + raise InvalidCorrectionStateTransitionError( + f"completed_at can only be set for terminal states " + f"(complete, failed), not '{state.value}'" + ) + + # -- Apply mutations (all validation passed) ------------------- + row.state = state.value # type: ignore[assignment] + # Auto-set completed_at for terminal transitions if not provided + effective_completed_at = completed_at + if ( + effective_completed_at is None + and state in CORRECTION_ATTEMPT_TERMINAL_STATES + ): + effective_completed_at = datetime.now(UTC) + if effective_completed_at is not None: + row.completed_at = format_sqlite_timestamp( # type: ignore[assignment] + effective_completed_at, + ) + if stripped_decision_id is not None: + row.new_decision_id = stripped_decision_id # type: ignore[assignment] + if stripped_artifacts_path is not None: + row.archived_artifacts_path = stripped_artifacts_path # type: ignore[assignment] + session.flush() + return row.to_domain() + except (CorrectionAttemptNotFoundError, InvalidCorrectionStateTransitionError): + raise + except IntegrityError as exc: + session.rollback() + exc_str = str(exc).upper() + if "FOREIGN KEY" in exc_str or "FOREIGN_KEY" in exc_str: + fk_details = ( + f"new_decision_id='{new_decision_id}'" + if new_decision_id is not None + else "referenced entity" + ) + raise DatabaseError( + f"Foreign key constraint violated updating correction " + f"attempt {correction_attempt_id}: verify that " + f"{fk_details} exists" + ) from exc + raise DatabaseError( + f"Failed to update correction attempt {correction_attempt_id}: {exc}" + ) from exc + except (OperationalError, SQLAlchemyDatabaseError) as exc: + session.rollback() + raise DatabaseError( + f"Failed to update correction attempt {correction_attempt_id}: {exc}" + ) from exc + + # --- DELETE ------------------------------------------------------------- + + @database_retry + def delete(self, correction_attempt_id: str) -> bool: + """Delete a correction attempt by ID. + + Args: + correction_attempt_id: ULID of the correction attempt. + + Returns: + ``True`` if deleted, ``False`` if not found. + + Raises: + DatabaseError: On transient or unexpected DB errors. + """ + session = self._session() + try: + row = ( + session.query(CorrectionAttemptModel) + .filter_by(correction_attempt_id=correction_attempt_id) + .first() + ) + if row is None: + return False + session.delete(row) + session.flush() + return True + except (OperationalError, SQLAlchemyDatabaseError) as exc: + session.rollback() + raise DatabaseError( + f"Failed to delete correction attempt {correction_attempt_id}: {exc}" + ) from exc diff --git a/src/cleveragents/infrastructure/database/unit_of_work.py b/src/cleveragents/infrastructure/database/unit_of_work.py index 29268fcb4..357bd6570 100644 --- a/src/cleveragents/infrastructure/database/unit_of_work.py +++ b/src/cleveragents/infrastructure/database/unit_of_work.py @@ -19,6 +19,7 @@ from cleveragents.infrastructure.database.repositories import ( ChangeRepository, CheckpointRepository, ContextRepository, + CorrectionAttemptRepository, DebugAttemptRepository, DecisionRepository, LifecyclePlanRepository, @@ -194,6 +195,7 @@ class UnitOfWorkContext: self._lifecycle_plans: LifecyclePlanRepository | None = None self._decisions: DecisionRepository | None = None self._checkpoints: CheckpointRepository | None = None + self._correction_attempts: CorrectionAttemptRepository | None = None def _session_factory(self) -> Session: """Return the transaction's session for factory-pattern repositories.""" @@ -293,6 +295,18 @@ class UnitOfWorkContext: ) return self._decisions + @property + def correction_attempts(self) -> CorrectionAttemptRepository: + """Get correction attempt repository for this transaction. + + Uses the session-factory pattern required by CorrectionAttemptRepository. + """ + if self._correction_attempts is None: + self._correction_attempts = CorrectionAttemptRepository( + session_factory=self._session_factory, + ) + return self._correction_attempts + def add(self, entity: Any) -> None: """Add an entity to the session. -- 2.52.0