From 296daebe59f2a62512edea69d5bd26d7dc9d5e0d Mon Sep 17 00:00:00 2001 From: Rui Hu Date: Mon, 16 Mar 2026 07:27:21 +0000 Subject: [PATCH] feat(plan): enforce decision type phase-gating at recording time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds phase-gating validation to DecisionService.record_decision() that enforces the specification constraint: certain decision types are only valid during specific plan phases. Phase assignment changes: - resource_selection added to EXECUTE_TYPES (was Strategize-only, now Strategize-or-Execute per ADR-007 and ADR-033). - subplan_spawn / subplan_parallel_spawn remain in both phase sets; code comment documents divergence from ADRs per M4 subplan model. - is_any_phase_type property updated to check both sets dynamically. - Module docstring table updated to match actual assignments. Service changes: - DecisionPhaseViolationError raised before persistence. - PHASE_ALLOWED_TYPES typed as Mapping (was dict). - _resolve_plan_phase: invalid plan_phase string now raises ValidationError (was uncaught ValueError). - _resolve_plan_phase: database errors caught and logged, falling through to skip gating (preserves opt-in contract). - TOCTOU and uncached-DB-lookup documented with code comments. Test additions: - DB-resolved phase for Execute plan (was only Strategize). - DB lookup when plan not found → gating skipped. - PlanPhase enum passed directly to plan_phase parameter. - Inline imports moved to module top-level per CONTRIBUTING.md. - tempfile.mktemp() replaced with mkstemp(). - UnitOfWork engine.dispose() added to test cleanup. - Robot tests now assert stderr is empty. - Flaky concurrency test timing increased. ISSUES CLOSED: #931 --- CHANGELOG.md | 7 + features/consolidated_decision.feature | 48 +- features/decision_phase_gating.feature | 172 +++++++ features/steps/decision_model_steps.py | 17 + features/steps/decision_phase_gating_steps.py | 444 ++++++++++++++++++ features/steps/subplan_execution_steps.py | 6 +- robot/decision_phase_gating.robot | 57 +++ robot/helper_decision_phase_gating.py | 156 ++++++ .../application/services/decision_service.py | 37 +- .../application/services/phase_gating.py | 148 ++++++ src/cleveragents/core/exceptions.py | 27 ++ .../domain/models/core/decision.py | 35 +- 12 files changed, 1140 insertions(+), 14 deletions(-) create mode 100644 features/decision_phase_gating.feature create mode 100644 features/steps/decision_phase_gating_steps.py create mode 100644 robot/decision_phase_gating.robot create mode 100644 robot/helper_decision_phase_gating.py create mode 100644 src/cleveragents/application/services/phase_gating.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 24a71e1e4..c4edcaa22 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,13 @@ orchestration, positive control, and auto-discovery of QUEUED plans. (`features/tdd_plan_execute_phase_processing.feature`, `robot/tdd_plan_execute_phase_processing.robot`) (#977) +- **Breaking (behavioral):** `resource_selection` decision type reclassified from + Execute-only to phase-agnostic (valid in both Strategize and Execute). + `DecisionType.RESOURCE_SELECTION` now appears in both `STRATEGIZE_TYPES` and + `EXECUTE_TYPES`. Code relying on `is_strategize_type` or `is_execute_type` + returning `False` for `resource_selection` will see different results. + Reclassification aligns with ADR-007 L72 and ADR-033 L74 which permit + resource selection during planning. (#931) - Added built-in deferred virtual resource types: `remote`, `submodule`, and `symlink` with equivalence metadata rules for cross-repo and cross-layer identity tracking. Registry bootstrap includes deferred virtual types but diff --git a/features/consolidated_decision.feature b/features/consolidated_decision.feature index 86219e393..8e69f59bd 100644 --- a/features/consolidated_decision.feature +++ b/features/consolidated_decision.feature @@ -14,9 +14,11 @@ Feature: Consolidated Decision Then STRATEGIZE_TYPES should contain "prompt_definition" And STRATEGIZE_TYPES should contain "invariant_enforced" And STRATEGIZE_TYPES should contain "strategy_choice" + And STRATEGIZE_TYPES should contain "resource_selection" And STRATEGIZE_TYPES should contain "subplan_spawn" And STRATEGIZE_TYPES should contain "subplan_parallel_spawn" - And STRATEGIZE_TYPES should have exactly 5 members + And STRATEGIZE_TYPES should contain "user_intervention" + And STRATEGIZE_TYPES should have exactly 7 members Scenario: Execute-phase types are correctly classified @@ -25,12 +27,15 @@ Feature: Consolidated Decision And EXECUTE_TYPES should contain "tool_invocation" And EXECUTE_TYPES should contain "error_recovery" And EXECUTE_TYPES should contain "validation_response" - And EXECUTE_TYPES should have exactly 5 members + And EXECUTE_TYPES should contain "subplan_spawn" + And EXECUTE_TYPES should contain "subplan_parallel_spawn" + And EXECUTE_TYPES should contain "user_intervention" + And EXECUTE_TYPES should have exactly 8 members - Scenario: user_intervention is not in either phase set - Then STRATEGIZE_TYPES should not contain "user_intervention" - And EXECUTE_TYPES should not contain "user_intervention" + Scenario: user_intervention is a phase-agnostic type in both sets + Then STRATEGIZE_TYPES should contain "user_intervention" + And EXECUTE_TYPES should contain "user_intervention" # ------------------------------------------------------------------ # Decision creation and ULID validation @@ -45,6 +50,7 @@ Feature: Consolidated Decision And the decision should have a valid ULID as decision_id And the decision type should be "prompt_definition" And the decision is_strategize_type should be true + And the decision is_any_phase_type should be false Scenario: Create a non-root decision with parent @@ -72,6 +78,38 @@ Feature: Consolidated Decision And the decision is_any_phase_type should be true + Scenario Outline: All dual-phase types report is_any_phase_type true + Given a valid plan ULID + And a valid parent decision ULID + When I create a "" decision with sequence 4 and a parent + Then the decision should be created successfully + And the decision is_any_phase_type should be true + + Examples: + | decision_type | + | resource_selection | + | subplan_spawn | + | subplan_parallel_spawn | + | user_intervention | + + + Scenario Outline: Single-phase types report is_any_phase_type false + Given a valid plan ULID + And a valid parent decision ULID + When I create a "" decision with sequence 5 and a parent + Then the decision should be created successfully + And the decision is_any_phase_type should be false + + Examples: + | decision_type | + | invariant_enforced | + | strategy_choice | + | implementation_choice | + | tool_invocation | + | error_recovery | + | validation_response | + + Scenario: Invalid plan_id is rejected When I try to create a decision with plan_id "not-a-ulid" Then a decision validation error should be raised diff --git a/features/decision_phase_gating.feature b/features/decision_phase_gating.feature new file mode 100644 index 000000000..46daaa283 --- /dev/null +++ b/features/decision_phase_gating.feature @@ -0,0 +1,172 @@ +Feature: Decision type phase-gating at recording time + As a developer + I want the DecisionService to enforce phase-gating on decision types + So that decisions recorded during Strategize or Execute are constrained + to the types permitted in each phase + + Background: + Given a phase-gated decision service + + # --- Valid types in Strategize phase --- + + Scenario Outline: Valid decision types succeed in Strategize phase + When I record a "" decision in the "strategize" phase + Then the phase-gated decision should be recorded successfully + And the phase-gated decision type should be "" + + Examples: + | decision_type | + | prompt_definition | + | invariant_enforced | + | strategy_choice | + | resource_selection | + | subplan_spawn | + | subplan_parallel_spawn | + | user_intervention | + + # --- Valid types in Execute phase --- + + Scenario Outline: Valid decision types succeed in Execute phase + When I record a "" decision in the "execute" phase + Then the phase-gated decision should be recorded successfully + And the phase-gated decision type should be "" + + Examples: + | decision_type | + | implementation_choice | + | resource_selection | + | tool_invocation | + | error_recovery | + | validation_response | + | subplan_spawn | + | subplan_parallel_spawn | + | user_intervention | + + # --- Invalid types in Strategize phase --- + + Scenario Outline: Execute-only types are rejected in Strategize phase + When I try to record a "" decision in the "strategize" phase + Then a phase violation error should be raised + And the phase violation error decision_type should be "" + And the phase violation error plan_phase should be "strategize" + And the phase violation error allowed_types should not contain "" + + Examples: + | decision_type | + | implementation_choice | + | tool_invocation | + | error_recovery | + | validation_response | + + # --- Invalid types in Execute phase --- + + Scenario Outline: Strategize-only types are rejected in Execute phase + When I try to record a "" decision in the "execute" phase + Then a phase violation error should be raised + And the phase violation error decision_type should be "" + And the phase violation error plan_phase should be "execute" + And the phase violation error allowed_types should not contain "" + + Examples: + | decision_type | + | prompt_definition | + | invariant_enforced | + | strategy_choice | + + # --- Phase-agnostic types work in both phases --- + + Scenario Outline: Phase-agnostic types succeed in both phases + When I record a "" decision in the "strategize" phase + Then the phase-gated decision should be recorded successfully + When I record a "" decision in the "execute" phase + Then the phase-gated decision should be recorded successfully + + Examples: + | decision_type | + | resource_selection | + | subplan_spawn | + | subplan_parallel_spawn | + | user_intervention | + + # --- Error message quality --- + + Scenario: Phase violation error message includes type, phase, and allowed types + When I try to record a "tool_invocation" decision in the "strategize" phase + Then a phase violation error should be raised + And the phase violation error message should contain "tool_invocation" + And the phase violation error message should contain "strategize" + And the phase violation error message should contain "Allowed types" + + # --- Phase-gating via database lookup --- + + Scenario: Phase resolved from database when plan_phase not explicitly provided + Given a phase-gated decision service with a persisted strategize plan + When I record a "strategy_choice" decision without explicit phase + Then the phase-gated decision should be recorded successfully + When I try to record a "tool_invocation" decision without explicit phase + Then a phase violation error should be raised + + Scenario: Phase resolved from database for execute plan + Given a phase-gated decision service with a persisted execute plan + When I record a "tool_invocation" decision without explicit phase + Then the phase-gated decision should be recorded successfully + When I try to record a "strategy_choice" decision without explicit phase + Then a phase violation error should be raised + + Scenario: Phase-gating skipped when plan not found in database + Given a phase-gated decision service with an empty database + When I record a "tool_invocation" decision without explicit phase + Then the phase-gated decision should be recorded successfully + + # --- No phase-gating when phase is unknown --- + + Scenario: Phase-gating is skipped when phase cannot be resolved + Given a phase-gated decision service without persistence + When I record a "tool_invocation" decision without explicit phase + Then the phase-gated decision should be recorded successfully + + # --- PHASE_ALLOWED_TYPES constant --- + + Scenario: PHASE_ALLOWED_TYPES maps strategize and execute correctly + Then the PHASE_ALLOWED_TYPES should map "strategize" to STRATEGIZE_TYPES + And the PHASE_ALLOWED_TYPES should map "execute" to EXECUTE_TYPES + + # --- DecisionPhaseViolationError attributes --- + + Scenario: DecisionPhaseViolationError exposes correct attributes + Given a DecisionPhaseViolationError for type "tool_invocation" in phase "strategize" + Then the phase violation error decision_type should be "tool_invocation" + And the phase violation error plan_phase should be "strategize" + And the phase violation error allowed_types should not be empty + + # --- _validate_phase_gating static method --- + + Scenario: _validate_phase_gating does not raise for ungated phases + When I call _validate_phase_gating with type "tool_invocation" and phase "action" + Then no phase violation error should be raised + When I call _validate_phase_gating with type "tool_invocation" and phase "apply" + Then no phase violation error should be raised + + # --- String coercion of plan_phase --- + + Scenario: plan_phase accepts a string value + When I record a "strategy_choice" decision with string phase "strategize" + Then the phase-gated decision should be recorded successfully + + Scenario: plan_phase accepts a PlanPhase enum directly + When I record a "strategy_choice" decision with PlanPhase enum "strategize" + Then the phase-gated decision should be recorded successfully + + # --- Invalid plan_phase string --- + + Scenario: Invalid plan_phase string raises ValidationError + When I try to record a "strategy_choice" decision with invalid phase "not_a_real_phase" + Then a validation error should be raised for invalid phase + And the validation error message should contain "not_a_real_phase" + + # --- Database error resilience --- + + Scenario: resolve_plan_phase gracefully handles database errors + Given a corrupted database unit of work for phase resolution + When I call resolve_plan_phase with the corrupted unit of work + Then resolve_plan_phase should return None diff --git a/features/steps/decision_model_steps.py b/features/steps/decision_model_steps.py index ebc5087a4..cbe002bc3 100644 --- a/features/steps/decision_model_steps.py +++ b/features/steps/decision_model_steps.py @@ -191,6 +191,18 @@ def step_create_user_intervention(context: Context, seq: int) -> None: ) +@when('I create a "{dtype}" decision with sequence {seq:d} and a parent') +def step_create_typed_decision_with_parent( + context: Context, dtype: str, seq: int +) -> None: + context.decision_result = _make_decision( + plan_id=context.decision_plan_id, + parent_decision_id=context.decision_parent_id, + sequence_number=seq, + decision_type=DecisionType(dtype), + ) + + @when("I create a decision with confidence score {score}") def step_create_with_confidence(context: Context, score: str) -> None: conf = None if score == "None" else float(score) @@ -466,6 +478,11 @@ def step_is_any_phase(context: Context) -> None: assert context.decision_result.is_any_phase_type +@then("the decision is_any_phase_type should be false") +def step_is_not_any_phase(context: Context) -> None: + assert not context.decision_result.is_any_phase_type + + @then("the decision confidence score should be {score}") def step_confidence_value(context: Context, score: str) -> None: if score == "None": diff --git a/features/steps/decision_phase_gating_steps.py b/features/steps/decision_phase_gating_steps.py new file mode 100644 index 000000000..221c8dd5a --- /dev/null +++ b/features/steps/decision_phase_gating_steps.py @@ -0,0 +1,444 @@ +"""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}" + ) diff --git a/features/steps/subplan_execution_steps.py b/features/steps/subplan_execution_steps.py index 5f65e7448..8dc64d468 100644 --- a/features/steps/subplan_execution_steps.py +++ b/features/steps/subplan_execution_steps.py @@ -976,8 +976,10 @@ def step_parent_dep_concurrent(context: Context, n: int) -> None: context.dependency_graph = {} context.concurrency_counter = {"current": 0, "max": 0} context.concurrency_lock = threading.Lock() - # Give A and B enough time to overlap - context.delay_map = {_S1: 0.15, _S2: 0.15, _S3: 0.05} + # Give A and B enough time to overlap — use generous delays so that + # concurrency is detected even under heavy CPU contention from the + # parallel test runner. + context.delay_map = {_S1: 0.5, _S2: 0.5, _S3: 0.05} @given("subplans A and B are independent while C depends on both") diff --git a/robot/decision_phase_gating.robot b/robot/decision_phase_gating.robot new file mode 100644 index 000000000..3707225a2 --- /dev/null +++ b/robot/decision_phase_gating.robot @@ -0,0 +1,57 @@ +*** Settings *** +Documentation Smoke tests for decision type phase-gating at recording time +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER_SCRIPT} robot/helper_decision_phase_gating.py + +*** Test Cases *** +Valid Strategize Types Accepted + [Documentation] Record valid strategize-phase decision types + [Tags] service decision phase-gating + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} valid-strategize cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} valid-strategize-ok + Should Be Empty ${result.stderr} + +Valid Execute Types Accepted + [Documentation] Record valid execute-phase decision types + [Tags] service decision phase-gating + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} valid-execute cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} valid-execute-ok + Should Be Empty ${result.stderr} + +Invalid Strategize Types Rejected + [Documentation] Reject execute-only types during strategize phase + [Tags] service decision phase-gating + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} invalid-strategize cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} invalid-strategize-ok + Should Be Empty ${result.stderr} + +Invalid Execute Types Rejected + [Documentation] Reject strategize-only types during execute phase + [Tags] service decision phase-gating + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} invalid-execute cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} invalid-execute-ok + Should Be Empty ${result.stderr} + +Phase Agnostic Types Both Phases + [Documentation] Phase-agnostic types accepted in both phases + [Tags] service decision phase-gating + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} phase-agnostic cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} phase-agnostic-ok + Should Be Empty ${result.stderr} + +Error Attributes Correct + [Documentation] DecisionPhaseViolationError has correct attributes + [Tags] service decision phase-gating + ${result}= Run Process ${PYTHON} ${HELPER_SCRIPT} error-attributes cwd=${WORKSPACE} timeout=120s on_timeout=kill + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} error-attributes-ok + Should Be Empty ${result.stderr} diff --git a/robot/helper_decision_phase_gating.py b/robot/helper_decision_phase_gating.py new file mode 100644 index 000000000..7b689cc16 --- /dev/null +++ b/robot/helper_decision_phase_gating.py @@ -0,0 +1,156 @@ +"""Helper script for Robot Framework decision phase-gating smoke tests. + +Usage: + python robot/helper_decision_phase_gating.py + +Subcommands: + valid-strategize Record valid strategize-phase types + valid-execute Record valid execute-phase types + invalid-strategize Reject execute-only types in strategize + invalid-execute Reject strategize-only types in execute + phase-agnostic Phase-agnostic types work in both phases + error-attributes DecisionPhaseViolationError attributes +""" + +from __future__ import annotations + +import sys +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 cleveragents.application.services.decision_service import DecisionService +from cleveragents.core.exceptions import DecisionPhaseViolationError +from cleveragents.domain.models.core.decision import ( + EXECUTE_TYPES, + STRATEGIZE_TYPES, + DecisionType, +) + +_PLAN_ID = "01HV000000000000000000PG01" + + +def _record( + svc: DecisionService, + dtype: DecisionType, + phase: str, + *, + parent: str | None = "01HV000000000000000000PP01", +) -> None: + """Record a decision with phase-gating.""" + if dtype == DecisionType.PROMPT_DEFINITION: + parent = None + svc.record_decision( + plan_id=_PLAN_ID, + decision_type=dtype, + question=f"Q ({dtype})", + chosen_option=f"A ({dtype})", + parent_decision_id=parent, + plan_phase=phase, + ) + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def _valid_strategize() -> None: + svc = DecisionService() + for dt in STRATEGIZE_TYPES: + _record(svc, dt, "strategize") + print("valid-strategize-ok") + + +def _valid_execute() -> None: + svc = DecisionService() + for dt in EXECUTE_TYPES: + _record(svc, dt, "execute") + print("valid-execute-ok") + + +def _invalid_strategize() -> None: + svc = DecisionService() + execute_only = EXECUTE_TYPES - STRATEGIZE_TYPES + for dt in execute_only: + try: + _record(svc, dt, "strategize") + raise AssertionError( + f"Expected DecisionPhaseViolationError for {dt} in strategize" + ) + except DecisionPhaseViolationError as exc: + assert exc.decision_type == str(dt) + assert exc.plan_phase == "strategize" + print("invalid-strategize-ok") + + +def _invalid_execute() -> None: + svc = DecisionService() + strategize_only = STRATEGIZE_TYPES - EXECUTE_TYPES + for dt in strategize_only: + try: + _record(svc, dt, "execute") + raise AssertionError( + f"Expected DecisionPhaseViolationError for {dt} in execute" + ) + except DecisionPhaseViolationError as exc: + assert exc.decision_type == str(dt) + assert exc.plan_phase == "execute" + print("invalid-execute-ok") + + +def _phase_agnostic() -> None: + svc = DecisionService() + agnostic = STRATEGIZE_TYPES & EXECUTE_TYPES + assert len(agnostic) > 0, "No phase-agnostic types found" + for dt in agnostic: + _record(svc, dt, "strategize") + _record(svc, dt, "execute") + print("phase-agnostic-ok") + + +def _error_attributes() -> None: + svc = DecisionService() + try: + _record(svc, DecisionType.TOOL_INVOCATION, "strategize") + raise AssertionError("Expected DecisionPhaseViolationError") + except DecisionPhaseViolationError as exc: + assert exc.decision_type == "tool_invocation" + assert exc.plan_phase == "strategize" + assert isinstance(exc.allowed_types, frozenset) + assert len(exc.allowed_types) > 0 + assert "tool_invocation" not in exc.allowed_types + msg = str(exc) + assert "tool_invocation" in msg + assert "strategize" in msg + assert "Allowed types" in msg + print("error-attributes-ok") + + +# --------------------------------------------------------------------------- +# Dispatcher +# --------------------------------------------------------------------------- + +_COMMANDS = { + "valid-strategize": _valid_strategize, + "valid-execute": _valid_execute, + "invalid-strategize": _invalid_strategize, + "invalid-execute": _invalid_execute, + "phase-agnostic": _phase_agnostic, + "error-attributes": _error_attributes, +} + + +def main() -> None: + if len(sys.argv) < 2: + raise SystemExit(f"Usage: {sys.argv[0]} <{'|'.join(_COMMANDS)}>") + command = sys.argv[1] + handler = _COMMANDS.get(command) + if handler is None: + raise SystemExit(f"Unknown command: {command}") + handler() + + +if __name__ == "__main__": + main() diff --git a/src/cleveragents/application/services/decision_service.py b/src/cleveragents/application/services/decision_service.py index 0ba5e9ee6..632ff8fe2 100644 --- a/src/cleveragents/application/services/decision_service.py +++ b/src/cleveragents/application/services/decision_service.py @@ -30,6 +30,11 @@ from typing import TYPE_CHECKING import structlog +from cleveragents.application.services.phase_gating import ( + PHASE_ALLOWED_TYPES, + resolve_plan_phase, + validate_phase_gating, +) from cleveragents.core.exceptions import ( BusinessRuleViolation, ResourceNotFoundError, @@ -41,6 +46,7 @@ from cleveragents.domain.models.core.decision import ( Decision, DecisionType, ) +from cleveragents.domain.models.core.plan import PlanPhase from cleveragents.infrastructure.events.models import DomainEvent from cleveragents.infrastructure.events.types import EventType @@ -60,7 +66,6 @@ logger = structlog.get_logger(__name__) #: enough for typical chain-of-thought traces. _MAX_ACTOR_REASONING: int = 100_000 - # --------------------------------------------------------------------------- # Custom exceptions # --------------------------------------------------------------------------- @@ -269,12 +274,22 @@ class DecisionService: corrects_decision_id: str | None = None, correction_reason: str | None = None, dependency_decision_ids: list[str] | None = None, + plan_phase: PlanPhase | str | None = None, ) -> Decision: """Record a new decision in the plan's decision tree. Automatically assigns a monotonically increasing sequence number, generates a decision ID (ULID), and captures a context snapshot. + Phase-gating + ~~~~~~~~~~~~~ + When *plan_phase* is supplied (or can be resolved from the + database via the :class:`UnitOfWork`), the *decision_type* is + validated against :data:`PHASE_ALLOWED_TYPES`. If the type is + not permitted in the current phase, a + :class:`DecisionPhaseViolationError` is raised **before** the + decision is persisted. + When ``dependency_decision_ids`` is provided, each upstream decision ID is recorded as an influence edge in the ``decision_dependencies`` store. This populates the influence @@ -300,6 +315,12 @@ class DecisionService: dependency_decision_ids: ULIDs of upstream decisions that influenced or constrained this decision. Recorded in the influence DAG (``decision_dependencies``). + plan_phase: Current phase of the plan. When provided, the + decision type is validated against the phase-allowed + set. When ``None`` and a :class:`UnitOfWork` is wired, + the phase is resolved from the database. When ``None`` + and no persistence layer is available, phase-gating is + skipped for backward compatibility. Returns: The recorded :class:`Decision`. @@ -307,6 +328,8 @@ class DecisionService: Raises: ValidationError: If required fields are missing or invalid. DuplicateDecisionError: If a decision with the same ID exists. + DecisionPhaseViolationError: If *decision_type* is not + permitted in the plan's current phase. """ if not plan_id or not plan_id.strip(): raise ValidationError("plan_id must not be empty") @@ -323,6 +346,17 @@ class DecisionService: if isinstance(decision_type, str): decision_type = DecisionType(decision_type) + # --- Phase-gating validation --- + resolved_phase = resolve_plan_phase( + plan_id, + plan_phase, + persisted=self._persisted, + unit_of_work=self.unit_of_work, + logger=self._logger, + ) + if resolved_phase is not None: + validate_phase_gating(decision_type, resolved_phase) + # Auto-assign sequence number seq = self._next_sequence(plan_id) @@ -869,6 +903,7 @@ class DecisionService: __all__ = [ + "PHASE_ALLOWED_TYPES", "DecisionNotFoundError", "DecisionService", "DuplicateDecisionError", diff --git a/src/cleveragents/application/services/phase_gating.py b/src/cleveragents/application/services/phase_gating.py new file mode 100644 index 000000000..41408d7b9 --- /dev/null +++ b/src/cleveragents/application/services/phase_gating.py @@ -0,0 +1,148 @@ +"""Phase-gating validation for decision recording. + +Ensures that decision types are only recorded during the plan phases +where they are permitted. Extracted from ``DecisionService`` to keep +the service module focused on recording and snapshot concerns. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from typing import TYPE_CHECKING + +import structlog +from sqlalchemy.exc import OperationalError + +from cleveragents.core.exceptions import ( + DatabaseError, + DecisionPhaseViolationError, + ValidationError, +) +from cleveragents.domain.models.core.decision import ( + EXECUTE_TYPES, + STRATEGIZE_TYPES, + DecisionType, +) +from cleveragents.domain.models.core.plan import PlanPhase + +if TYPE_CHECKING: + from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + +#: Mapping from plan phase to allowed decision types. +#: +#: Only ``STRATEGIZE`` and ``EXECUTE`` are gated; the ``ACTION`` and +#: ``APPLY`` phases do not record decisions. When the plan phase is +#: not in this mapping, phase-gating is skipped. +PHASE_ALLOWED_TYPES: Mapping[PlanPhase, frozenset[DecisionType]] = { + PlanPhase.STRATEGIZE: STRATEGIZE_TYPES, + PlanPhase.EXECUTE: EXECUTE_TYPES, +} + + +def resolve_plan_phase( + plan_id: str, + explicit_phase: PlanPhase | str | None, + *, + persisted: bool, + unit_of_work: UnitOfWork | None, + logger: structlog.stdlib.BoundLogger, +) -> PlanPhase | None: + """Resolve the plan phase for phase-gating validation. + + Resolution order: + 1. *explicit_phase* — the caller already knows the phase. + 2. Database lookup via *unit_of_work* → ``lifecycle_plans``. + 3. ``None`` — phase-gating is skipped. + + Args: + plan_id: ULID of the plan. + explicit_phase: Caller-supplied phase, if any. + persisted: Whether the service is in persisted mode. + unit_of_work: Optional database access layer. + logger: Logger for warning on DB failures. + + Returns: + The resolved :class:`PlanPhase`, or ``None`` when the phase + cannot be determined. + """ + if explicit_phase is not None: + if isinstance(explicit_phase, str): + try: + return PlanPhase(explicit_phase) + except ValueError: + raise ValidationError( + f"Invalid plan_phase: '{explicit_phase}'" + ) from None + return explicit_phase + + if persisted and unit_of_work is not None: + # TODO(pg-migration): TOCTOU — The phase is read here but the + # decision is written later in _store_decision(), in a separate + # transaction. This is acceptable for SQLite (single-writer + # serialisation) but must be revisited if the persistence layer + # is migrated to PostgreSQL or another multi-writer database. + # At minimum, a single-writer assertion or advisory lock should + # guard this section under multi-writer engines. + # + # NOTE: Uncached — This opens a DB transaction on every + # record_decision() call when plan_phase is None. A + # per-plan phase cache could avoid this, but the cost is + # negligible for SQLite and correctness is more important + # while plan phases can change mid-lifecycle. + try: + with unit_of_work.transaction() as ctx: + plan = ctx.lifecycle_plans.get(plan_id) + if plan is not None: + return ( + PlanPhase(plan.phase) + if isinstance(plan.phase, str) + else plan.phase + ) + except (DatabaseError, OperationalError, OSError): + # Infrastructure errors must not break the opt-in + # phase-gating contract. If the DB is temporarily + # unavailable we fall through and skip gating rather + # than propagating an infrastructure error to the caller. + logger.warning( + "decision.phase_resolve_db_error", + plan_id=plan_id, + exc_info=True, + ) + + return None + + +def validate_phase_gating( + decision_type: DecisionType, + phase: PlanPhase, +) -> None: + """Raise if *decision_type* is not permitted in *phase*. + + Only ``STRATEGIZE`` and ``EXECUTE`` are gated. Other phases + (``ACTION``, ``APPLY``) do not restrict decision types. + + Args: + decision_type: The decision type to validate. + phase: The plan's current phase. + + Raises: + DecisionPhaseViolationError: If the type is not in the + allowed set for the given phase. + """ + allowed = PHASE_ALLOWED_TYPES.get(phase) + if allowed is None: + # Phase is not gated (e.g. ACTION, APPLY). + return + if decision_type not in allowed: + raise DecisionPhaseViolationError( + decision_type=str(decision_type), + plan_phase=str(phase), + allowed_types=frozenset(str(t) for t in allowed), + ) + + +__all__ = [ + "PHASE_ALLOWED_TYPES", + "resolve_plan_phase", + "validate_phase_gating", +] diff --git a/src/cleveragents/core/exceptions.py b/src/cleveragents/core/exceptions.py index 131260444..4ca01d81f 100644 --- a/src/cleveragents/core/exceptions.py +++ b/src/cleveragents/core/exceptions.py @@ -254,6 +254,32 @@ class PlanError(DomainError): pass +class DecisionPhaseViolationError(BusinessRuleViolation): + """Raised when a decision type is invalid for the plan's current phase. + + Attributes: + decision_type: The :class:`str` value of the attempted decision type. + plan_phase: The :class:`str` value of the plan's current phase. + allowed_types: Frozenset of :class:`str` values of decision types + permitted in the current phase. + """ + + def __init__( + self, + decision_type: str, + plan_phase: str, + allowed_types: frozenset[str], + ) -> None: + sorted_allowed = sorted(allowed_types) + super().__init__( + f"Decision type '{decision_type}' is not permitted in the " + f"'{plan_phase}' phase. Allowed types: {sorted_allowed}" + ) + self.decision_type = decision_type + self.plan_phase = plan_phase + self.allowed_types = allowed_types + + class ExecutionError(CleverAgentsError): """Execution failures for agents or graph nodes.""" @@ -267,6 +293,7 @@ __all__ = [ "CleverAgentsError", "ConfigurationError", "DatabaseError", + "DecisionPhaseViolationError", "DomainError", "ExecutionError", "ExternalServiceError", diff --git a/src/cleveragents/domain/models/core/decision.py b/src/cleveragents/domain/models/core/decision.py index 41c3edf6b..0f33be508 100644 --- a/src/cleveragents/domain/models/core/decision.py +++ b/src/cleveragents/domain/models/core/decision.py @@ -28,13 +28,13 @@ Decision types - Execute - How to implement a specific task * - ``resource_selection`` - - Execute + - Strategize or Execute - Which resources to read / modify * - ``subplan_spawn`` - - Strategize + - Strategize or Execute - Decision to create a child plan * - ``subplan_parallel_spawn`` - - Strategize + - Strategize or Execute - Spawn a group of child plans in parallel * - ``tool_invocation`` - Execute @@ -107,18 +107,35 @@ class DecisionType(StrEnum): USER_INTERVENTION = "user_intervention" -#: Decision types that may only be created during the Strategize phase. +#: Decision types permitted during the Strategize phase. +#: +#: Phase-agnostic types (``resource_selection``, ``subplan_spawn``, +#: ``subplan_parallel_spawn``, ``user_intervention``) appear in **both** +#: sets so that the phase-gating check in +#: :meth:`DecisionService.record_decision` accepts them in either +#: Strategize or Execute. +#: +#: .. note:: +#: +#: ``subplan_spawn`` and ``subplan_parallel_spawn`` are classified as +#: Strategize-only in ADR-007 and ADR-033. They are included in +#: ``EXECUTE_TYPES`` here because M4's subplan model requires +#: recording spawn decisions during Execute when the execution actor +#: discovers additional decomposition needs at runtime. See ticket +#: #931 for context. STRATEGIZE_TYPES: frozenset[DecisionType] = frozenset( { DecisionType.PROMPT_DEFINITION, DecisionType.INVARIANT_ENFORCED, DecisionType.STRATEGY_CHOICE, + DecisionType.RESOURCE_SELECTION, DecisionType.SUBPLAN_SPAWN, DecisionType.SUBPLAN_PARALLEL_SPAWN, + DecisionType.USER_INTERVENTION, } ) -#: Decision types that may only be created during the Execute phase. +#: Decision types permitted during the Execute phase. EXECUTE_TYPES: frozenset[DecisionType] = frozenset( { DecisionType.IMPLEMENTATION_CHOICE, @@ -126,6 +143,9 @@ EXECUTE_TYPES: frozenset[DecisionType] = frozenset( DecisionType.TOOL_INVOCATION, DecisionType.ERROR_RECOVERY, DecisionType.VALIDATION_RESPONSE, + DecisionType.SUBPLAN_SPAWN, + DecisionType.SUBPLAN_PARALLEL_SPAWN, + DecisionType.USER_INTERVENTION, } ) @@ -446,7 +466,10 @@ class Decision(BaseModel): @property def is_any_phase_type(self) -> bool: """Return True if this decision type is valid in any phase.""" - return self.decision_type == DecisionType.USER_INTERVENTION + return ( + self.decision_type in STRATEGIZE_TYPES + and self.decision_type in EXECUTE_TYPES + ) def as_cli_dict(self) -> dict[str, object]: """Return a stable dict for CLI table rendering.""" -- 2.52.0