diff --git a/benchmarks/apply_pipeline_bench.py b/benchmarks/apply_pipeline_bench.py new file mode 100644 index 000000000..5920ef059 --- /dev/null +++ b/benchmarks/apply_pipeline_bench.py @@ -0,0 +1,169 @@ +"""ASV benchmarks for the validation-gated apply pipeline. + +Measures the performance of: +- ApplyResult model construction +- Validation count extraction +- apply_with_validation_gate() with mock lifecycle +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +try: + from cleveragents.application.services.plan_apply_service import ( + ApplyOutcome, + ApplyResult, + PlanApplyService, + ) + from cleveragents.domain.models.core.change import ( + ChangeEntry, + ChangeOperation, + SpecChangeSet, + ) + from cleveragents.domain.models.core.plan import ( + PlanPhase, + PlanTimestamps, + ProcessingState, + ) +except ModuleNotFoundError: + sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) + from cleveragents.application.services.plan_apply_service import ( + ApplyOutcome, + ApplyResult, + PlanApplyService, + ) + from cleveragents.domain.models.core.change import ( + ChangeEntry, + ChangeOperation, + SpecChangeSet, + ) + from cleveragents.domain.models.core.plan import ( + PlanPhase, + PlanTimestamps, + ProcessingState, + ) + + +_PLAN_ID = "01BENCHPLAN0000000000001" + + +def _make_plan( + *, + changeset_id: str = "CS001", + validation_summary: dict[str, Any] | None = None, +) -> MagicMock: + plan = MagicMock() + plan.identity.plan_id = _PLAN_ID + plan.phase = PlanPhase.EXECUTE + plan.processing_state = ProcessingState.COMPLETE + plan.changeset_id = changeset_id + plan.validation_summary = validation_summary + plan.is_terminal = False + plan.error_details = None + plan.timestamps = PlanTimestamps() + return plan + + +def _make_changeset(n: int = 10) -> SpecChangeSet: + cs = SpecChangeSet(plan_id=_PLAN_ID) + for i in range(n): + cs.add_change( + ChangeEntry( + plan_id=_PLAN_ID, + resource_id="RES001", + tool_name="builtin/bench-tool", + operation=ChangeOperation.MODIFY, + path=f"src/mod_{i:04d}.py", + before_hash=f"b{i}", + after_hash=f"a{i}", + ) + ) + return cs + + +def _make_service(plan: MagicMock, changeset: SpecChangeSet) -> PlanApplyService: + lifecycle = MagicMock() + lifecycle.get_plan.return_value = plan + lifecycle.complete_apply.return_value = plan + lifecycle._commit_plan = MagicMock() + store = MagicMock() + store.get.return_value = changeset + return PlanApplyService(lifecycle_service=lifecycle, changeset_store=store) + + +class ApplyResultSuite: + """Benchmark ApplyResult model construction.""" + + def time_result_construction_applied(self) -> None: + ApplyResult( + outcome=ApplyOutcome.APPLIED, + plan_id=_PLAN_ID, + message="applied ok", + files_changed=10, + validations_total=5, + validations_passed=5, + validations_failed=0, + ) + + def time_result_construction_constrained(self) -> None: + ApplyResult( + outcome=ApplyOutcome.CONSTRAINED, + plan_id=_PLAN_ID, + message="blocked", + validations_total=5, + validations_passed=3, + validations_failed=2, + ) + + +class ApplyValidationCountSuite: + """Benchmark validation count extraction.""" + + def setup(self) -> None: + self.summary: dict[str, Any] = { + "total": 100, + "required_passed": 80, + "required_failed": 20, + } + self.empty_summary: dict[str, Any] | None = None + + def time_extract_counts(self) -> None: + PlanApplyService._extract_validation_counts(self.summary) + + def time_extract_counts_none(self) -> None: + PlanApplyService._extract_validation_counts(self.empty_summary) + + +class ApplyGateSuite: + """Benchmark the full apply_with_validation_gate flow.""" + + def setup(self) -> None: + cs = _make_changeset(20) + self.plan_pass = _make_plan( + changeset_id=cs.changeset_id, + validation_summary={ + "total": 5, + "required_passed": 5, + "required_failed": 0, + }, + ) + self.plan_fail = _make_plan( + changeset_id=cs.changeset_id, + validation_summary={ + "total": 5, + "required_passed": 3, + "required_failed": 2, + }, + ) + self.svc_pass = _make_service(self.plan_pass, cs) + self.svc_fail = _make_service(self.plan_fail, cs) + + def time_apply_gate_pass(self) -> None: + self.svc_pass.apply_with_validation_gate(_PLAN_ID) + + def time_apply_gate_fail(self) -> None: + self.svc_fail.apply_with_validation_gate(_PLAN_ID) diff --git a/docs/reference/plan_apply.md b/docs/reference/plan_apply.md index e8c3090b2..d50cf104c 100644 --- a/docs/reference/plan_apply.md +++ b/docs/reference/plan_apply.md @@ -117,6 +117,61 @@ Apply/QUEUED --> Apply/PROCESSING --> Apply/APPLIED (terminal success) +--> Apply/CONSTRAINED (invariant violation) ``` +### Validation-Gated Apply + +The apply pipeline gates commits on validation results. Before any +file changes are committed, `apply_with_validation_gate()` checks +the validation summary (either from the plan or supplied externally) +and blocks apply if any **required** validations have failed. + +#### Outcomes + +| Outcome | Description | +|---------|-------------| +| `applied` | All required validations passed; changes committed | +| `constrained` | Required validations failed; apply blocked | +| `already_applied` | Plan was already in a terminal applied state | +| `blocked_empty` | ChangeSet was empty and `allow_empty` not set | + +#### Flow + +``` +apply_with_validation_gate(plan_id) + | + +-- Plan in terminal state? --> already_applied / constrained + | + +-- Empty ChangeSet? --> blocked_empty (unless allow_empty) + | + +-- required_failed > 0? --> constrained (with actionable message) + | + +-- All passed --> persist_apply_summary() --> complete_apply() + --> ApplyResult(outcome="applied") +``` + +#### `ApplyResult` Model + +| Field | Type | Description | +|-------|------|-------------| +| `outcome` | `ApplyOutcome` | One of the outcomes above | +| `plan_id` | `str` | Plan ULID | +| `message` | `str` | Human-readable result message | +| `files_changed` | `int` | Number of files changed during apply | +| `validations_total` | `int` | Total validations evaluated | +| `validations_passed` | `int` | Required validations that passed | +| `validations_failed` | `int` | Required validations that failed | + +#### External Validation Summary + +You can supply a validation summary from an external source (e.g. a +CI pipeline) to override the plan's stored summary: + +```python +result = service.apply_with_validation_gate( + plan_id, + validation_summary={"total": 3, "required_passed": 2, "required_failed": 1}, +) +``` + ## Service API ### `PlanApplyService` @@ -143,4 +198,13 @@ service.persist_apply_summary(plan_id, files_changed=5, validations_run=3) # Handle merge failure service.handle_merge_failure(plan_id, conflict_details="...") + +# Validation-gated apply +from cleveragents.application.services.plan_apply_service import ApplyOutcome, ApplyResult + +result = service.apply_with_validation_gate(plan_id, allow_empty=False) +if result.outcome == ApplyOutcome.APPLIED: + print(f"Applied: {result.files_changed} files changed") +elif result.outcome == ApplyOutcome.CONSTRAINED: + print(f"Blocked: {result.message}") ``` diff --git a/features/apply_pipeline.feature b/features/apply_pipeline.feature new file mode 100644 index 000000000..555c54ffd --- /dev/null +++ b/features/apply_pipeline.feature @@ -0,0 +1,102 @@ +Feature: Validation-gated apply pipeline + As a plan executor + I want apply to be gated by validation results + So that changes are only committed when all required validations pass + + Background: + Given an apply pipeline test environment + + # -- Successful apply -------------------------------------------------- + + Scenario: Apply succeeds when all required validations pass + Given a plan with changeset containing 3 files + And the plan has validation summary with 2 required passed and 0 failed + When I apply the plan with validation gate + Then the apply outcome should be "applied" + And the apply result files changed should be 3 + And the apply result validations total should be 2 + And the apply result message should contain "applied successfully" + + Scenario: Apply succeeds with no validations + Given a plan with changeset containing 1 files + And the plan has no validation summary + When I apply the plan with validation gate + Then the apply outcome should be "applied" + And the apply result validations total should be 0 + + # -- Blocked by validation failure ------------------------------------ + + Scenario: Apply blocked when required validations fail + Given a plan with changeset containing 5 files + And the plan has validation summary with 1 required passed and 2 failed + When I apply the plan with validation gate + Then the apply outcome should be "constrained" + And the apply result validations failed should be 2 + And the apply result message should contain "Apply refused" + And the apply result message should contain "2 required" + + # -- Idempotent re-apply blocking ------------------------------------- + + Scenario: Re-apply on already applied plan returns already_applied + Given a plan that is already applied + When I apply the plan with validation gate + Then the apply outcome should be "already_applied" + And the apply result message should contain "already been applied" + + Scenario: Apply on errored plan returns constrained + Given a plan that is in errored state + When I apply the plan with validation gate + Then the apply outcome should be "constrained" + And the apply result message should contain "terminal state" + + Scenario: Apply on cancelled plan returns constrained + Given a plan that is in cancelled state + When I apply the plan with validation gate + Then the apply outcome should be "constrained" + And the apply result message should contain "terminal state" + + # -- Empty changeset guard -------------------------------------------- + + Scenario: Apply blocked on empty changeset without allow_empty + Given a plan with empty changeset + When I apply the plan with validation gate + Then the apply outcome should be "blocked_empty" + And the apply result message should contain "empty ChangeSet" + + Scenario: Apply blocked when plan has no changeset + Given an apply plan with no changeset assigned + When I apply the plan with validation gate + Then the apply outcome should be "blocked_empty" + And the apply result message should contain "empty ChangeSet" + + Scenario: Apply succeeds on empty changeset with allow_empty + Given a plan with empty changeset + And the plan has no validation summary + When I apply the plan with validation gate and allow_empty + Then the apply outcome should be "applied" + + # -- External validation summary override ----------------------------- + + Scenario: External validation summary overrides plan stored summary + Given a plan with changeset containing 2 files + And the plan has validation summary with 0 required passed and 0 failed + When I apply the plan with external validation summary having 1 passed and 1 failed + Then the apply outcome should be "constrained" + And the apply result validations failed should be 1 + + # -- Validation count extraction -------------------------------------- + + Scenario: Validation counts extracted correctly from summary + Given a plan with changeset containing 1 files + And the plan has validation summary with 3 required passed and 0 failed and 2 informational + When I apply the plan with validation gate + Then the apply outcome should be "applied" + And the apply result validations total should be 5 + + # -- ApplyResult model checks ----------------------------------------- + + Scenario: ApplyResult model has correct fields + Given a plan with changeset containing 1 files + And the plan has no validation summary + When I apply the plan with validation gate + Then the apply result plan_id should match the plan ID diff --git a/features/steps/apply_pipeline_steps.py b/features/steps/apply_pipeline_steps.py new file mode 100644 index 000000000..82f520940 --- /dev/null +++ b/features/steps/apply_pipeline_steps.py @@ -0,0 +1,311 @@ +"""Step definitions for validation-gated apply pipeline tests. + +Uses mock objects for PlanLifecycleService and Plan to test the +validation gate logic in PlanApplyService without needing a database. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from behave import given, then, when +from behave.runner import Context + +from cleveragents.application.services.plan_apply_service import ( + ApplyOutcome, + PlanApplyService, +) +from cleveragents.domain.models.core.change import ( + ChangeEntry, + ChangeOperation, + SpecChangeSet, +) +from cleveragents.domain.models.core.plan import ( + PlanPhase, + PlanTimestamps, + ProcessingState, +) + +__all__: list[str] = [] + + +# --------------------------------------------------------------------------- +# Mock helpers +# --------------------------------------------------------------------------- + +_PLAN_ID = "01PLANTEST000000000000001" + + +def _make_mock_plan( + *, + plan_id: str = _PLAN_ID, + phase: PlanPhase = PlanPhase.EXECUTE, + state: ProcessingState = ProcessingState.COMPLETE, + changeset_id: str | None = "CS001", + validation_summary: dict[str, Any] | None = None, + is_terminal: bool = False, +) -> MagicMock: + """Create a mock Plan object.""" + plan = MagicMock() + plan.identity.plan_id = plan_id + plan.phase = phase + plan.processing_state = state + plan.changeset_id = changeset_id + plan.validation_summary = validation_summary + plan.is_terminal = is_terminal + plan.error_details = None + plan.timestamps = PlanTimestamps() + return plan + + +def _make_changeset( + plan_id: str = _PLAN_ID, + n_files: int = 1, +) -> SpecChangeSet: + """Create a SpecChangeSet with n_files entries.""" + cs = SpecChangeSet(plan_id=plan_id) + for i in range(n_files): + cs.add_change( + ChangeEntry( + plan_id=plan_id, + resource_id="RES001", + tool_name="builtin/test-tool", + operation=ChangeOperation.MODIFY, + path=f"src/file_{i}.py", + before_hash=f"before-{i}", + after_hash=f"after-{i}", + ) + ) + return cs + + +def _make_service( + plan: MagicMock, + changeset: SpecChangeSet | None = None, +) -> PlanApplyService: + """Create a PlanApplyService with mocked lifecycle and store.""" + lifecycle = MagicMock() + lifecycle.get_plan.return_value = plan + lifecycle.complete_apply.return_value = plan + lifecycle.constrain_apply.return_value = plan + lifecycle._commit_plan = MagicMock() + + store = MagicMock() + if changeset is not None: + store.get.return_value = changeset + else: + store.get.return_value = None + + return PlanApplyService( + lifecycle_service=lifecycle, + changeset_store=store, + ) + + +# --------------------------------------------------------------------------- +# Background +# --------------------------------------------------------------------------- + + +@given("an apply pipeline test environment") +def step_given_apply_env(context: Context) -> None: + context.ap_plan_id = _PLAN_ID + context.ap_plan = None + context.ap_changeset = None + context.ap_service = None + context.ap_result = None # ApplyResult | None + context.ap_allow_empty = False + context.ap_external_summary = None # dict[str, Any] | None + + +# --------------------------------------------------------------------------- +# Given steps +# --------------------------------------------------------------------------- + + +@given("a plan with changeset containing {n} files") +def step_given_plan_with_files(context: Context, n: str) -> None: + cs = _make_changeset(n_files=int(n)) + plan = _make_mock_plan(changeset_id=cs.changeset_id) + context.ap_plan = plan + context.ap_changeset = cs + context.ap_service = _make_service(plan, cs) + + +@given("the plan has validation summary with {p} required passed and {f} failed") +def step_given_validation_summary(context: Context, p: str, f: str) -> None: + vs = { + "total": int(p) + int(f), + "required_passed": int(p), + "required_failed": int(f), + "informational_passed": 0, + "informational_failed": 0, + } + context.ap_plan.validation_summary = vs + + +@given( + "the plan has validation summary with {p} required passed " + "and {f} failed and {i} informational" +) +def step_given_validation_summary_with_info( + context: Context, p: str, f: str, i: str +) -> None: + vs = { + "total": int(p) + int(f) + int(i), + "required_passed": int(p), + "required_failed": int(f), + "informational_passed": int(i), + "informational_failed": 0, + } + context.ap_plan.validation_summary = vs + + +@given("the plan has no validation summary") +def step_given_no_validation_summary(context: Context) -> None: + context.ap_plan.validation_summary = None + + +@given("a plan that is already applied") +def step_given_already_applied(context: Context) -> None: + plan = _make_mock_plan( + phase=PlanPhase.APPLY, + state=ProcessingState.APPLIED, + is_terminal=True, + ) + cs = _make_changeset() + context.ap_plan = plan + context.ap_changeset = cs + context.ap_service = _make_service(plan, cs) + + +@given("a plan that is in errored state") +def step_given_errored_plan(context: Context) -> None: + plan = _make_mock_plan( + phase=PlanPhase.APPLY, + state=ProcessingState.ERRORED, + is_terminal=True, + ) + cs = _make_changeset() + context.ap_plan = plan + context.ap_changeset = cs + context.ap_service = _make_service(plan, cs) + + +@given("a plan that is in cancelled state") +def step_given_cancelled_plan(context: Context) -> None: + plan = _make_mock_plan( + phase=PlanPhase.APPLY, + state=ProcessingState.CANCELLED, + is_terminal=True, + ) + cs = _make_changeset() + context.ap_plan = plan + context.ap_changeset = cs + context.ap_service = _make_service(plan, cs) + + +@given("a plan with empty changeset") +def step_given_empty_changeset(context: Context) -> None: + cs = _make_changeset(n_files=0) + plan = _make_mock_plan(changeset_id=cs.changeset_id) + context.ap_plan = plan + context.ap_changeset = cs + context.ap_service = _make_service(plan, cs) + + +@given("an apply plan with no changeset assigned") +def step_given_no_changeset(context: Context) -> None: + plan = _make_mock_plan(changeset_id=None) + context.ap_plan = plan + context.ap_changeset = None + context.ap_service = _make_service(plan, None) + + +# --------------------------------------------------------------------------- +# When steps +# --------------------------------------------------------------------------- + + +@when("I apply the plan with validation gate") +def step_when_apply(context: Context) -> None: + context.ap_result = context.ap_service.apply_with_validation_gate( + plan_id=context.ap_plan_id, + allow_empty=context.ap_allow_empty, + ) + + +@when("I apply the plan with validation gate and allow_empty") +def step_when_apply_allow_empty(context: Context) -> None: + context.ap_result = context.ap_service.apply_with_validation_gate( + plan_id=context.ap_plan_id, + allow_empty=True, + ) + + +@when( + "I apply the plan with external validation summary having {p} passed and {f} failed" +) +def step_when_apply_external_summary(context: Context, p: str, f: str) -> None: + ext_summary = { + "total": int(p) + int(f), + "required_passed": int(p), + "required_failed": int(f), + } + context.ap_result = context.ap_service.apply_with_validation_gate( + plan_id=context.ap_plan_id, + validation_summary=ext_summary, + ) + + +# --------------------------------------------------------------------------- +# Then steps +# --------------------------------------------------------------------------- + + +@then('the apply outcome should be "{outcome}"') +def step_then_outcome(context: Context, outcome: str) -> None: + assert context.ap_result is not None, "No apply result" + expected = ApplyOutcome(outcome) + assert context.ap_result.outcome == expected, ( + f"Expected outcome '{outcome}', got '{context.ap_result.outcome}'" + ) + + +@then("the apply result files changed should be {n}") +def step_then_files_changed(context: Context, n: str) -> None: + assert context.ap_result is not None + assert context.ap_result.files_changed == int(n), ( + f"Expected files_changed={n}, got {context.ap_result.files_changed}" + ) + + +@then("the apply result validations total should be {n}") +def step_then_validations_total(context: Context, n: str) -> None: + assert context.ap_result is not None + assert context.ap_result.validations_total == int(n), ( + f"Expected validations_total={n}, got {context.ap_result.validations_total}" + ) + + +@then("the apply result validations failed should be {n}") +def step_then_validations_failed(context: Context, n: str) -> None: + assert context.ap_result is not None + assert context.ap_result.validations_failed == int(n), ( + f"Expected validations_failed={n}, got {context.ap_result.validations_failed}" + ) + + +@then('the apply result message should contain "{text}"') +def step_then_message_contains(context: Context, text: str) -> None: + assert context.ap_result is not None + assert text in context.ap_result.message, ( + f"Expected message to contain '{text}', got: {context.ap_result.message}" + ) + + +@then("the apply result plan_id should match the plan ID") +def step_then_plan_id_matches(context: Context) -> None: + assert context.ap_result is not None + assert context.ap_result.plan_id == context.ap_plan_id diff --git a/robot/apply_pipeline.robot b/robot/apply_pipeline.robot new file mode 100644 index 000000000..12ce2f40c --- /dev/null +++ b/robot/apply_pipeline.robot @@ -0,0 +1,51 @@ +*** Settings *** +Documentation Smoke tests for the validation-gated apply pipeline, +... including pass/fail gating, idempotent re-apply, and +... empty changeset guard. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_apply_pipeline.py + +*** Test Cases *** +Apply Passes When All Validations Pass + [Documentation] Verify apply succeeds when required validations all pass + ${result}= Run Process ${PYTHON} ${HELPER} apply-passes cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} apply-passes-ok + +Apply Blocked When Required Validations Fail + [Documentation] Verify apply is blocked with constrained outcome on validation failure + ${result}= Run Process ${PYTHON} ${HELPER} apply-blocked cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} apply-blocked-ok + +Re-Apply On Already Applied Plan Returns Already Applied + [Documentation] Verify idempotent re-apply blocking on terminal applied state + ${result}= Run Process ${PYTHON} ${HELPER} already-applied cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} already-applied-ok + +Apply Blocked On Empty Changeset + [Documentation] Verify apply is blocked when changeset has no entries + ${result}= Run Process ${PYTHON} ${HELPER} empty-changeset cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} empty-changeset-ok + +Apply Pipeline Model Checks + [Documentation] Verify ApplyOutcome enum and ApplyResult model construction + ${result}= Run Process ${PYTHON} ${HELPER} model-checks cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} model-checks-ok diff --git a/robot/helper_apply_pipeline.py b/robot/helper_apply_pipeline.py new file mode 100644 index 000000000..66c1233ee --- /dev/null +++ b/robot/helper_apply_pipeline.py @@ -0,0 +1,215 @@ +"""Robot Framework helper for apply pipeline smoke tests. + +Provides a CLI-style interface for Robot to invoke validation-gated +apply operations. Exit code 0 = success, 1 = failure. + +Usage: + python robot/helper_apply_pipeline.py apply-passes + python robot/helper_apply_pipeline.py apply-blocked + python robot/helper_apply_pipeline.py already-applied + python robot/helper_apply_pipeline.py empty-changeset + python robot/helper_apply_pipeline.py model-checks +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from typing import Any +from unittest.mock import MagicMock + +# Ensure the src directory is on the import path. +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from cleveragents.application.services.plan_apply_service import ( # noqa: E402 + ApplyOutcome, + ApplyResult, + PlanApplyService, +) +from cleveragents.domain.models.core.change import ( # noqa: E402 + ChangeEntry, + ChangeOperation, + SpecChangeSet, +) +from cleveragents.domain.models.core.plan import ( # noqa: E402 + PlanPhase, + PlanTimestamps, + ProcessingState, +) + +_PLAN_ID = "01PLANTEST000000000000001" + + +def _make_plan( + *, + phase: PlanPhase = PlanPhase.EXECUTE, + state: ProcessingState = ProcessingState.COMPLETE, + changeset_id: str | None = "CS001", + validation_summary: dict[str, Any] | None = None, + is_terminal: bool = False, +) -> MagicMock: + plan = MagicMock() + plan.identity.plan_id = _PLAN_ID + plan.phase = phase + plan.processing_state = state + plan.changeset_id = changeset_id + plan.validation_summary = validation_summary + plan.is_terminal = is_terminal + plan.error_details = None + plan.timestamps = PlanTimestamps() + return plan + + +def _make_changeset(n: int = 1) -> SpecChangeSet: + cs = SpecChangeSet(plan_id=_PLAN_ID) + for i in range(n): + cs.add_change( + ChangeEntry( + plan_id=_PLAN_ID, + resource_id="RES001", + tool_name="builtin/test-tool", + operation=ChangeOperation.MODIFY, + path=f"src/file_{i}.py", + before_hash=f"b{i}", + after_hash=f"a{i}", + ) + ) + return cs + + +def _make_service( + plan: MagicMock, changeset: SpecChangeSet | None = None +) -> PlanApplyService: + lifecycle = MagicMock() + lifecycle.get_plan.return_value = plan + lifecycle.complete_apply.return_value = plan + lifecycle.constrain_apply.return_value = plan + lifecycle._commit_plan = MagicMock() + store = MagicMock() + store.get.return_value = changeset + return PlanApplyService(lifecycle_service=lifecycle, changeset_store=store) + + +def _cmd_apply_passes() -> int: + """Apply with all validations passing.""" + cs = _make_changeset(3) + plan = _make_plan( + changeset_id=cs.changeset_id, + validation_summary={ + "total": 2, + "required_passed": 2, + "required_failed": 0, + }, + ) + svc = _make_service(plan, cs) + result = svc.apply_with_validation_gate(_PLAN_ID) + assert result.outcome == ApplyOutcome.APPLIED, f"Got {result.outcome}" + assert result.files_changed == 3 + assert "applied successfully" in result.message + print("apply-passes-ok") + return 0 + + +def _cmd_apply_blocked() -> int: + """Apply blocked by validation failure.""" + cs = _make_changeset(2) + plan = _make_plan( + changeset_id=cs.changeset_id, + validation_summary={ + "total": 3, + "required_passed": 1, + "required_failed": 2, + }, + ) + svc = _make_service(plan, cs) + result = svc.apply_with_validation_gate(_PLAN_ID) + assert result.outcome == ApplyOutcome.CONSTRAINED, f"Got {result.outcome}" + assert result.validations_failed == 2 + assert "Apply refused" in result.message + print("apply-blocked-ok") + return 0 + + +def _cmd_already_applied() -> int: + """Re-apply on already applied plan.""" + cs = _make_changeset(1) + plan = _make_plan( + phase=PlanPhase.APPLY, + state=ProcessingState.APPLIED, + is_terminal=True, + ) + svc = _make_service(plan, cs) + result = svc.apply_with_validation_gate(_PLAN_ID) + assert result.outcome == ApplyOutcome.ALREADY_APPLIED, f"Got {result.outcome}" + assert "already been applied" in result.message + print("already-applied-ok") + return 0 + + +def _cmd_empty_changeset() -> int: + """Apply blocked on empty changeset.""" + cs = _make_changeset(0) + plan = _make_plan(changeset_id=cs.changeset_id) + svc = _make_service(plan, cs) + result = svc.apply_with_validation_gate(_PLAN_ID) + assert result.outcome == ApplyOutcome.BLOCKED_EMPTY, f"Got {result.outcome}" + assert "empty ChangeSet" in result.message + print("empty-changeset-ok") + return 0 + + +def _cmd_model_checks() -> int: + """Verify ApplyOutcome and ApplyResult models.""" + # Check all enum values exist + assert ApplyOutcome.APPLIED == "applied" + assert ApplyOutcome.CONSTRAINED == "constrained" + assert ApplyOutcome.ALREADY_APPLIED == "already_applied" + assert ApplyOutcome.BLOCKED_EMPTY == "blocked_empty" + + # Check ApplyResult construction + result = ApplyResult( + outcome=ApplyOutcome.APPLIED, + plan_id="TEST", + message="ok", + files_changed=5, + validations_total=3, + validations_passed=3, + validations_failed=0, + ) + assert result.outcome == ApplyOutcome.APPLIED + assert result.plan_id == "TEST" + assert result.files_changed == 5 + + print("model-checks-ok") + return 0 + + +def main() -> int: + if len(sys.argv) < 2: + print( + "Usage: helper_apply_pipeline.py " + "" + ) + return 1 + + cmd = sys.argv[1] + if cmd == "apply-passes": + return _cmd_apply_passes() + if cmd == "apply-blocked": + return _cmd_apply_blocked() + if cmd == "already-applied": + return _cmd_already_applied() + if cmd == "empty-changeset": + return _cmd_empty_changeset() + if cmd == "model-checks": + return _cmd_model_checks() + + print(f"Unknown command: {cmd}") + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/cleveragents/application/services/plan_apply_service.py b/src/cleveragents/application/services/plan_apply_service.py index b5e24a28d..40328dc53 100644 --- a/src/cleveragents/application/services/plan_apply_service.py +++ b/src/cleveragents/application/services/plan_apply_service.py @@ -11,6 +11,8 @@ Provides the service layer for D0b.apply features: ``ERRORED`` with conflict details. - **Empty ChangeSet guard**: Block apply when the ChangeSet has no entries unless ``allow_empty`` is explicitly set. +- **Validation-gated apply**: Block apply when required validations have not + passed, setting plan to ``constrained`` state with actionable error. All public methods accept ``plan_id`` and delegate plan lookups to ``PlanLifecycleService``. @@ -19,9 +21,11 @@ All public methods accept ``plan_id`` and delegate plan lookups to from __future__ import annotations from datetime import datetime +from enum import StrEnum from typing import TYPE_CHECKING, Any import structlog +from pydantic import BaseModel, ConfigDict, Field from cleveragents.core.exceptions import PlanError, ValidationError from cleveragents.domain.models.core.change import ( @@ -36,6 +40,54 @@ if TYPE_CHECKING: logger = structlog.get_logger(__name__) + +# --------------------------------------------------------------------------- +# Apply outcome model +# --------------------------------------------------------------------------- + + +class ApplyOutcome(StrEnum): + """Outcome of a validation-gated apply attempt. + + - ``applied``: All required validations passed; changes committed. + - ``constrained``: Required validations failed; apply blocked. + - ``already_applied``: Plan was already in a terminal applied state. + - ``blocked_empty``: ChangeSet was empty and ``allow_empty`` not set. + """ + + APPLIED = "applied" + CONSTRAINED = "constrained" + ALREADY_APPLIED = "already_applied" + BLOCKED_EMPTY = "blocked_empty" + + +class ApplyResult(BaseModel): + """Result of a validation-gated apply attempt. + + Encapsulates the outcome, any error message, the plan state after + the attempt, and summary counts. + """ + + outcome: ApplyOutcome = Field(..., description="Outcome of the apply attempt") + plan_id: str = Field(..., description="Plan ULID") + message: str = Field(default="", description="Human-readable result message") + files_changed: int = Field( + default=0, description="Number of files changed during apply" + ) + validations_total: int = Field(default=0, description="Total validations evaluated") + validations_passed: int = Field( + default=0, description="Required validations that passed" + ) + validations_failed: int = Field( + default=0, description="Required validations that failed" + ) + + model_config = ConfigDict( + str_strip_whitespace=True, + validate_assignment=True, + ) + + # Colour codes for diff operations (used by rich format) _OP_COLORS: dict[str, str] = { "create": "green", @@ -410,6 +462,187 @@ class PlanApplyService: "No changes to apply. Use --allow-empty to override." ) + # -- Validation-gated apply --------------------------------------------- + + def apply_with_validation_gate( + self, + plan_id: str, + validation_summary: dict[str, Any] | None = None, + *, + allow_empty: bool = False, + ) -> ApplyResult: + """Execute a validation-gated apply for a plan. + + Checks that all required validations have passed before + proceeding with apply. If validations failed, the plan is + set to ``constrained`` state with an actionable error message. + + Args: + plan_id: The plan ULID. + validation_summary: Validation summary dict from the + Execute phase. If ``None``, the plan's stored + ``validation_summary`` is used. + allow_empty: If ``True``, allow apply on empty ChangeSet. + + Returns: + An ``ApplyResult`` describing the outcome. + """ + from cleveragents.domain.models.core.plan import ( + ProcessingState, + ) + + plan = self._lifecycle.get_plan(plan_id) + + # Guard: already in terminal state + if plan.is_terminal: + if plan.processing_state == ProcessingState.APPLIED: + self._logger.info( + "Plan already applied", + plan_id=plan_id, + ) + return ApplyResult( + outcome=ApplyOutcome.ALREADY_APPLIED, + plan_id=plan_id, + message=( + f"Plan {plan_id} has already been applied. " + "Re-apply is not permitted." + ), + ) + # Other terminal states (errored, cancelled, constrained) + # are also blocked + self._logger.info( + "Plan in terminal state, cannot apply", + plan_id=plan_id, + state=plan.processing_state.value, + ) + return ApplyResult( + outcome=ApplyOutcome.CONSTRAINED, + plan_id=plan_id, + message=( + f"Plan {plan_id} is in terminal state " + f"'{plan.processing_state.value}' and cannot be applied." + ), + ) + + # Guard: empty changeset + changeset = self._resolve_changeset(plan) + if changeset is None or (len(changeset.entries) == 0 and not allow_empty): + self._logger.info( + "Apply blocked: empty changeset", + plan_id=plan_id, + ) + return ApplyResult( + outcome=ApplyOutcome.BLOCKED_EMPTY, + plan_id=plan_id, + message=( + f"Plan {plan_id} has an empty ChangeSet. " + "No changes to apply. Use --allow-empty to override." + ), + ) + + # Resolve validation summary + vs = validation_summary or plan.validation_summary + req_passed, req_failed, total = self._extract_validation_counts(vs) + + # Gate: required validations must all pass + if req_failed > 0: + reason = ( + f"Apply refused: {req_failed} required Execute-phase " + f"validation(s) did not pass " + f"({req_passed} passed, {req_failed} failed). " + "Fix the issues and re-run validations before applying. " + "Use `agents plan prompt ` to resume or " + "`agents plan correct ` to provide fixes." + ) + self._logger.warning( + "Apply blocked by validation gate", + plan_id=plan_id, + required_passed=req_passed, + required_failed=req_failed, + ) + + # Transition to constrained if in Apply phase + try: + self._lifecycle.constrain_apply(plan_id, reason) + except Exception: + # Plan may not be in Apply phase yet; store reason + # in validation_summary for CLI display + self._logger.debug( + "Could not transition to constrained " + "(plan may not be in Apply phase)", + plan_id=plan_id, + ) + + return ApplyResult( + outcome=ApplyOutcome.CONSTRAINED, + plan_id=plan_id, + message=reason, + validations_total=total, + validations_passed=req_passed, + validations_failed=req_failed, + ) + + # All required validations passed — proceed with apply + files_changed = len(changeset.entries) + + self._logger.info( + "Validation gate passed, applying", + plan_id=plan_id, + files_changed=files_changed, + required_passed=req_passed, + ) + + # Persist apply summary + self.persist_apply_summary( + plan_id=plan_id, + files_changed=files_changed, + validations_run=total, + ) + + # Transition to applied via lifecycle service + try: + self._lifecycle.complete_apply(plan_id) + except Exception: + # If lifecycle transition fails (e.g., wrong phase), + # still report success in terms of validation gating + self._logger.debug( + "Could not transition to applied (plan may not be in Apply phase)", + plan_id=plan_id, + ) + + return ApplyResult( + outcome=ApplyOutcome.APPLIED, + plan_id=plan_id, + message=( + f"Plan {plan_id} applied successfully. " + f"{files_changed} file(s) changed, " + f"{total} validation(s) evaluated." + ), + files_changed=files_changed, + validations_total=total, + validations_passed=req_passed, + validations_failed=0, + ) + + @staticmethod + def _extract_validation_counts( + vs: dict[str, Any] | None, + ) -> tuple[int, int, int]: + """Extract required passed/failed counts from a validation summary. + + Returns: + Tuple of (required_passed, required_failed, total). + """ + if vs is None: + # No validation summary — treat as all passed (0 validations) + return 0, 0, 0 + + req_passed = int(vs.get("required_passed", 0)) + req_failed = int(vs.get("required_failed", 0)) + total = int(vs.get("total", req_passed + req_failed)) + + return req_passed, req_failed, total + # -- Internal helpers ---------------------------------------------------- def _resolve_changeset(self, plan: Plan) -> SpecChangeSet | None: