cleveragents/cleveragents-core
4
0
Fork 3
Code Issues 6.0k Pull Requests 115 Actions Packages Projects Releases Wiki Activity

feat(apply): add validation-gated apply pipeline

Browse Source 
ISSUES CLOSED: #155
This commit is contained in:
CoreRasurae
2026-02-22 10:46:49 +00:00
parent 1282bc03b3
commit ab6a3fd732
7 changed files with 1145 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
benchmarks/apply_pipeline_bench.py
+169
View File
@@ -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)
docs/reference/plan_apply.md
+64
View File
@@ -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}")
```
features/apply_pipeline.feature
+102
View File
@@ -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
features/steps/apply_pipeline_steps.py
+311
View File
@@ -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
robot/apply_pipeline.robot
+51
View File
@@ -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
robot/helper_apply_pipeline.py
+215
View File
@@ -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 "
"<apply-passes|apply-blocked|already-applied|"
"empty-changeset|model-checks>"
)
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())
src/cleveragents/application/services/plan_apply_service.py
+233
View File
@@ -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 <PLAN_ID>` to resume or "
"`agents plan correct <PLAN_ID>` 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: