"""Step definitions for decision_phase_gating.feature. All ``Then`` step texts are prefixed with ``phase-gated`` or use ``phase violation`` to avoid collisions with existing decision step definitions. """ from __future__ import annotations import contextlib import os import tempfile import structlog from behave import given, then, when from behave.runner import Context from ulid import ULID from cleveragents.application.services.decision_service import DecisionService from cleveragents.application.services.phase_gating import ( PHASE_ALLOWED_TYPES, resolve_plan_phase, validate_phase_gating, ) from cleveragents.core.exceptions import DecisionPhaseViolationError, ValidationError from cleveragents.domain.models.core.decision import ( EXECUTE_TYPES, STRATEGIZE_TYPES, DecisionType, ) from cleveragents.domain.models.core.plan import ( NamespacedName, Plan, PlanIdentity, PlanPhase, ProcessingState, ) from cleveragents.infrastructure.database.unit_of_work import UnitOfWork def _pg_plan_id(context: Context) -> str: """Return or create a stable ULID for the phase-gating test plan.""" if not hasattr(context, "_pg_plan_id"): context._pg_plan_id = str(ULID()) return context._pg_plan_id # --------------------------------------------------------------------------- # Background # --------------------------------------------------------------------------- @given("a phase-gated decision service") def step_phase_gated_service(context: Context) -> None: context.pg_service = DecisionService() context.pg_result = None context.pg_error = None context._pg_plan_id = str(ULID()) context._pg_seq = 0 # --------------------------------------------------------------------------- # Given # --------------------------------------------------------------------------- @given("a phase-gated decision service with a persisted strategize plan") def step_persisted_strategize_plan(context: Context) -> None: """Create a DecisionService backed by a real database with a plan in Strategize.""" _setup_persisted_plan(context, PlanPhase.STRATEGIZE) @given("a phase-gated decision service with a persisted execute plan") def step_persisted_execute_plan(context: Context) -> None: """Create a DecisionService backed by a real database with a plan in Execute.""" _setup_persisted_plan(context, PlanPhase.EXECUTE) @given("a phase-gated decision service with an empty database") def step_persisted_empty_db(context: Context) -> None: """Create a DecisionService with a wired UoW but no plans in the DB.""" fd, db_path = tempfile.mkstemp(suffix=".db", prefix="pg_empty_") os.close(fd) context._pg_db_path = db_path uow = UnitOfWork(f"sqlite:///{db_path}") uow.init_database() context._pg_plan_id = str(ULID()) context.pg_service = DecisionService(unit_of_work=uow) context.pg_result = None context.pg_error = None context._pg_seq = 0 _register_db_cleanup(context, db_path, uow) def _setup_persisted_plan(context: Context, phase: PlanPhase) -> None: """Shared helper: create a persisted plan in the given *phase*.""" fd, db_path = tempfile.mkstemp(suffix=".db", prefix="pg_persisted_") os.close(fd) context._pg_db_path = db_path uow = UnitOfWork(f"sqlite:///{db_path}") uow.init_database() plan_id = str(ULID()) context._pg_plan_id = plan_id plan = Plan( identity=PlanIdentity(plan_id=plan_id), namespaced_name=NamespacedName(namespace="local", name="pg-test"), description="Phase gating test plan", action_name="local/pg-test-action", phase=phase, processing_state=ProcessingState.PROCESSING, ) with uow.transaction() as txn: txn.lifecycle_plans.create(plan) context.pg_service = DecisionService(unit_of_work=uow) context.pg_result = None context.pg_error = None context._pg_seq = 0 _register_db_cleanup(context, db_path, uow) def _register_db_cleanup(context: Context, db_path: str, uow: UnitOfWork) -> None: """Register DB cleanup handler for the test context.""" def _cleanup_db() -> None: uow.engine.dispose() for suffix in ("", "-journal", "-wal", "-shm"): with contextlib.suppress(OSError): os.unlink(db_path + suffix) if not hasattr(context, "_cleanup_handlers"): context._cleanup_handlers = [] context._cleanup_handlers.append(_cleanup_db) @given("a phase-gated decision service without persistence") def step_no_persistence_service(context: Context) -> None: context.pg_service = DecisionService() context.pg_result = None context.pg_error = None context._pg_plan_id = str(ULID()) context._pg_seq = 0 @given('a DecisionPhaseViolationError for type "{dtype}" in phase "{phase}"') def step_create_phase_violation_error(context: Context, dtype: str, phase: str) -> None: allowed = PHASE_ALLOWED_TYPES.get(PlanPhase(phase), frozenset()) context.pg_error = DecisionPhaseViolationError( decision_type=dtype, plan_phase=phase, allowed_types=frozenset(str(t) for t in allowed), ) # --------------------------------------------------------------------------- # When — Recording with phase # --------------------------------------------------------------------------- @when('I record a "{dtype}" decision in the "{phase}" phase') def step_record_with_phase(context: Context, dtype: str, phase: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) # Determine parent_decision_id: prompt_definition must be root parent_id = None dt = DecisionType(dtype) if dt != DecisionType.PROMPT_DEFINITION: parent_id = str(ULID()) d = svc.record_decision( plan_id=plan_id, decision_type=dtype, question=f"Phase-gated question ({dtype})", chosen_option=f"Chosen ({dtype})", parent_decision_id=parent_id, plan_phase=phase, ) context.pg_result = d context.pg_error = None @when('I try to record a "{dtype}" decision in the "{phase}" phase') def step_try_record_with_phase(context: Context, dtype: str, phase: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) parent_id = str(ULID()) try: svc.record_decision( plan_id=plan_id, decision_type=dtype, question=f"Phase-gated question ({dtype})", chosen_option=f"Chosen ({dtype})", parent_decision_id=parent_id, plan_phase=phase, ) context.pg_error = None except DecisionPhaseViolationError as exc: context.pg_error = exc @when('I record a "{dtype}" decision without explicit phase') def step_record_without_phase(context: Context, dtype: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) parent_id = str(ULID()) dt = DecisionType(dtype) if dt == DecisionType.PROMPT_DEFINITION: parent_id = None try: d = svc.record_decision( plan_id=plan_id, decision_type=dtype, question=f"No-phase question ({dtype})", chosen_option=f"Chosen ({dtype})", parent_decision_id=parent_id, ) context.pg_result = d context.pg_error = None except DecisionPhaseViolationError as exc: context.pg_error = exc context.pg_result = None @when('I try to record a "{dtype}" decision without explicit phase') def step_try_record_without_phase(context: Context, dtype: str) -> None: step_record_without_phase(context, dtype) @when('I call _validate_phase_gating with type "{dtype}" and phase "{phase}"') def step_validate_phase_gating_direct(context: Context, dtype: str, phase: str) -> None: try: validate_phase_gating( DecisionType(dtype), PlanPhase(phase), ) context.pg_error = None except DecisionPhaseViolationError as exc: context.pg_error = exc @when('I record a "{dtype}" decision with string phase "{phase}"') def step_record_with_string_phase(context: Context, dtype: str, phase: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) parent_id = None dt = DecisionType(dtype) if dt != DecisionType.PROMPT_DEFINITION: parent_id = str(ULID()) d = svc.record_decision( plan_id=plan_id, decision_type=dtype, question=f"String-phase question ({dtype})", chosen_option=f"Chosen ({dtype})", parent_decision_id=parent_id, plan_phase=phase, ) context.pg_result = d context.pg_error = None @when('I record a "{dtype}" decision with PlanPhase enum "{phase}"') def step_record_with_enum_phase(context: Context, dtype: str, phase: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) parent_id = None dt = DecisionType(dtype) if dt != DecisionType.PROMPT_DEFINITION: parent_id = str(ULID()) d = svc.record_decision( plan_id=plan_id, decision_type=dtype, question=f"Enum-phase question ({dtype})", chosen_option=f"Chosen ({dtype})", parent_decision_id=parent_id, plan_phase=PlanPhase(phase), ) context.pg_result = d context.pg_error = None # --------------------------------------------------------------------------- # Then — Assertions # --------------------------------------------------------------------------- @then("the phase-gated decision should be recorded successfully") def step_pg_recorded(context: Context) -> None: assert context.pg_result is not None, "Expected a recorded decision but got None" assert context.pg_error is None @then('the phase-gated decision type should be "{dtype}"') def step_pg_type(context: Context, dtype: str) -> None: assert str(context.pg_result.decision_type) == dtype @then("a phase violation error should be raised") def step_pg_error_raised(context: Context) -> None: assert context.pg_error is not None, ( "Expected DecisionPhaseViolationError but no error was raised" ) assert isinstance(context.pg_error, DecisionPhaseViolationError) @then('the phase violation error decision_type should be "{dtype}"') def step_pg_error_dtype(context: Context, dtype: str) -> None: assert context.pg_error.decision_type == dtype @then('the phase violation error plan_phase should be "{phase}"') def step_pg_error_phase(context: Context, phase: str) -> None: assert context.pg_error.plan_phase == phase @then('the phase violation error allowed_types should not contain "{dtype}"') def step_pg_error_not_in_allowed(context: Context, dtype: str) -> None: assert dtype not in context.pg_error.allowed_types @then("the phase violation error allowed_types should not be empty") def step_pg_error_allowed_not_empty(context: Context) -> None: assert len(context.pg_error.allowed_types) > 0 @then('the phase violation error message should contain "{text}"') def step_pg_error_message_contains(context: Context, text: str) -> None: assert text in str(context.pg_error), ( f"Expected '{text}' in error message: {context.pg_error}" ) @then("no phase violation error should be raised") def step_pg_no_error(context: Context) -> None: assert context.pg_error is None @then('the PHASE_ALLOWED_TYPES should map "{phase}" to STRATEGIZE_TYPES') def step_phase_map_strategize(context: Context, phase: str) -> None: assert PHASE_ALLOWED_TYPES[PlanPhase(phase)] is STRATEGIZE_TYPES @then('the PHASE_ALLOWED_TYPES should map "{phase}" to EXECUTE_TYPES') def step_phase_map_execute(context: Context, phase: str) -> None: assert PHASE_ALLOWED_TYPES[PlanPhase(phase)] is EXECUTE_TYPES # --------------------------------------------------------------------------- # Invalid plan_phase string # --------------------------------------------------------------------------- @when('I try to record a "{dtype}" decision with invalid phase "{phase}"') def step_try_record_invalid_phase(context: Context, dtype: str, phase: str) -> None: svc: DecisionService = context.pg_service plan_id = _pg_plan_id(context) parent_id = str(ULID()) try: svc.record_decision( plan_id=plan_id, decision_type=dtype, question=f"Invalid-phase question ({dtype})", chosen_option=f"Chosen ({dtype})", parent_decision_id=parent_id, plan_phase=phase, ) context.pg_error = None except ValidationError as exc: context.pg_error = exc @then("a validation error should be raised for invalid phase") def step_validation_error_raised(context: Context) -> None: assert context.pg_error is not None, ( "Expected ValidationError but no error was raised" ) assert isinstance(context.pg_error, ValidationError) @then('the validation error message should contain "{text}"') def step_validation_error_message(context: Context, text: str) -> None: assert text in str(context.pg_error), ( f"Expected '{text}' in error message: {context.pg_error}" ) # --------------------------------------------------------------------------- # Database error resilience # --------------------------------------------------------------------------- @given("a corrupted database unit of work for phase resolution") def step_corrupted_uow_for_resolve(context: Context) -> None: """Create a real UoW whose DB is then corrupted for resolve_plan_phase testing.""" fd, db_path = tempfile.mkstemp(suffix=".db", prefix="pg_corrupt_") os.close(fd) uow = UnitOfWork(f"sqlite:///{db_path}") uow.init_database() # Dispose connections and corrupt the database file so the next # transaction attempt raises OperationalError/DatabaseError. uow.engine.dispose() with open(db_path, "wb") as f: f.write(b"\x00" * 16) context._pg_plan_id = str(ULID()) context._pg_corrupted_uow = uow context._pg_resolve_result = None _register_db_cleanup(context, db_path, uow) @when("I call resolve_plan_phase with the corrupted unit of work") def step_call_resolve_with_corrupted_uow(context: Context) -> None: logger = structlog.get_logger("test.phase_gating") result = resolve_plan_phase( plan_id=context._pg_plan_id, explicit_phase=None, persisted=True, unit_of_work=context._pg_corrupted_uow, logger=logger, ) context._pg_resolve_result = result @then("resolve_plan_phase should return None") def step_resolve_returns_none(context: Context) -> None: assert context._pg_resolve_result is None, ( f"Expected None but got {context._pg_resolve_result}" )