HAL9000/temp
0
0
Fork 0
forked from HAL9000/cleveragents-core
Code Pull Requests Activity
Files
master
temp/features/steps/correction_attempt_persistence_steps.py
T
Luis Mendes f678d611bb 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
2026-03-29 16:23:04 +01:00

1264 lines
44 KiB
Python
Raw Permalink Blame History

"""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
View Git Blame Copy Permalink