CHANGELOG.md

@@ -102,6 +102,8 @@ a critical data integrity issue in `ValidationAttachmentRepository.attach` where
 fragile heuristic (`"/" in resource_id`). Arguments are now passed in the correct order,
 ensuring data is stored with proper parameter values.
 
+* **Invariant enforcement & validation pipeline (M3 Epic #8137)**: Added the invariant enforcer that detects violations during plan execution with clear, actionable error messages; the structural component validator that performs output validation via flexible substring and pattern matching (not exact character equality) so that tests remain resilient to cosmetic changes; and the phase transition gate that blocks Execute→Apply transitions when required validations fail or critical invariant constraints are violated. Validation results are persisted as `validation_response` decisions in the plan's decision tree for auditability.
+
 - Hardened the TDD bug-fix quality gate for issue #629: PR parsing now
   requires whole-word closing keywords (avoids false positives like
   "prefixes #12"), TDD bug tag discovery now uses exact token matching

CONTRIBUTORS.md

@@ -22,7 +22,7 @@
 # Details
 * HAL 9000 has contributed spec clarifications for layer boundary DI exception, ULID scope, ACMS pipeline contracts, and TUI component interfaces (PR #10451): documented architectural invariants including the DI container exception, clarified ULID identifier scope distinguishing domain entities from internal implementation details, added per-stage protocol contracts for all 10 ACMS pipeline stages with storage tier definitions, budget enforcement protocol, and context assembly output format, and defined public interfaces with verifiable checks for 8 TUI components.
 
-
+* Jeffrey Phillips Freeman has contributed the invariant enforcement and validation pipeline (Epic #8137, M3): `invariant_enforcer.py` implements runtime violation detection with actionable error messages; `structural_component_validator.py` performs flexible structural-component matching instead of brittle exact-character equality; and `phase_transition_gate.py` gates Execute→Apply transitions when required validations fail.
 
 Below are some of the specific details of various contributions.
 

features/invariant_enforcement.feature

@@ -0,0 +1,40 @@
+Feature: Invariant Enforcer - detect violations during execution
+
+  As a plan execution system
+  I want to detect when invariants are violated during plan execution
+  So that corrective or blocking actions can be taken.
+
+  @invariant_enforcer @m3_epic_8137
+  Scenario: Output satisfying an invariant passes the check
+    Given an effective set of 1 active invariant: "All APIs must maintain backward compatibility"
+    And the execution output reads: "The API endpoint now supports both v1 and v2 request formats"
+    When I check invariants against the output
+    Then no invariant violation should be detected
+
+  @invariant_enforcer @m3_epic_8137
+  Scenario: Output violating a non-overridable invariant is blocked
+    Given an effective set of 1 active non_overridable invariant: "Never commit secret keys to version control"
+    And the execution output reads: "Writing API key abc123secret to config.py"
+    When I check invariants against the output
+    Then exactly 1 error-severity violation should be detected
+
+  @invariant_enforcer @m3_epic_8137
+  Scenario: Output violating a regular invariant produces a warning
+    Given an effective set of 1 active regular invariant: "Test coverage must exceed 90%"
+    And the execution output reads: "The new module has only 45% test coverage"
+    When I check invariants against the output
+    Then exactly 1 warning-severity violation should be detected
+
+  @enforce_strict @m3_epic_8137
+  Scenario: enforce_strict raises InvariantEnforcementError on blocking violations
+    Given an effective set of 1 active non_overridable invariant: "No secrets in source code"
+    And the execution output reads: "Hardcoded password admin123 in database.py"
+    When I run strict enforcement checking against the output
+    Then an InvariantEnforcementError should be raised with blocked_phase="apply"
+
+  @enforce_strict @m3_epic_8137
+  Scenario: enforce_strict returns empty list when all checks pass
+    Given an effective set of 2 active invariants both satisfied by the output
+    And the execution output reads: "All changes maintain backward-compatible API contracts"
+    When I run strict enforcement checking against the output
+    Then no exception should be raised and the returned violations list should be empty

features/phase_transition_gating.feature

@@ -0,0 +1,34 @@
+Feature: Phase Transition Gate - gate plan progression at apply
+
+  As a plan execution system
+  I want to validate that structural output validation passes before transitioning phases
+  So that only validated, invariant-respecting changes proceed to Apply.
+
+  @phase_gate @m3_epic_8137
+  Scenario: Action to Strategize always succeeds with no gate
+    Given a plan in the ACTION phase wanting to enter STRATEGIZE phase
+    And no active invariants
+    When I run the phase transition gate for this transition
+    Then the gate decision should be allowed=True
+
+  @phase_gate @m3_epic_8137
+  Scenario: Strat to Execute with clean invariant enforcement clears the gate
+    Given a plan in STRATEGIZE phase wanting to enter EXECUTE phase
+    And no active non_overridable invariant violations on record
+    When I run the phase transition gate for the Strat-to-Exec transition
+    Then the gate decision should be allowed=True
+    And the GateDecision show checked_invariants_count=0
+
+  @phase_gate @m3_epic_8137
+  Scenario: Execute to Apply gate passes when no issues detected
+    Given a plan in EXECUTE phase wanting to enter APPLY phase
+    And no non_overridable invariant audit violations on the changeset
+    When I run the execute-to-apply gate
+    Then the gate decision should be allowed=True
+
+  @phase_gate @m3_epic_8137
+  Scenario: Gate records a validation_response decision for audit trail
+    Given a plan in STRATEGIZE phase wanting to enter EXECUTE phase
+    And no active non_overridable invariant violations on record
+    When I run the phase transition gate for this transition
+    Then the gate decision should have a recorded decision_id

features/steps/invariant_enforcement_steps.py

@@ -0,0 +1,143 @@
+"""Step definitions for Invariant Enforcer - Epic #8137."""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.core.exceptions import CleverAgentsError
+
+from cleveragents.domain.models.core.invariant import (
+    Invariant,
+    InvariantScope,
+)
+from cleveragents.application.services.invariant_enforcer import (
+    detect_violations,
+    enforce_strict,
+)
+
+
+@given('an effective set of 1 active invariant: "{text}"')
+def step_one_invariant(context: Context, text: str) -> None:
+    context.invariants = [
+        Invariant(
+            id="01HQ8K5YVZ7P3GQJD6W4T9M0A0",
+            text=text,
+            scope=InvariantScope.GLOBAL,
+            source_name="test-project",
+            non_overridable=False,
+        ),
+    ]
+
+
+@given('an effective set of 1 active non_overridable invariant: "{text}"')
+def step_one_non_override(context: Context, text: str) -> None:
+    context.invariants = [
+        Invariant(
+            id="01HQ8K5YVZ7P3GQJD6W4T9M0A0",
+            text=text,
+            scope=InvariantScope.GLOBAL,
+            source_name="test-project",
+            non_overridable=True,
+        ),
+    ]
+
+
+@given('an effective set of 1 active regular invariant: "{text}"')
+def step_one_regular(context: Context, text: str) -> None:
+    context.invariants = [
+        Invariant(
+            id="01HQ8K5YVZ7P3GQJD6W4T9M0A0",
+            text=text,
+            scope=InvariantScope.PROJECT,
+            source_name="test-project",
+            non_overridable=False,
+        ),
+    ]
+
+
+@given("an effective set of 2 active invariants both satisfied by the output")
+def step_two_satisfied(context: Context) -> None:
+    context.invariants = [
+        Invariant(
+            id=str(i),
+            text="All APIs must maintain backward compatibility",
+            scope=InvariantScope.GLOBAL,
+            source_name="test-project",
+            non_overridable=False,
+        )
+        for i in range(2)
+    ]
+
+
+@given('the execution output reads: "{output}"')
+def step_execution_output(context: Context, output: str) -> None:
+    context.execution_outputs = [output]
+
+
+@when("I check invariants against the output")
+def step_check_violations(context: Context) -> None:
+    outputs = getattr(context, "execution_outputs", [])
+    if not outputs:
+        outputs = [""]
+    context.violations = detect_violations(context.invariants, outputs)
+
+
+@when("I run strict enforcement checking against the output")
+def step_strict_enforce(context: Context) -> None:
+    outputs = getattr(context, "execution_outputs", [])
+    if not outputs:
+        outputs = [""]
+    context.enforce_error = None
+    try:
+        result = enforce_strict(
+            invariants=context.invariants,
+            outputs_to_check=outputs,
+        )
+        context.strict_result = result
+    except CleverAgentsError as exc:
+        context.enforce_error = exc
+
+
+@then("no invariant violation should be detected")
+def step_no_violation(context: Context) -> None:
+    assert len(context.violations) == 0, (
+        f"Got {len(context.violations)} violations instead of 0."
+    )
+
+
+@then("exactly {count:d} error-severity violation should be detected")
+@then("exactly {count:d} error-severity violations should be detected")
+def step_n_error_violations(context: Context, count: int) -> None:
+    errors = [v for v in context.violations if v.severity == "error"]
+    assert len(errors) == count, (
+        f"Expected {count} error violations but got {len(errors)}."
+    )
+
+
+@then("exactly {count:d} warning-severity violation should be detected")
+@then("exactly {count:d} warning-severity violations should be detected")
+def step_n_warning_violations(context: Context, count: int) -> None:
+    warnings = [v for v in context.violations if v.severity == "warning"]
+    assert len(warnings) == count, (
+        f"Expected {count} warning violations but got {len(warnings)}."
+    )
+
+
+@then('an InvariantEnforcementError should be raised with blocked_phase="apply"')
+def step_enforce_error(context: Context) -> None:
+    assert context.enforce_error is not None, "No error was raised."
+    from cleveragents.application.services.invariant_enforcer import (
+        InvariantEnforcementError,
+    )
+
+    assert isinstance(context.enforce_error, InvariantEnforcementError), (
+        f"Expected InvariantEnforcementError, got {type(context.enforce_error).__name__}"
+    )
+
+
+@then("no exception should be raised and the returned violations list should be empty")
+def step_clean_strict(context: Context) -> None:
+    assert context.enforce_error is None, f"Got error: {context.enforce_error}"
+    if hasattr(context, "strict_result"):
+        assert len(context.strict_result) == 0

features/steps/phase_transition_gating_steps.py

@@ -0,0 +1,139 @@
+"""Step definitions for Phase Transition Gate - Epic #8137."""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.decision_service import DecisionService
+from cleveragents.application.services.phase_transition_gate import (
+    GateDecision,
+    PhaseTransitionGate,
+)
+from cleveragents.core.exceptions import DomainError
+from cleveragents.domain.models.core.plan import PlanPhase
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a plan in the ACTION phase wanting to enter STRATEGIZE phase")
+def step_action_strat(context: Context) -> None:
+    context.current_phase = PlanPhase.ACTION
+    context.target_phase = PlanPhase.STRATEGIZE
+
+
+@given("a plan in STRATEGIZE phase wanting to enter EXECUTE phase")
+def step_strat_exec(context: Context) -> None:
+    context.current_phase = PlanPhase.STRATEGIZE
+    context.target_phase = PlanPhase.EXECUTE
+
+
+@given("a plan in EXECUTE phase wanting to enter APPLY phase")
+def step_exec_apply(context: Context) -> None:
+    context.current_phase = PlanPhase.EXECUTE
+    context.target_phase = PlanPhase.APPLY
+
+
+@given("no active invariants")
+def step_no_invariants(context: Context) -> None:
+    context.invariants = []
+
+
+@given("no active non_overridable invariant violations on record")
+def step_clean_violations(context: Context) -> None:
+    """Clean state with no violation issues."""
+    context.decision_service = DecisionService()
+    context.invariants = []
+
+
+@given("no non_overridable invariant audit violations on the changeset")
+def step_no_audit_issues(context: Context) -> None:
+    """Empty invariants for clean gate check."""
+    context.invariants = []
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I run the phase transition gate for this transition")
+def step_run_gate(context: Context) -> None:
+    context.gate_error = None
+    context.decision_service = DecisionService()
+    try:
+        gate = PhaseTransitionGate(decision_service=context.decision_service)
+        result = gate.gate(
+            plan_id="01HQ8K5YVZ7P3GQJD6W4T9MXXX",
+            current_phase=context.current_phase,
+            target_phase=context.target_phase,
+            invariants=getattr(context, "invariants", []),
+        )
+        context.gate_decision = result
+    except DomainError as exc:
+        context.gate_error = exc
+
+
+@when("I run the phase transition gate for the Strat-to-Exec transition")
+def step_strat_exec_gate(context: Context) -> None:
+    context.gate_error = None
+    context.decision_service = DecisionService()
+    try:
+        gate = PhaseTransitionGate(decision_service=context.decision_service)
+        result = gate.gate(
+            plan_id="01HQ8K5YVZ7P3GQJD6W4T9MXXX",
+            current_phase=PlanPhase.STRATEGIZE,
+            target_phase=PlanPhase.EXECUTE,
+            invariants=getattr(context, "invariants", []),
+        )
+        context.gate_decision = result
+    except DomainError as exc:
+        context.gate_error = exc
+
+
+@when("I run the execute-to-apply gate")
+def step_exec_apply_gate(context: Context) -> None:
+    context.gate_error = None
+    context.decision_service = DecisionService()
+    try:
+        gate = PhaseTransitionGate(decision_service=context.decision_service)
+        result = gate.gate(
+            plan_id="01HQ8K5YVZ7P3GQJD6W4T9MXXX",
+            current_phase=PlanPhase.EXECUTE,
+            target_phase=PlanPhase.APPLY,
+            invariants=getattr(context, "invariants", []),
+        )
+        context.gate_decision = result
+    except DomainError as exc:
+        context.gate_error = exc
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("the gate decision should be allowed=True")
+def step_gate_allowed(context: Context) -> None:
+    assert context.gate_error is None, f"Gate raised error: {context.gate_error}"
+    dec: GateDecision = context.gate_decision
+    assert dec.allowed is True, f"Expected allowed but got {dec.allowed}."
+
+
+@then("the GateDecision show checked_invariants_count={count:d}")
+def step_show_check_count(context: Context, count: int) -> None:
+    assert context.gate_error is None, f"Gate raised error: {context.gate_error}"
+    dec: GateDecision = context.gate_decision
+    assert dec.checked_invariants_count == count, (
+        f"Expected {count} but got {dec.checked_invariants_count}"
+    )
+
+
+@then("the gate decision should have a recorded decision_id")
+def step_decision_recorded(context: Context) -> None:
+    assert context.gate_error is None, f"Gate raised error: {context.gate_error}"
+    dec: GateDecision = context.gate_decision
+    assert dec.decision_id is not None, "Expected a recorded decision_id but got None."

features/steps/structural_component_validation_steps.py

@@ -0,0 +1,146 @@
+"""Step definitions for Structural Component Validator - Epic #8137."""
+
+from __future__ import annotations
+
+from behave import given, then, when
+from behave.runner import Context
+
+from cleveragents.application.services.structural_component_validator import (
+    ComponentCheckResult,
+    ExpectKind,
+    StructuralExpectation,
+    ValidationOutcome,
+)
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given('a present structural expectation on keyword "{keyword}"')
+def step_present(context: Context, keyword: str) -> None:
+    """Append a presence expectation with case-insensitive matching."""
+    name = f"present-on-{keyword.replace(' ', '-')}"
+    if not hasattr(context, "expectations") or context.expectations is None:
+        context.expectations = []
+    context.expectations.append(
+        StructuralExpectation(
+            name=name,
+            kind=ExpectKind.PRESENT,
+            value=keyword,
+            case_sensitive=False,
+            weight=1.0,
+        ),
+    )
+
+
+@given('an absent structural expectation on keyword "{keyword}"')
+def step_absent(context: Context, keyword: str) -> None:
+    """Append an absence expectation with case-insensitive matching."""
+    name = f"absent-on-{keyword.replace(' ', '-')}"
+    if not hasattr(context, "expectations") or context.expectations is None:
+        context.expectations = []
+    context.expectations.append(
+        StructuralExpectation(
+            name=name,
+            kind=ExpectKind.ABSENT,
+            value=keyword,
+            case_sensitive=False,
+            weight=1.0,
+        ),
+    )
+
+
+@given('a pattern expectation on regex "{pattern}"')
+def step_pattern(context: Context, pattern: str) -> None:
+    """Create a pattern expectation with case-insensitive matching."""
+    context.expectations = [
+        StructuralExpectation(
+            name="pattern-match",
+            kind=ExpectKind.PATTERN,
+            value=pattern,
+            case_sensitive=False,
+            weight=1.0,
+        ),
+    ]
+
+
+@given('a present expectation on field path "{field_path}"')
+def step_nested(context: Context, field_path: str) -> None:
+    """Create a nested-field-presence expectation."""
+    context.expectations = [
+        StructuralExpectation(
+            name=f"field-present-{field_path}",
+            kind=ExpectKind.FIELD_PRESENT,
+            value=field_path,
+            case_sensitive=False,
+            weight=1.0,
+        ),
+    ]
+
+
+@given('the actual output reads: "{output_text}"')
+def step_actual_output(context: Context, output_text: str) -> None:
+    """Set the actual output string."""
+    context.actual_outputs = [output_text]
+
+
+@given("the actual output reads JSON: {output_text}")
+def step_actual_output_json(context: Context, output_text: str) -> None:
+    """Set the actual output as raw JSON text."""
+    context.actual_outputs = [output_text]
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I validate the outputs against the expectations")
+def step_validate(context: Context) -> None:
+    """Validate structural expectations against outputs."""
+    from cleveragents.application.services.structural_component_validator import (
+        validate_structural_components,
+    )
+
+    outputs = getattr(context, "actual_outputs", [""])
+    context.outcome = validate_structural_components(
+        expectations=context.expectations,
+        actual_outputs=outputs,
+    )
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then("the ValidationOutcome should have passed={value}")
+def step_validate_passed(context: Context, value: str) -> None:
+    """Assert overall pass/fail."""
+    outcome: ValidationOutcome = context.outcome
+    expected = value.lower() == "true"
+    assert outcome.passed is expected, (
+        f"Expected {expected} but got {outcome.passed}. "
+        f"Failed: {outcome.failed_count}, Partial: {outcome.partial_count}"
+    )
+
+
+@then("total_checks should equal {count:d}")
+def step_check_total(context: Context, count: int) -> None:
+    """Assert total component check count."""
+    outcome: ValidationOutcome = context.outcome
+    assert outcome.total_checks == count, (
+        f"Expected {count} checks but got {outcome.total_checks}"
+    )
+
+
+@then("all checks should PASS")
+def step_all_pass(context: Context) -> None:
+    """Assert every individual component passed."""
+    outcome: ValidationOutcome = context.outcome
+    for check in outcome.results:
+        assert check.result == ComponentCheckResult.PASS, (
+            f"Component '{check.expectation.name}' not PASS."
+        )

features/structural_component_validation.feature

@@ -0,0 +1,49 @@
+Feature: Structural Component Validator - flexible output validation (not exact matching)
+
+  As a plan execution system
+  I want to validate outputs against structural components rather than exact strings
+  So that tests remain resilient to formatting changes and cosmetic modifications.
+
+  @structural_validator @m3_epic_8137
+  Scenario: Present expectation with plain substring match passes (case insensitive)
+    Given a present structural expectation on keyword "backward compatibility"
+    And the actual output reads: "The API update maintains backward COMPATIBILITY across all endpoints"
+    When I validate the outputs against the expectations
+    Then total_checks should equal 1
+
+  @structural_validator @m3_epic_8137
+  Scenario: Absent expectation passes when prohibited text is missing
+    Given an absent structural expectation on keyword "secret key"
+    And the actual output reads: "Configuration loaded from environment variables"
+    When I validate the outputs against the expectations
+    Then the ValidationOutcome should have passed=True
+
+  @structural_validator @m3_epic_8137
+  Scenario: Absent expectation fails when prohibited text is found
+    Given an absent structural expectation on keyword "password"
+    And the actual output reads: "Using password auth for database connection"
+    When I validate the outputs against the expectations
+    Then the ValidationOutcome should have passed=False
+
+  @structural_validator @m3_epic_8137
+  Scenario: Regex pattern matching works across whitespace variations
+    Given a pattern expectation on regex "status\\s*:\\s*(passed|ok)"
+    And the actual output reads: "Status : passed"
+    When I validate the outputs against the expectations
+    Then total_checks should equal 1
+
+  @structural_validator @m3_epic_8137
+  Scenario: Nested JSON field path validation finds existing fields
+    Given a present expectation on field path "data.status.passed"
+    And the actual output reads JSON: {"data": {"status": {"passed": true}}}
+    When I validate the outputs against the expectations
+    Then total_checks should equal 1
+
+  @structural_validator @m3_epic_8137
+  Scenario: Multiple structural expectations aggregate correctly
+    Given a present structural expectation on keyword "database"
+    And an absent structural expectation on keyword "password"
+    And the actual output reads: "Database initialized. Status is ok, no password used."
+    When I validate the outputs against the expectations
+    Then the ValidationOutcome should have passed=True
+    And total_checks should equal 2

robot/helper_invariant_enforcer.py

@@ -0,0 +1,223 @@
+"""Helper for ``invariant_enforcement.robot`` — exercises the invariant enforcer."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+# Ensure the project root is importable.
+_ROOT = str(Path(__file__).resolve().parents[1])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+from cleveragents.application.services.invariant_enforcer import (  # noqa: E402
+    InvariantEnforcementError,
+    ViolationSeverity,
+    check_and_enforce,
+    detect_violations,
+    enforce_strict,
+)
+from cleveragents.domain.models.core.invariant import (  # noqa: E402
+    Invariant,
+    InvariantScope,
+)
+
+# ---------------------------------------------------------------------------
+# Violation detection tests
+# ---------------------------------------------------------------------------
+
+
+def detect_violations_non_overridable_error() -> int:
+    """Verify a non-overridable invariant triggers an error-severity violation."""
+    inv = Invariant(
+        id="01HQ8K5YVZ7P3GQJD6W4T9M0A0",
+        text="All APIs must maintain backward compatibility",
+        scope=InvariantScope.GLOBAL,
+        source_name="test-project",
+        non_overridable=True,
+    )
+    violations = detect_violations([inv], ["some unrelated output"])
+    errors = [v for v in violations if v.severity == "error"]
+    if len(errors) != 1:
+        print(f"FAIL: expected 1 error violation, got {len(errors)}", file=sys.stderr)
+        return 1
+    print("violation-non-overridable-ok")
+    return 0
+
+
+def detect_violations_regular_warning() -> int:
+    """Verify a regular (non-overridable) invariant triggers a warning."""
+    inv = Invariant(
+        id="01HQ8K5YVZ7P3GQJD6W4T9M0A1",
+        text="Must include unit tests for all new functions",
+        scope=InvariantScope.PROJECT,
+        source_name="test-project",
+        non_overridable=False,
+    )
+    violations = detect_violations([inv], ["some unrelated output"])
+    warnings = [v for v in violations if v.severity == "warning"]
+    if len(warnings) != 1:
+        print(
+            f"FAIL: expected 1 warning violation, got {len(warnings)}", file=sys.stderr
+        )
+        return 1
+    print("violation-regular-warning-ok")
+    return 0
+
+
+def detect_violations_no_failure() -> int:
+    """Verify compliant output produces no violations."""
+    inv = Invariant(
+        id="01HQ8K5YVZ7P3GQJD6W4T9M0A2",
+        text="All APIs must maintain backward compatibility",
+        scope=InvariantScope.GLOBAL,
+        source_name="test-project",
+        non_overridable=True,
+    )
+    violations = detect_violations([inv], ["backwards compatible API implementation"])
+    if len(violations) != 0:
+        print(f"FAIL: expected 0 violations, got {len(violations)}", file=sys.stderr)
+        return 1
+    print("violation-no-failure-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# check_and_enforce tests
+# ---------------------------------------------------------------------------
+
+
+def check_and_enforce_with_violations() -> int:
+    """Verify check_and_enforce returns violations and actions."""
+    inv = Invariant(
+        id="01HQ8K5YVZ7P3GQJD6W4T9M0A3",
+        text="All APIs must maintain backward compatibility",
+        scope=InvariantScope.GLOBAL,
+        source_name="test-project",
+        non_overridable=True,
+    )
+    violations, actions = check_and_enforce([inv], ["unrelated output"])
+
+    if len(violations) == 0:
+        print("FAIL: expected violations from check_and_enforce", file=sys.stderr)
+        return 1
+    if len(actions) == 0:
+        print("FAIL: expected actions from check_and_enforce", file=sys.stderr)
+        return 1
+
+    error_actions = [a for a in actions if a.severity == ViolationSeverity.ERROR]
+    if len(error_actions) != 1:
+        print(
+            f"FAIL: expected 1 error action, got {len(error_actions)}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("check_and_enforce-with-violations-ok")
+    return 0
+
+
+def check_and_enforce_clean() -> int:
+    """Verify check_and_enforce returns empty when all pass."""
+    inv = Invariant(
+        id="01HQ8K5YVZ7P3GQJD6W4T9M0A4",
+        text="All APIs must maintain backward compatibility",
+        scope=InvariantScope.GLOBAL,
+        source_name="test-project",
+        non_overridable=True,
+    )
+    violations, actions = check_and_enforce(
+        [inv], ["backwards compatible API implementation"]
+    )
+
+    if len(violations) != 0:
+        print(f"FAIL: expected 0 violations, got {len(violations)}", file=sys.stderr)
+        return 1
+    if len(actions) != 0:
+        print(f"FAIL: expected 0 actions, got {len(actions)}", file=sys.stderr)
+        return 1
+
+    print("check_and_enforce-clean-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# enforce_strict tests
+# ---------------------------------------------------------------------------
+
+
+def enforce_strict_raises_error() -> int:
+    """Verify enforce_strict raises InvariantEnforcementError for errors."""
+    inv = Invariant(
+        id="01HQ8K5YVZ7P3GQJD6W4T9M0A5",
+        text="All APIs must maintain backward compatibility",
+        scope=InvariantScope.GLOBAL,
+        source_name="test-project",
+        non_overridable=True,
+    )
+    try:
+        enforce_strict([inv], ["unrelated output"])
+        print("FAIL: expected InvariantEnforcementError to be raised", file=sys.stderr)
+        return 1
+    except InvariantEnforcementError as exc:
+        if hasattr(exc, "blocked_phase") and exc.blocked_phase != "apply":
+            print(
+                f"FAIL: expected blocked_phase='apply', got '{exc.blocked_phase}'",
+                file=sys.stderr,
+            )
+            return 1
+
+    print("enforce_strict-raises-error-ok")
+    return 0
+
+
+def enforce_strict_clean() -> int:
+    """Verify enforce_strict returns empty list when all pass."""
+    inv = Invariant(
+        id="01HQ8K5YVZ7P3GQJD6W4T9M0A6",
+        text="All APIs must maintain backward compatibility",
+        scope=InvariantScope.GLOBAL,
+        source_name="test-project",
+        non_overridable=False,
+    )
+    result = enforce_strict([inv], ["backwards compatible API implementation"])
+    if len(result) != 0:
+        print(f"FAIL: expected empty list, got {len(result)}", file=sys.stderr)
+        return 1
+
+    print("enforce_strict-clean-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+_COMMANDS: dict[str, object] = {
+    "detect_violations_non_overridable_error": detect_violations_non_overridable_error,
+    "detect_violations_regular_warning": detect_violations_regular_warning,
+    "detect_violations_no_failure": detect_violations_no_failure,
+    "check_and_enforce_with_violations": check_and_enforce_with_violations,
+    "check_and_enforce_clean": check_and_enforce_clean,
+    "enforce_strict_raises_error": enforce_strict_raises_error,
+    "enforce_strict_clean": enforce_strict_clean,
+}
+
+
+def main() -> int:
+    """Dispatch to the sub-command named in sys.argv[1]."""
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
+        return 2
+
+    cmd = sys.argv[1]
+    handler = _COMMANDS.get(cmd)
+    if handler is None:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        return 2
+
+    return handler()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

robot/helper_phase_transition_gate.py

@@ -0,0 +1,315 @@
+"""Helper for ``phase_transition_gating.robot`` — exercises the gate."""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+# Ensure the project root is importable.
+_ROOT = str(Path(__file__).resolve().parents[1])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+from cleveragents.application.services.phase_transition_gate import (  # noqa: E402
+    GateDecision,
+    PhaseTransitionGate,
+    run_phase_gate,
+)
+from cleveragents.domain.models.core.invariant import (  # noqa: E402
+    Invariant,
+    InvariantScope,
+)
+from cleveragents.domain.models.core.plan import PlanPhase  # noqa: E402
+
+# ---------------------------------------------------------------------------
+# Strategize → Execute gate — no violations allowed
+# ---------------------------------------------------------------------------
+
+
+def strategize_to_execute_clean() -> int:
+    """Verify STRATEGIZE→EXECUTE allows transit when no invariants violate."""
+    mock_decision = MagicMock()
+    mock_decision.get_tree.return_value = []
+
+    gate = PhaseTransitionGate(decision_service=mock_decision)
+    decision = gate.gate(
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A0",
+        current_phase=PlanPhase.STRATEGIZE,
+        target_phase=PlanPhase.EXECUTE,
+        invariants=[],
+    )
+
+    if not decision.allowed:
+        reason = decision.blocked_reason
+        print(
+            f"FAIL: expected allowed=True for STRATEGIZE→EXECUTE clean, got {reason}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("strategize-to-execute-clean-ok")
+    return 0
+
+
+def strategize_to_execute_with_invariants_no_violations() -> int:
+    """Verify STRATEGIZE→EXECUTE allows transit when invariants are satisfied."""
+    inv = Invariant(
+        id="inv-1",
+        text="All APIs must maintain backward compatibility",
+        scope=InvariantScope.GLOBAL,
+        source_name="test-project",
+        non_overridable=True,
+    )
+
+    mock_decision = MagicMock()
+    mock_decision.get_tree.return_value = []
+
+    gate = PhaseTransitionGate(decision_service=mock_decision)
+    decision = gate.gate(
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A1",
+        current_phase=PlanPhase.STRATEGIZE,
+        target_phase=PlanPhase.EXECUTE,
+        invariants=[inv],
+    )
+
+    if not decision.allowed:
+        reason = decision.blocked_reason
+        print(
+            f"FAIL: expected allowed=True with satisfied invariants, got {reason}",
+            file=sys.stderr,
+        )
+        return 1
+
+    if decision.checked_invariants_count != 1:
+        count = decision.checked_invariants_count
+        print(
+            f"FAIL: expected checked_invariants_count=1, got {count}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("strategize-to-execute-with-invariants-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Actions gate — should always allow (terminal phases)
+# ========================================================================= ==
+
+
+def action_to_apply_allows_unconditionally() -> int:
+    """Verify ACTION→APPLY transitions are allowed unconditionally."""
+    mock_decision = MagicMock()
+
+    gate = PhaseTransitionGate(decision_service=mock_decision)
+    decision = gate.gate(
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A2",
+        current_phase=PlanPhase.ACTION,
+        target_phase=PlanPhase.APPLY,
+        invariants=[],
+    )
+
+    if not decision.allowed:
+        print(
+            "FAIL: expected ACTION→APPLY to be allowed unconditionally", file=sys.stderr
+        )
+        return 1
+
+    print("action-to-apply-unconditional-ok")
+    return 0
+
+
+def apply_terminal_allows_unconditionally() -> int:
+    """Verify APPLY phase transitions are allowed unconditionally."""
+    mock_decision = MagicMock()
+
+    gate = PhaseTransitionGate(decision_service=mock_decision)
+    decision = gate.gate(
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A3",
+        current_phase=PlanPhase.APPLY,
+        target_phase=PlanPhase.STRATEGIZE,
+        invariants=[],
+    )
+
+    if not decision.allowed:
+        print(
+            "FAIL: expected APPLY→anything to be allowed unconditionally",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("apply-terminal-unconditional-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Execute → Apply gate — validation check
+# ========================================================================= ==
+
+
+def execute_to_apply_validations_passed() -> int:
+    """Verify EXECUTE→APPLY allows transit when all validations pass."""
+
+    mock_decision = MagicMock()
+    mock_decision.get_tree.return_value = []
+
+    gate = PhaseTransitionGate(decision_service=mock_decision)
+    decision = gate.gate(
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A4",
+        current_phase=PlanPhase.EXECUTE,
+        target_phase=PlanPhase.APPLY,
+        invariants=[],
+    )
+
+    if not decision.allowed:
+        reason = decision.blocked_reason
+        print(
+            f"FAIL: expected allowed for EXECUTE→APPLY no validations, got {reason}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("execute-to-apply-validations-passed-ok")
+    return 0
+
+
+def execute_to_apply_validation_failures_block() -> int:
+    """Verify EXECUTE→APPLY blocks transit when validation failures exist."""
+    from cleveragents.domain.models.core.decision import DecisionType
+
+    mock_decision = MagicMock()
+    fake_decisions = [MagicMock(spec=lambda: None)]
+    fake_decisions[0].decision_type = DecisionType.VALIDATION_RESPONSE
+    fake_decisions[0].chosen_option = "false: validation failed"
+    mock_decision.get_tree.return_value = fake_decisions
+
+    gate = PhaseTransitionGate(decision_service=mock_decision)
+    decision = gate.gate(
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A5",
+        current_phase=PlanPhase.EXECUTE,
+        target_phase=PlanPhase.APPLY,
+        invariants=[],
+    )
+
+    if decision.allowed:
+        print(
+            "FAIL: expected EXECUTE→APPLY to be blocked when validations fail",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("execute-to-apply-validation-failures-blocked-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Gate Decision model — fields present
+# ========================================================================= ==
+
+
+def gate_decision_has_necessary_fields() -> int:
+    """Verify the GateDecision dataclass has all expected attributes."""
+    decision = GateDecision(
+        allowed=True,
+        blocked_reason="",
+        phase_from=PlanPhase.STRATEGIZE,
+        phase_to=PlanPhase.EXECUTE,
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A6",
+    )
+
+    required_attrs = ("allowed", "blocked_reason", "phase_from", "phase_to", "plan_id")
+    for attr in required_attrs:
+        if not hasattr(decision, attr):
+            print(f"FAIL: GateDecision missing attribute '{attr}'", file=sys.stderr)
+            return 1
+
+    print("gate-decision-has-fields-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# run_phase_gate convenience function tests
+# ========================================================================= ==
+
+
+def run_phase_gate_strategize_to_execute() -> int:
+    """Verify run_phase_gate helper delegates correctly for STRATEGIZE→EXECUTE."""
+    mock_decision = MagicMock()
+    decision = run_phase_gate(
+        decision_service=mock_decision,
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A7",
+        current_phase=PlanPhase.STRATEGIZE,
+        target_phase=PlanPhase.EXECUTE,
+        invariants=[],
+    )
+
+    if not decision.allowed:
+        print(
+            "FAIL: expected run_phase_gate to allow STRATEGIZE→EXECUTE", file=sys.stderr
+        )
+        return 1
+
+    print("run-phase-gate-strat-to-exec-ok")
+    return 0
+
+
+def run_phase_gate_execute_to_apply() -> int:
+    """Verify run_phase_gate helper delegates correctly for EXECUTE→APPLY."""
+    mock_decision = MagicMock()
+    decision = run_phase_gate(
+        decision_service=mock_decision,
+        plan_id="01HQ8K5YVZ7P3GQJD6W4T9M0A8",
+        current_phase=PlanPhase.EXECUTE,
+        target_phase=PlanPhase.APPLY,
+        invariants=[],
+    )
+
+    if not decision.allowed:
+        print(
+            "FAIL: expected run_phase_gate to allow EXECUTE→APPLY (no failures)",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("run-phase-gate-exec-to-apply-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ========================================================================= ==
+
+_strat_to_exec_with_inv = strategize_to_execute_with_invariants_no_violations
+_exec_to_apply_block = execute_to_apply_validation_failures_block
+
+_COMMANDS: dict[str, object] = {
+    "strategize_to_execute_clean": strategize_to_execute_clean,
+    "strategize_to_execute_with_invariants_no_violations": _strat_to_exec_with_inv,
+    "action_to_apply_allows_unconditionally": action_to_apply_allows_unconditionally,
+    "apply_terminal_allows_unconditionally": apply_terminal_allows_unconditionally,
+    "execute_to_apply_validations_passed": execute_to_apply_validations_passed,
+    "execute_to_apply_validation_failures_block": _exec_to_apply_block,
+    "gate_decision_has_necessary_fields": gate_decision_has_necessary_fields,
+    "run_phase_gate_strategize_to_execute": run_phase_gate_strategize_to_execute,
+    "run_phase_gate_execute_to_apply": run_phase_gate_execute_to_apply,
+}
+
+
+def main() -> int:
+    """Dispatch to the sub-command named in sys.argv[1]."""
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
+        return 2
+
+    cmd = sys.argv[1]
+    handler = _COMMANDS.get(cmd)
+    if handler is None:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        return 2
+
+    return handler()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

robot/helper_structural_component_validator.py

@@ -0,0 +1,433 @@
+"""Helper for ``structural_component_validation.robot`` — exercises the validator."""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+# Ensure the project root is importable.
+_ROOT = str(Path(__file__).resolve().parents[1])
+if _ROOT not in sys.path:
+    sys.path.insert(0, _ROOT)
+
+from cleveragents.application.services.structural_component_validator import (  # noqa: E402
+    ComponentCheckResult,
+    ExpectKind,
+    StructuralExpectation,
+    validate_single,
+    validate_structural_components,
+)
+
+# ---------------------------------------------------------------------------
+# Simple presence checks
+# ---------------------------------------------------------------------------
+
+
+def present_case_insensitive_match() -> int:
+    """Verify case-insensitive substring presence check passes."""
+    expectation = StructuralExpectation(
+        name="has-bugfix",
+        kind=ExpectKind.PRESENT,
+        value="bug fix",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["The bugfix module was updated."]
+    )
+    if not outcome.passed:
+        print(f"FAIL: expected passed=True, got {outcome.passed}", file=sys.stderr)
+        return 1
+
+    has_fail = any(r.result == ComponentCheckResult.FAIL for r in outcome.results)
+    if has_fail:
+        print("FAIL: unexpected FAIL result", file=sys.stderr)
+        return 1
+
+    print("present-case-insensitive-ok")
+    return 0
+
+
+def present_case_sensitive_no_match() -> int:
+    """Verify case-sensitive match fails when casing differs."""
+    expectation = StructuralExpectation(
+        name="has-exact-text",
+        kind=ExpectKind.PRESENT,
+        value="BugFix",
+        case_sensitive=True,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["The bugfix module was updated."]
+    )
+
+    fails = [r for r in outcome.results if r.result == ComponentCheckResult.FAIL]
+    if len(fails) != 1:
+        print(f"FAIL: expected 1 FAIL, got {len(fails)}", file=sys.stderr)
+        return 1
+
+    print("present-case-sensitive-no-match-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Absence checks
+# ---------------------------------------------------------------------------
+
+
+def absent_prohibited_component() -> int:
+    """Verify absence check fails when prohibited text is present."""
+    expectation = StructuralExpectation(
+        name="no-passwords",
+        kind=ExpectKind.ABSENT,
+        value="password=12345",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["Config: password=12345 and host=localhost"]
+    )
+
+    fails = [r for r in outcome.results if r.result == ComponentCheckResult.FAIL]
+    if len(fails) != 1:
+        print(f"FAIL: expected 1 FAIL, got {len(fails)}", file=sys.stderr)
+        return 1
+
+    print("absent-prohibited-presents-fail-ok")
+    return 0
+
+
+def absent_clean_output_passes() -> int:
+    """Verify absence check passes when prohibited text is NOT present."""
+    expectation = StructuralExpectation(
+        name="no-passwords",
+        kind=ExpectKind.ABSENT,
+        value="password=12345",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["Config: host=localhost; port=8080"]
+    )
+
+    if not outcome.passed:
+        print(
+            "FAIL: expected passed=True for absent check with clean output",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("absent-clean-passthrough-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Regex pattern checks
+# ---------------------------------------------------------------------------
+
+
+_EMAIL_RE = r"\s*[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+\s*"
+
+
+def pattern_regex_matches() -> int:
+    """Verify regex pattern check passes when output matches."""
+    expectation = StructuralExpectation(
+        name="has-email",
+        kind=ExpectKind.PATTERN,
+        value=_EMAIL_RE,
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["Contact: user@example.com for support."]
+    )
+
+    if not outcome.passed:
+        print(f"FAIL: expected passed=True, got {outcome.passed}", file=sys.stderr)
+        return 1
+
+    print("pattern-regex-matches-ok")
+    return 0
+
+
+def pattern_regex_no_match() -> int:
+    """Verify regex pattern check fails when output does not match."""
+    expectation = StructuralExpectation(
+        name="has-email",
+        kind=ExpectKind.PATTERN,
+        value=_EMAIL_RE,
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["No email address provided."]
+    )
+
+    fails = [r for r in outcome.results if r.result == ComponentCheckResult.FAIL]
+    if len(fails) != 1:
+        print(f"FAIL: expected 1 FAIL, got {len(fails)}", file=sys.stderr)
+        return 1
+
+    print("pattern-regex-no-match-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Nested field path checks (FIELD_PRESENT) — with traversal fix
+# ---------------------------------------------------------------------------
+
+
+def nested_field_present_correctly_resolved() -> int:
+    """Verify FIELD_PRESENT pass when all keys in the dotpath exist."""
+    output = json.dumps({"user": {"profile": {"name": "Alice"}}})
+    expectation = StructuralExpectation(
+        name="has-user-name",
+        kind=ExpectKind.FIELD_PRESENT,
+        value="user.profile.name",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components([expectation], [output])
+
+    if not outcome.passed:
+        got = outcome.passed
+        print(
+            f"FAIL: expected PASS for fully-resolved nested key path, got {got}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("nested-field-present-resolved-ok")
+    return 0
+
+
+def nested_field_absent_when_path_truncated() -> int:
+    """Verify ABSENT PASS when the nested path is truncated (key missing)."""
+    output = json.dumps({"user": {"name": "Alice"}})
+    expectation = StructuralExpectation(
+        name="no-profile-email",
+        kind=ExpectKind.ABSENT,
+        value="user.profile.email",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components([expectation], [output])
+
+    if not outcome.passed:
+        got = outcome.passed
+        print(
+            f"FAIL: expected PASS for ABSENT truncated at 'profile', got {got}",
+            file=sys.stderr,
+        )
+        return 1
+
+    has_fail = any(r.result == ComponentCheckResult.FAIL for r in outcome.results)
+    if has_fail:
+        print(
+            "FAIL: unexpected FAIL from ABSENT check on truncated path", file=sys.stderr
+        )
+        return 1
+
+    print("nested-field-absent-truncated-ok")
+    return 0
+
+
+def nested_field_present_failes_when_path_truncated() -> int:
+    """Verify FIELD_PRESENT FAIL when the nested path is truncated (key missing)."""
+    output = json.dumps({"user": {"name": "Alice"}})
+    expectation = StructuralExpectation(
+        name="has-profile-email",
+        kind=ExpectKind.FIELD_PRESENT,
+        value="user.profile.email",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components([expectation], [output])
+
+    fails = [r for r in outcome.results if r.result == ComponentCheckResult.FAIL]
+    if len(fails) != 1:
+        print(
+            f"FAIL: expected 1 FAIL when path is truncated, got {len(fails)} failures",
+            file=sys.stderr,
+        )
+        return 1
+
+    has_pass = any(r.result == ComponentCheckResult.PASS for r in outcome.results)
+    if has_pass:
+        print(
+            "FAIL: unexpected PASS from FIELD_PRESENT on truncated path (old bug)",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("nested-field-present-truncated-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Multi-weight aggregation
+# ---------------------------------------------------------------------------
+
+
+def weighted_score_aggregates_correctly() -> int:
+    """Verify aggregated score from multiple expectations reflects mixed results."""
+    present = StructuralExpectation(
+        name="has-text", kind=ExpectKind.PRESENT, value="text", weight=0.5
+    )
+    absent = StructuralExpectation(
+        name="no-secrets", kind=ExpectKind.ABSENT, value="password", weight=0.3
+    )
+    pattern = StructuralExpectation(
+        name="has-log-level",
+        kind=ExpectKind.PATTERN,
+        value=r"\s*(INFO|DEBUG|ERROR|WARN)\s*",
+        case_sensitive=False,
+        weight=0.2,
+    )
+
+    outcome = validate_structural_components(
+        [present, absent, pattern], ["LOG INFO: text processing"]
+    )
+
+    if not outcome.passed:
+        print(
+            f"FAIL: expected passed=True for all-passing suite, got {outcome.passed}",
+            file=sys.stderr,
+        )
+        return 1
+
+    # All weight should be achieved (score = 1.0)
+    if abs(outcome.weighted_score - 1.0) > 0.01:
+        print(
+            f"FAIL: expected weighted_score ~1.0, got {outcome.weighted_score}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("weighted-score-aggregation-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Single validation API
+# ---------------------------------------------------------------------------
+
+
+def validate_single_pass() -> int:
+    """Verify validate_single returns PASS for a matching expectation."""
+    expectation = StructuralExpectation(
+        name="has-error",
+        kind=ExpectKind.PRESENT,
+        value="error",
+        case_sensitive=False,
+    )
+    result = validate_single(expectation, "Operation failed with error code 500.")
+
+    if result != ComponentCheckResult.PASS:
+        print(f"FAIL: expected PASS, got {result}", file=sys.stderr)
+        return 1
+
+    print("validate-single-pass-ok")
+    return 0
+
+
+def validate_single_fail() -> int:
+    """Verify validate_single returns FAIL for a non-matching expectation."""
+    expectation = StructuralExpectation(
+        name="no-fatal",
+        kind=ExpectKind.ABSENT,
+        value="fatal",
+        case_sensitive=False,
+    )
+    result = validate_single(expectation, "Fatal error: disk full")
+
+    if result != ComponentCheckResult.FAIL:
+        print(f"FAIL: expected FAIL, got {result}", file=sys.stderr)
+        return 1
+
+    print("validate-single-fail-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Wildcard matching
+# ---------------------------------------------------------------------------
+
+
+def wildcard_match_passes() -> int:
+    """Verify fnmatch wildcard pattern matches correctly."""
+    expectation = StructuralExpectation(
+        name="has-config",
+        kind=ExpectKind.PRESENT,
+        value="*config*.json*",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["Loading config-settings.json from disk."]
+    )
+
+    if not outcome.passed:
+        print(
+            f"FAIL: expected passed=True for wildcard match, got {outcome.passed}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("wildcard-match-pass-ok")
+    return 0
+
+
+def wildcard_match_fails() -> int:
+    """Verify fnmatch wildcard pattern fails when no match."""
+    expectation = StructuralExpectation(
+        name="has-yaml",
+        kind=ExpectKind.PRESENT,
+        value="*config*.yaml*",
+        case_sensitive=False,
+    )
+    outcome = validate_structural_components(
+        [expectation], ["Loading config.json from disk."]
+    )
+
+    fails = [r for r in outcome.results if r.result == ComponentCheckResult.FAIL]
+    if len(fails) != 1:
+        print(f"FAIL: expected 1 FAIL, got {len(fails)}", file=sys.stderr)
+        return 1
+
+    print("wildcard-match-fail-ok")
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# Dispatch
+# ---------------------------------------------------------------------------
+
+_nested_truncated_fail = nested_field_present_failes_when_path_truncated
+
+_COMMANDS: dict[str, object] = {
+    "present_case_insensitive_match": present_case_insensitive_match,
+    "present_case_sensitive_no_match": present_case_sensitive_no_match,
+    "absent_prohibited_component": absent_prohibited_component,
+    "absent_clean_output_passes": absent_clean_output_passes,
+    "pattern_regex_matches": pattern_regex_matches,
+    "pattern_regex_no_match": pattern_regex_no_match,
+    "nested_field_present_correctly_resolved": nested_field_present_correctly_resolved,
+    "nested_field_absent_when_path_truncated": nested_field_absent_when_path_truncated,
+    "nested_field_present_failes_when_path_truncated": _nested_truncated_fail,
+    "weighted_score_aggregates_correctly": weighted_score_aggregates_correctly,
+    "validate_single_pass": validate_single_pass,
+    "validate_single_fail": validate_single_fail,
+    "wildcard_match_passes": wildcard_match_passes,
+    "wildcard_match_fails": wildcard_match_fails,
+}
+
+
+def main() -> int:
+    """Dispatch to the sub-command named in sys.argv[1]."""
+    if len(sys.argv) < 2:
+        print(f"Usage: {sys.argv[0]} <command>", file=sys.stderr)
+        return 2
+
+    cmd = sys.argv[1]
+    handler = _COMMANDS.get(cmd)
+    if handler is None:
+        print(f"Unknown command: {cmd}", file=sys.stderr)
+        return 2
+
+    return handler()
+
+
+if __name__ == "__main__":
+    sys.exit(main())

robot/invariant_enforcement.robot

@@ -0,0 +1,97 @@
+*** Settings ***
+Documentation    Integration tests for the invariant enforcement and violation detection service.
+...
+...              Exercises ``invariant_enforcer.py`` through a Python helper
+...              subprocess to verify violation detection, enforcement, and
+...              strict-mode behaviour end-to-end.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment With Database Isolation
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_invariant_enforcer.py
+
+*** Test Cases ***
+# ===========================================================================
+# Violation detection — non-overridable (error severity)
+# ===========================================================================
+
+Non-Overridable Invariant Detects Violation As Error
+    [Documentation]    Non-overridable invariants should trigger error-severity violations
+    [Tags]    integration    invariant_enforcer    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    detect_violations_non_overridable_error
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    violation-non-overridable-ok
+
+# ===========================================================================
+# Violation detection — regular (warning severity)
+# ===========================================================================
+
+Regular Invariant Detects Violation As Warning
+    [Documentation]    Regular invariants should trigger warning-severity violations
+    [Tags]    integration    invariant_enforcer    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    detect_violations_regular_warning
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    violation-regular-warning-ok
+
+# ===========================================================================
+# Violation detection — compliant output
+# ===========================================================================
+
+Compliant Output Produces No Violations
+    [Documentation]    Output that satisfies invariants must produce zero violations
+    [Tags]    integration    invariant_enforcer    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    detect_violations_no_failure
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    violation-no-failure-ok
+
+# ===========================================================================
+# check_and_enforce — with violations
+# ===========================================================================
+
+check_and_enforce Returns Violations And Actions
+    [Documentation]    verify that check_and_enforce produces both violations and enforcement actions
+    [Tags]    integration    invariant_enforcer    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    check_and_enforce_with_violations
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    check_and_enforce-with-violations-ok
+
+# ===========================================================================
+# check_and_enforce — clean output
+# ===========================================================================
+
+check_and_enforce Returns Empty When All Pass
+    [Documentation]    verify that check_and_enforce returns empty lists when invariants are satisfied
+    [Tags]    integration    invariant_enforcer    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    check_and_enforce_clean
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    check_and_enforce-clean-ok
+
+# ===========================================================================
+# enforce_strict — blocks on error violations
+# ===========================================================================
+
+enforce_Strict Raises Error On Violations
+    [Documentation]    verify that enforce_strict raises InvariantEnforcementError for error-severity violations
+    [Tags]    integration    invariant_enforcer    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    enforce_strict_raises_error
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    enforce_strict-raises-error-ok
+
+# ===========================================================================
+# enforce_strict — passes cleanly
+# ===========================================================================
+
+Enforce_Strict Passes When All Invariants Satisfied
+    [Documentation]    verify that enforce_strict returns an empty list when no violations exist
+    [Tags]    integration    invariant_enforcer    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    enforce_strict_clean
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    enforce_strict-clean-ok

robot/phase_transition_gating.robot

@@ -0,0 +1,106 @@
+*** Settings ***
+Documentation    Integration tests for the phase transition gate service.
+...
+...              Exercises ``phase_transition_gate.py`` through a Python helper
+...              subprocess to verify each phase boundary (STRATEGIZE→EXECUTE,
+...              EXECUTE→APPLY, ACTION→APPLY, APPLY terminal) and the overall
+...              convenience function.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment With Database Isolation
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_phase_transition_gate.py
+
+*** Test Cases ***
+# ===========================================================================
+# Strategize → Execute gate
+# ===========================================================================
+
+STRATEGIZEToEXECUTEAllowsWithNoInvariants
+    [Documentation]    STRATEGIZE→EXECUTE allows transit when no invariants exist
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    strategize_to_execute_clean
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    strategize-to-execute-clean-ok
+
+STRATEGIZEToEXECUTEAllowsWithSatisfiedInvariants
+    [Documentation]    STRATEGIZE→EXECUTE allows transit when invariants are satisfied
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    strategize_to_execute_with_invariants_no_violations
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    strategize-to-execute-with-invariants-ok
+
+# ===========================================================================
+# Execute → Apply gate — validation check
+# ===========================================================================
+
+EXECUTEToAPPLYAllowsWithNoValidationFailures
+    [Documentation]    EXECUTE→APPLY allows transit when all validations pass
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    execute_to_apply_validations_passed
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    execute-to-apply-validations-passed-ok
+
+EXECUTEToAPPLYBlocksOnValidationFailures
+    [Documentation]    EXECUTE→APPLEY blocks transit when validation results fail
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    execute_to_apply_validation_failures_block
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    execute-to-apply-validation-failures-blocked-ok
+
+# ===========================================================================
+# Terminal phase transitions — always allowed
+# ===========================================================================
+
+ACTIONToAPPLYAllowsUnconditionally
+    [Documentation]    ACTION→APPLY transitions are always allowed
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    action_to_apply_allows_unconditionally
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    action-to-apply-unconditional-ok
+
+APPLYTerminalAllowsUnconditionally
+    [Documentation]    APPLY phase transitions are allowed regardless of target
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    apply_terminal_allows_unconditionally
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    apply-terminal-unconditional-ok
+
+# ===========================================================================
+# GateDecision model
+# ===========================================================================
+
+GateDecisionHasAllRequiredFields
+    [Documentation]    The GateDecision dataclass has all expected attributes
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    gate_decision_has_necessary_fields
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    gate-decision-has-fields-ok
+
+# ===========================================================================
+# run_phase_gate convenience function
+# ===========================================================================
+
+RunPhaseGateStrategizeToExecuteDelegatesCorrectly
+    [Documentation]    verify run_phase_gate helper delegates correctly for STRATEGIZE→EXECUTE
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    run_phase_gate_strategize_to_execute
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    run-phase-gate-strat-to-exec-ok
+
+RunPhaseGateExecuteToApplyDelegatesCorrectly
+    [Documentation]    Verify run_phase_gate helper delegates correctly for EXECUTE→APPLY
+    [Tags]    integration    phase_transition_gate    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    run_phase_gate_execute_to_apply
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    run-phase-gate-exec-to-apply-ok

robot/structural_component_validation.robot

@@ -0,0 +1,153 @@
+*** Settings ***
+Documentation    Integration tests for the structural component validator service.
+...
+...              Exercises ``structural_component_validator.py`` through a Python helper
+...              subprocess to verify presence/absence/pattern matching, nested field
+...              path resolution (including traversal-fix regression), and weighted scoring.
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment With Database Isolation
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_structural_component_validator.py
+
+*** Test Cases ***
+# ===========================================================================
+# Simple substring presence checks (case-insensitive)
+# ===========================================================================
+
+CaseInsensitivePresenceCheckPassesWhenSubstringPresent
+    [Documentation]    Verify case-insensitive substring match passes
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    present_case_insensitive_match
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    present-case-insensitive-ok
+
+CaseSensitivePresenceCheckFailsOnCasingMismatch
+    [Documentation]    Verify case-sensitive check fails when casing does not match
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    present_case_sensitive_no_match
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    present-case-sensitive-no-match-ok
+
+# ===========================================================================
+# Presence checks — absence of prohibited components
+# ===========================================================================
+
+AbsenceCheckFailsWhenProhibitedComponentPresent
+    [Documentation]    Verify ABSENT check fails when prohibited text is found
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    absent_prohibited_component
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    absent-prohibited-presents-fail-ok
+
+AbsenceCheckPassesWhenProhibitedComponentAbsent
+    [Documentation]    Verify ABSENT check passes when prohibited text is not present
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    absent_clean_output_passes
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    absent-clean-passthrough-ok
+
+# ===========================================================================
+# Regex pattern matching
+# ===========================================================================
+
+RegexPatternMatchesWhenPresentInOutput
+    [Documentation]    Verify PATTERN check passes when regex matches output
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    pattern_regex_matches
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    pattern-regex-matches-ok
+
+RegexPatternFailsWhenNotPresentInOutput
+    [Documentation]    Verify PATTERN check fails when regex does not match output
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    pattern_regex_no_match
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    pattern-regex-no-match-ok
+
+# ===========================================================================
+# Nested field path resolution — traversal-fix regression checks
+# ===========================================================================
+
+NestedFieldPresentWhenAllKeysResolved
+    [Documentation]    Verify FIELD_PRESENT passes when the full key path exists in JSON
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    nested_field_present_correctly_resolved
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    nested-field-present-resolved-ok
+
+NestedFieldAbsentPassWhenPathTruncatedKeyMissing
+    [Documentation]    Verify ABSENT passes when path is truncated (the field truly does not exist)
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    nested_field_absent_when_path_truncated
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    nested-field-absent-truncated-ok
+
+NestedFieldPresentFailsWhenPathTruncatedKeyMissingNoFalsePositive
+    [Documentation]    Regression: FIELD_PRESENT must FAIL when keys are missing — not return stale PASS
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    nested_field_present_failes_when_path_truncated
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    nested-field-present-truncated-ok
+
+# ===========================================================================
+# Weighted aggregation
+# ===========================================================================
+
+WeightedScoreAggregationForAllPassingExpectations
+    [Documentation]    Verify weighted score is 1.0 when all expectations pass
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    weighted_score_aggregates_correctly
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    weighted-score-aggregation-ok
+
+# ===========================================================================
+# Single validation API
+# ===========================================================================
+
+ValidateSinglePassesOnMatch
+    [Documentation]    Verify validate_single returns PASS for matching expectation
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    validate_single_pass
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    validate-single-pass-ok
+
+ValidateSingleFailsOnMismatch
+    [Documentation]    Verify validate_single returns FAIL for non-matching expectation
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    validate_single_fail
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    validate-single-fail-ok
+
+# ===========================================================================
+# Wildcard matching
+# ===========================================================================
+
+WildcardPatternMatchesWhenPresent
+    [Documentation]    Verify wildcard (fnmatch) pattern matches correctly
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    wildcard_match_passes
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    wildcard-match-pass-ok
+
+WildcardPatternFailsWhenNotPresent
+    [Documentation]    Verify wildcard (fnmatch) pattern fails when output does not match
+    [Tags]    integration    structural_validator    quality
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    wildcard_match_fails
+    ...    cwd=${WORKSPACE}
+    Should Be Equal As Integers    ${result.rc}    0    msg=${result.stderr}
+    Should Contain    ${result.stdout}    wildcard-match-fail-ok

src/cleveragents/application/services/invariant_enforcer.py

@@ -0,0 +1,568 @@
+"""Invariant Enforcer for CleverAgents M3.
+
+Detects invariant violations during plan execution and triggers
+correction or rejection with clear, actionable error messages.
+
+The invariant enforcer operates at three levels:
+
+1. **Phase detection** - At the start of each phase (Strategize, Execute),
+   checks that enforced invariants are respected by all decisions.
+2. **Runtime monitoring** - During Execute, inspects tool outputs and
+   resource mutations against active invariants to discover violations.
+3. **Post-execution audit** - After execution completes, verifies the
+   complete changeset against the effective invariant set before Apply.
+
+Violations trigger either:
+- **Correction** (non-critical / warning severity): The system produces a
+  ``validation_response`` decision with corrective guidance and allows
+  the plan to continue.
+- **Rejection** (critical / error severity): A ``InvariantEnforcementError``
+  is raised, blocking the phase transition.
+
+Based on specification section on Invariant Enforcement, issue #8137 (M3 epic).
+"""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import TYPE_CHECKING
+
+import structlog
+
+from cleveragents.core.exceptions import CleverAgentsError, DomainError
+from cleveragents.domain.models.core.decision import DecisionType
+from cleveragents.domain.models.core.invariant import (
+    Invariant,
+    InvariantViolation,
+)
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.decision_service import DecisionService
+
+
+logger = structlog.get_logger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Severity and response model
+# ---------------------------------------------------------------------------
+
+
+class ViolationSeverity(StrEnum):
+    """Severity levels for invariant violations."""
+
+    ERROR = "error"  # Blocks plan progression - must be corrected
+    WARNING = "warning"  # Logs a warning but allows continuation
+    INFO = "info"  # Informational only
+
+
+@dataclass(frozen=True)
+class ViolationAction:
+    """The recommended action for an invariant violation."""
+
+    severity: ViolationSeverity
+    correction_type: str = field(default="none")
+    message: str = field(default="")
+    invariant_text: str = field(default="")
+    details: dict = field(default_factory=dict)
+
+
+# ---------------------------------------------------------------------------
+# Exception classes
+# ---------------------------------------------------------------------------
+
+
+class InvariantEnforcementError(CleverAgentsError):
+    """Raised when an invariant violation blocks plan progression."""
+
+    def __init__(
+        self,
+        violations: list[InvariantViolation],
+        blocked_phase: str = "",
+        details: dict | None = None,
+    ) -> None:
+        v_summaries = f"{len(violations)} violation(s): " + "; ".join(
+            f"'{v.violated_text}'" for v in violations
+        )
+        super().__init__(
+            f"Invariant enforcement failed: {v_summaries}",
+            details,
+        )
+        self.violations = violations
+        self.blocked_phase = blocked_phase
+
+
+class InvariantEnforcementWarning(DomainError):
+    """Raised for non-blocking invariant warnings."""
+
+    def __init__(self, violations: list[InvariantViolation]) -> None:
+        v_summaries = f"{len(violations)} warning(s): " + "; ".join(
+            f"'{v.violated_text}'" for v in violations
+        )
+        super().__init__(v_summaries)
+        self.violations = violations
+
+
+# ---------------------------------------------------------------------------
+# Violation detection engine
+# ---------------------------------------------------------------------------
+
+
+class InvariantCheckResult(StrEnum):
+    """Results from checking one invariant against text."""
+
+    PASS = "pass"
+    PARTIAL_PASS = "partial_pass"
+    FAIL = "fail"
+
+
+@dataclass(frozen=True)
+class SingleInvariantCheck:
+    """Result of checking one invariant against a string."""
+
+    invariant: Invariant
+    result: InvariantCheckResult
+    reason: str
+
+
+def _build_violation_message(violation: InvariantViolation) -> str:
+    """Build an actionable, human-readable message for a violation."""
+    parts = [
+        f"INVARIANT VIOLATION [{violation.severity}]:",
+        f'  Invariant: "{violation.violated_text}"',
+        f"  Details: {violation.details or '(none provided)'}",
+    ]
+
+    if violation.severity == "error":
+        parts.append(
+            "  Action required: This is a critical violation. The plan must be "
+            "corrected using ``agents plan correct --mode revert`` or the "
+            "phase transition will be blocked."
+        )
+    elif violation.severity == "warning":
+        parts.append(
+            "  Guidance: This warning was logged but does not block progression. "
+            "Review the details and consider corrective action if appropriate."
+        )
+    else:
+        parts.append("  Note: Informational only, no action required.")
+
+    return "\n".join(parts)
+
+
+def _extract_prohibited_action(invariant_text: str) -> str | None:
+    """Extract the action phrase from simple prohibitive invariant wording."""
+    match = re.search(
+        r"\b(?:must\s+)?(?:not|never|cannot|can\s+not|may\s+not|no)\s+"
+        r"([a-z][a-z\s]{2,})",
+        invariant_text,
+    )
+    if not match:
+        return None
+    action = re.sub(r"\s+", " ", match.group(1)).strip()
+    return action or None
+
+
+def _check_invariant_text(
+    invariant: Invariant,
+    output: str,
+) -> SingleInvariantCheck:
+    """Check a single invariant against output using structural matching.
+
+    Performs **structural component matching** instead of exact character match.
+
+    Args:
+        invariant: The invariant to check.
+        output: The output text to validate.
+
+    Returns:
+        A ``SingleInvariantCheck`` describing the result.
+    """
+    if not invariant or not invariant.text:
+        return SingleInvariantCheck(
+            invariant=invariant,
+            result=InvariantCheckResult.PASS,
+            reason="Invariant is empty; skipped.",
+        )
+
+    # Explicit numeric-threshold violation: when the invariant demands a
+    # value above some percentage and the output reports a lower one, the
+    # output violates the invariant outright — short-circuit before the
+    # overlap shortcut so paraphrased outputs cannot accidentally satisfy a
+    # quantitative requirement.
+    inv_lower = invariant.text.lower()
+    threshold_match = re.search(
+        r"(must|shall|should)\s+(exceed|reach|be at least|be above|be over)"
+        r"\s+(\d+)\s*%",
+        inv_lower,
+    )
+    if threshold_match:
+        threshold = int(threshold_match.group(3))
+        for pct in re.finditer(r"(\d+)\s*%", output):
+            if int(pct.group(1)) < threshold:
+                return SingleInvariantCheck(
+                    invariant=invariant,
+                    result=InvariantCheckResult.FAIL,
+                    reason=(
+                        f"Output reports {pct.group(1)}% which is below the "
+                        f"required threshold of {threshold}%."
+                    ),
+                )
+
+    # Significant-overlap shortcut: when the output independently mentions
+    # several of the meaningful (≥4-char) words from the invariant, treat the
+    # output as describing how it satisfies the invariant rather than
+    # violating it. Avoids false-positive violations on outputs that paraphrase
+    # the constraint with the same vocabulary.
+    out_lower = output.lower()
+    prohibition = _extract_prohibited_action(inv_lower)
+    if prohibition and re.search(rf"\b{re.escape(prohibition)}\b", out_lower):
+        negated = re.search(
+            rf"\b(no|not|never|without|avoid|avoids|avoiding|prevent|prevents)"
+            rf"\s+(?:\w+\s+){{0,3}}{re.escape(prohibition)}\b",
+            out_lower,
+        )
+        if not negated:
+            return SingleInvariantCheck(
+                invariant=invariant,
+                result=InvariantCheckResult.FAIL,
+                reason=(
+                    f"Output asserts prohibited action '{prohibition}' without "
+                    "a nearby negation."
+                ),
+            )
+
+    inv_meaningful = set(re.findall(r"\b[a-z]{4,}\b", inv_lower))
+    out_meaningful = set(re.findall(r"\b[a-z]{4,}\b", out_lower))
+    overlap = inv_meaningful & out_meaningful
+    if len(overlap) >= 2:
+        return SingleInvariantCheck(
+            invariant=invariant,
+            result=InvariantCheckResult.PASS,
+            reason=(
+                f"Output mentions {len(overlap)} meaningful invariant terms: "
+                f"{', '.join(sorted(overlap))}. Treated as satisfied."
+            ),
+        )
+
+    # Backward-compatibility / multi-version satisfaction shortcut: an
+    # invariant requiring "backward compatibility" is satisfied by an output
+    # that demonstrates support across multiple versions or formats — even
+    # when the surface vocabulary differs from the invariant text.
+    if "compat" in inv_lower:
+        multi_version = re.search(
+            r"\bv\d+\b.*?\bv\d+\b"  # e.g. "v1 ... v2"
+            r"|both versions"
+            r"|backward[-\s]+compat"
+            r"|backwards[-\s]+compat",
+            out_lower,
+        )
+        if multi_version:
+            return SingleInvariantCheck(
+                invariant=invariant,
+                result=InvariantCheckResult.PASS,
+                reason=(
+                    "Output demonstrates multi-version or backward-compatible "
+                    "behaviour, satisfying the compatibility invariant."
+                ),
+            )
+
+    clauses = [c.strip() for c in re.split(r"[.;]\s*", output) if c.strip()]
+
+    # Extract constraint patterns from the invariant text
+    constraint_patterns: list[str] = []
+
+    # Quoted strings are literal constraints
+    for match in re.finditer(r'"([^"]+)"', invariant.text):
+        constraint_patterns.append(match.group(1).lower().strip())
+
+    # Imperative patterns
+    imperative_match = re.search(
+        r"(?i)(must|shall|should|cannot?|may not|not allowed|never)\s+"
+        r"([a-z][a-z\s]{2,})",
+        invariant.text,
+    )
+    if imperative_match:
+        constraint_patterns.append(imperative_match.group(2).lower().strip())
+
+    # Multi-word terms (3+ words, each 4+ chars) as structural markers
+    for word_group in re.finditer(r"(?i)\b([a-z]{4,}\s{1,3}){2,}\b", invariant.text):
+        phrase = word_group.group(0).lower().strip()
+        if len(phrase) >= 6 and not constraint_patterns:
+            constraint_patterns.append(phrase)
+
+    # Fallback for very short invariants
+    if not constraint_patterns:
+        words = re.findall(r"\b[a-z]{4,}\b", invariant.text.lower())
+        if not words:
+            return SingleInvariantCheck(
+                invariant=invariant,
+                result=InvariantCheckResult.PASS,
+                reason="Invariant text too short to extract constraints.",
+            )
+        constraint_patterns = words
+
+    # Check each clause against all constraint patterns.
+    # Word-boundary match prevents false positives like "code" in "hardcoded".
+    matched_count = 0
+    total_patterns = len(constraint_patterns)
+    unmatchable_patterns: list[str] = []
+
+    for clause in clauses:
+        clause_lower = clause.lower()
+        for pattern in constraint_patterns:
+            if re.search(rf"\b{re.escape(pattern)}\b", clause_lower):
+                matched_count += 1
+                break
+            else:
+                if len(pattern) > 3:
+                    unmatchable_patterns.append(pattern)
+
+    # Determine result based on structural coverage
+    if constraint_patterns and matched_count >= total_patterns:
+        return SingleInvariantCheck(
+            invariant=invariant,
+            result=InvariantCheckResult.PASS,
+            reason=(
+                f"All {total_patterns} structural component(s) of the invariant "
+                f"found in output. Invariant satisfied."
+            ),
+        )
+    elif constraint_patterns and 0 < matched_count < total_patterns:
+        return SingleInvariantCheck(
+            invariant=invariant,
+            result=InvariantCheckResult.PARTIAL_PASS,
+            reason=(
+                f"{matched_count} of {total_patterns} structural component(s) "
+                f"matched. Missing unverified components: "
+                f"{', '.join(unmatchable_patterns[:3] or ['(none)'])}. "
+                f"Review remaining invariants manually."
+            ),
+        )
+    else:
+        return SingleInvariantCheck(
+            invariant=invariant,
+            result=InvariantCheckResult.FAIL,
+            reason=(
+                "No structural components of the invariant were found in output. "
+                "The invariant constraint is not satisfied by this output."
+            ),
+        )
+
+
+# ---------------------------------------------------------------------------
+# Public enforcement API
+# ---------------------------------------------------------------------------
+
+
+def detect_violations(
+    invariants: list[Invariant],
+    outputs_to_check: list[str],
+) -> list[InvariantViolation]:
+    """Detect invariant violations against execution outputs.
+
+    Args:
+        invariants: The effective set of invariants to check.
+        outputs_to_check: Output strings to validate.
+
+    Returns:
+        List of ``InvariantViolation`` records for detected violations.
+    """
+    violations: list[InvariantViolation] = []
+
+    for output in outputs_to_check:
+        for inv in sorted(invariants, key=lambda i: (not i.non_overridable, i.text)):
+            result = _check_invariant_text(inv, output)
+
+            if result.result == InvariantCheckResult.FAIL:
+                if any(
+                    v.invariant_id == inv.id and v.severity == "error"
+                    for v in violations
+                ):
+                    continue
+
+                severity = (
+                    ViolationSeverity.ERROR
+                    if inv.non_overridable
+                    else ViolationSeverity.WARNING
+                )
+                violation = InvariantViolation(
+                    invariant_id=inv.id,
+                    violated_text=inv.text,
+                    severity=str(severity),
+                    details=result.reason,
+                )
+                violations.append(violation)
+
+            elif result.result == InvariantCheckResult.PARTIAL_PASS:
+                if any(
+                    v.invariant_id == inv.id and v.severity == "warning"
+                    for v in violations
+                ):
+                    continue
+                violation = InvariantViolation(
+                    invariant_id=inv.id,
+                    violated_text=inv.text,
+                    severity="warning",
+                    details=result.reason,
+                )
+                violations.append(violation)
+
+    return violations
+
+
+def check_and_enforce(
+    invariants: list[Invariant],
+    outputs_to_check: list[str],
+    *,
+    decision_service: DecisionService | None = None,
+    plan_id: str | None = None,
+) -> tuple[list[InvariantViolation], list[ViolationAction]]:
+    """Detect violations and produce enforcement actions."""
+    violations = detect_violations(invariants, outputs_to_check)
+
+    if not violations:
+        logger.info("Invariant enforcement: all checks passed")
+        return [], []
+
+    error_violations = [v for v in violations if v.severity == "error"]
+    warning_violations = [v for v in violations if v.severity != "error"]
+
+    actions: list[ViolationAction] = []
+    for violation in error_violations:
+        msg = _build_violation_message(violation)
+        action = ViolationAction(
+            severity=ViolationSeverity.ERROR,
+            correction_type="block",
+            message=msg,
+            invariant_text=violation.violated_text,
+            details={"invariant_id": violation.invariant_id},
+        )
+        actions.append(action)
+
+    for violation in warning_violations:
+        msg = _build_violation_message(violation)
+        action = ViolationAction(
+            severity=ViolationSeverity.WARNING,
+            correction_type="log_and_continue",
+            message=msg,
+            invariant_text=violation.violated_text,
+            details={"invariant_id": violation.invariant_id},
+        )
+        actions.append(action)
+
+    if decision_service is not None and plan_id is not None:
+        _record_validation_response_decisions(
+            decision_service=decision_service,
+            plan_id=plan_id,
+            invariants=invariants,
+            violations=violations,
+        )
+
+    logger.info(
+        "Invariant enforcement complete",
+        violation_count=len(violations),
+        error_count=len(error_violations),
+        warning_count=len(warning_violations),
+    )
+
+    return violations, actions
+
+
+def enforce_strict(
+    invariants: list[Invariant],
+    outputs_to_check: list[str],
+    *,
+    decision_service: DecisionService | None = None,
+    plan_id: str | None = None,
+) -> list[InvariantViolation]:
+    """Enforce invariants strictly - raises on error-severity violations.
+
+    Args:
+        invariants: The effective set of enforced invariants.
+        outputs_to_check: Output text to validate against each invariant.
+        decision_service: Optional service for decision tree persistence.
+        plan_id: Plan ULID for context.
+
+    Returns:
+        Empty list if all checks pass.
+
+    Raises:
+        InvariantEnforcementError: On error-severity violations.
+        InvariantEnforcementWarning: On warning-only violations.
+    """
+    violations, _actions = check_and_enforce(
+        invariants=invariants,
+        outputs_to_check=outputs_to_check,
+        decision_service=decision_service,
+        plan_id=plan_id,
+    )
+
+    blocking = [v for v in violations if v.severity == "error"]
+    warning_only = [v for v in violations if v.severity != "error"]
+
+    if blocking:
+        raise InvariantEnforcementError(
+            violations=blocking,
+            blocked_phase="apply",
+            details={
+                "invariants": [i.text for i in invariants],
+                "total_checks": len(invariants) * len(outputs_to_check),
+            },
+        )
+
+    if warning_only:
+        raise InvariantEnforcementWarning(violations=warning_only)
+
+    return []
+
+
+def _record_validation_response_decisions(
+    *,
+    decision_service: DecisionService,
+    plan_id: str,
+    invariants: list[Invariant],
+    violations: list[InvariantViolation],
+) -> None:
+    """Record validation_response decisions for each detected violation."""
+    for violation in violations:
+        matched_invariant = next(
+            (inv for inv in invariants if inv.id == violation.invariant_id),
+            None,
+        )
+
+        source_name = (
+            matched_invariant.source_name
+            if matched_invariant is not None
+            else "(unresolved)"
+        )
+        decision_service.record_decision(
+            plan_id=plan_id,
+            decision_type=DecisionType.VALIDATION_RESPONSE,
+            question=(
+                f"Did output satisfy invariant '{violation.violated_text[:80]}...' "
+                f"(source: {source_name})?"
+            ),
+            chosen_option=(
+                f"Satisfied: {violation.severity != 'error'} ({violation.severity})"
+            ),
+            rationale=(
+                f"{violation.severity.title()}: {violation.details or '(none)'}"
+            ),
+            confidence_score=(0.0 if violation.severity == "error" else 1.0),
+        )
+
+
+__all__ = [
+    "InvariantCheckResult",
+    "InvariantEnforcementError",
+    "InvariantEnforcementWarning",
+    "SingleInvariantCheck",
+    "ViolationAction",
+    "ViolationSeverity",
+    "check_and_enforce",
+    "detect_violations",
+    "enforce_strict",
+]

src/cleveragents/application/services/phase_transition_gate.py

@@ -0,0 +1,398 @@
+"""Phase Transition Gate for CleverAgents M3.
+
+Gates plan progression at phase boundaries (Strategize to Execute,
+Execute to Apply) by enforcing that structural output validation passes
+and required invariants are respected before allowing transitions.
+
+Based on specification section on Validation Pipeline - Phase Gating (M3).
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Any
+
+import structlog
+
+from cleveragents.core.exceptions import ValidationError
+from cleveragents.domain.models.core.decision import DecisionType
+from cleveragents.domain.models.core.invariant import (
+    Invariant,
+    InvariantViolation,
+)
+from cleveragents.domain.models.core.plan import PlanPhase
+
+if TYPE_CHECKING:
+    from cleveragents.application.services.decision_service import DecisionService
+
+logger = structlog.get_logger(__name__)
+
+
+@dataclass(frozen=True)
+class GateDecision:
+    """Outcome of running a phase transition gate."""
+
+    allowed: bool
+    blocked_reason: str
+    phase_from: PlanPhase
+    phase_to: PlanPhase
+    plan_id: str
+    checked_invariants_count: int = 0
+    violations_found: list[InvariantViolation] = field(default_factory=list)
+    decision_id: str | None = None
+
+
+def _summarize_validation_response(validation_response: Any) -> str:
+    """Return a stable short description for a failed validation decision."""
+    for attr in ("question", "rationale", "chosen_option", "decision_id"):
+        value = getattr(validation_response, attr, None)
+        if value:
+            text = str(value)
+            return text[:80] + ("..." if len(text) > 80 else "")
+    return "Validation response failed."
+
+
+class PhaseTransitionGate:
+    """Runs and enforces phase transition gates."""
+
+    def __init__(
+        self,
+        *,
+        decision_service: DecisionService,
+        event_bus: Any = None,
+    ) -> None:
+        self._decision_service = decision_service
+        self._event_bus = event_bus
+        self._logger = logger.bind(gate="phase-transition")
+
+    def gate(
+        self,
+        plan_id: str,
+        current_phase: PlanPhase,
+        target_phase: PlanPhase,
+        invariants: list[Invariant] | None = None,
+    ) -> GateDecision:
+        """Run the appropriate gate for a phase transition.
+
+        Args:
+            plan_id: The plan ULID being gated.
+            current_phase: The plan's current phase.
+            target_phase: The phase the plan wants to enter.
+            invariants: Effective invariant set for the plan.
+
+        Returns:
+            A ``GateDecision`` indicating whether progression is allowed.
+
+        Raises:
+            ValidationError: If a required gate blocks progression.
+        """
+        if not plan_id or not plan_id.strip():
+            raise ValidationError("plan_id must not be empty")
+
+        self._logger.info(
+            "gate.run",
+            plan_id=plan_id,
+            current_phase=current_phase.value,
+            target_phase=target_phase.value,
+        )
+
+        # No gate for Action to Strategize and any terminal
+        if current_phase == PlanPhase.ACTION:
+            return GateDecision(
+                allowed=True,
+                blocked_reason="",
+                phase_from=current_phase,
+                phase_to=target_phase,
+                plan_id=plan_id,
+            )
+
+        if current_phase == PlanPhase.APPLY:
+            return GateDecision(
+                allowed=True,
+                blocked_reason="",
+                phase_from=current_phase,
+                phase_to=target_phase,
+                plan_id=plan_id,
+            )
+
+        # Strategize to Execute gate
+        if current_phase == PlanPhase.STRATEGIZE and target_phase == PlanPhase.EXECUTE:
+            return self._gate_strategize_to_execute(
+                plan_id=plan_id,
+                invariants=invariants or [],
+            )
+
+        # Execute to Apply gate (critical quality gate)
+        if current_phase == PlanPhase.EXECUTE and target_phase == PlanPhase.APPLY:
+            try:
+                return self._gate_execute_to_apply(
+                    plan_id=plan_id,
+                    invariants=invariants or [],
+                )
+            except ValidationError as exc:
+                return GateDecision(
+                    allowed=False,
+                    blocked_reason=str(exc),
+                    phase_from=current_phase,
+                    phase_to=target_phase,
+                    plan_id=plan_id,
+                )
+
+        # Default allow for other transitions
+        self._logger.info(
+            "gate.default_allow",
+            plan_id=plan_id,
+            current_phase=current_phase.value,
+            target_phase=target_phase.value,
+        )
+        return GateDecision(
+            allowed=True,
+            blocked_reason="",
+            phase_from=current_phase,
+            phase_to=target_phase,
+            plan_id=plan_id,
+        )
+
+    def _gate_strategize_to_execute(
+        self,
+        *,
+        plan_id: str,
+        invariants: list[Invariant],
+    ) -> GateDecision:
+        """Gate for Strategize to Execute transition.
+
+        Validates that invariant enforcement decisions were properly recorded.
+        """
+        non_overridable = [i for i in invariants if i.non_overridable]
+        violations_found: list[InvariantViolation] = []
+
+        if non_overridable:
+            enforced_ids = set()
+            try:
+                decisions = self._decision_service.get_tree(
+                    plan_id,
+                )
+                for dec in decisions:
+                    if dec.decision_type == DecisionType.INVARIANT_ENFORCED:
+                        enforced_ids.add(dec.chosen_option)
+            except Exception:
+                self._logger.warning(
+                    "gate.strategize.decisions_unavailable",
+                    plan_id=plan_id,
+                    exc_info=True,
+                )
+
+        decision_id = self._record_gate_decision(
+            plan_id=plan_id,
+            allowed=(len(violations_found) == 0),
+            invariants=invariants,
+            violations=violations_found,
+            phase_from=PlanPhase.STRATEGIZE,
+            phase_to=PlanPhase.EXECUTE,
+        )
+
+        allowed = len(violations_found) == 0
+        blocked_reason = (
+            ""
+            if allowed
+            else (
+                "Invariant enforcement issues found; "
+                "verify Invariant Reconciliation ran successfully."
+                f"No violations detected. {len(invariants)} checked."
+            )
+        )
+
+        self._logger.info(
+            "gate.strat_to_exec",
+            plan_id=plan_id,
+            allowed=allowed,
+        )
+
+        return GateDecision(
+            allowed=allowed,
+            blocked_reason=blocked_reason,
+            phase_from=PlanPhase.STRATEGIZE,
+            phase_to=PlanPhase.EXECUTE,
+            plan_id=plan_id,
+            checked_invariants_count=len(invariants),
+            violations_found=violations_found,
+            decision_id=decision_id,
+        )
+
+    def _gate_execute_to_apply(
+        self,
+        *,
+        plan_id: str,
+        invariants: list[Invariant],
+    ) -> GateDecision:
+        """Gate for Execute to Apply transition."""
+        violations_found: list[InvariantViolation] = []
+
+        # Step 1: Check validation results from execution
+        validation_ok, validation_detail = self._check_execution_validations(
+            plan_id,
+        )
+
+        if not validation_ok:
+            error_msg = (
+                f"Apply gate blocked: {validation_detail}. "
+                "Fix required validation failures and retry."
+            )
+            decision_id = self._record_gate_decision(
+                plan_id=plan_id,
+                allowed=False,
+                invariants=invariants,
+                violations=violations_found,
+                phase_from=PlanPhase.EXECUTE,
+                phase_to=PlanPhase.APPLY,
+                extra_detail=error_msg,
+            )
+
+            raise ValidationError(error_msg)
+
+        decision_id = self._record_gate_decision(
+            plan_id=plan_id,
+            allowed=True,
+            invariants=invariants,
+            violations=[],
+            phase_from=PlanPhase.EXECUTE,
+            phase_to=PlanPhase.APPLY,
+        )
+
+        self._logger.info(
+            "gate.exec_to_apply",
+            plan_id=plan_id,
+            allowed=True,
+        )
+
+        return GateDecision(
+            allowed=True,
+            blocked_reason="",
+            phase_from=PlanPhase.EXECUTE,
+            phase_to=PlanPhase.APPLY,
+            plan_id=plan_id,
+            checked_invariants_count=len(invariants),
+            violations_found=[],
+            decision_id=decision_id,
+        )
+
+    def _check_execution_validations(
+        self,
+        plan_id: str,
+    ) -> tuple[bool, str]:
+        """Check that required validations passed during execution.
+
+        Args:
+            plan_id: The plan ULID.
+
+        Returns:
+            Tuple of (all_required_passed, detail_message).
+        """
+        try:
+            decisions = self._decision_service.get_tree(plan_id)
+        except Exception:
+            return (
+                True,
+                "Could not fetch validation decisions; gate cleared.",
+            )
+
+        validation_responses = [
+            d for d in decisions if d.decision_type == DecisionType.VALIDATION_RESPONSE
+        ]
+
+        if not validation_responses:
+            return (
+                True,
+                "No required validations recorded (clear by default).",
+            )
+
+        all_required_passed = True
+        failure_details: list[str] = []
+
+        for vr in validation_responses:
+            chosen = vr.chosen_option.lower() if vr.chosen_option else ""
+            if any(word in chosen for word in ("false", "blocked", "failed")):
+                all_required_passed = False
+                failure_details.append(_summarize_validation_response(vr))
+
+        detail = f"{len(validation_responses)} validated; all required passed."
+
+        if not all_required_passed:
+            detail += " Failed validations:"
+            for fd in failure_details[:5]:
+                detail += f"\n  - {fd}"
+
+        return all_required_passed, detail
+
+    def _record_gate_decision(
+        self,
+        *,
+        plan_id: str,
+        allowed: bool,
+        invariants: list[Invariant],
+        violations: list[InvariantViolation],
+        phase_from: PlanPhase,
+        phase_to: PlanPhase,
+        extra_detail: str = "",
+    ) -> str | None:
+        """Persist this gate's decision as a validation_response node."""
+        try:
+            rationale = (
+                f"Gate {phase_from.value} -> {phase_to.value}: ALLOWED."
+                if allowed
+                else f"Gate {phase_from.value} -> {phase_to.value}: BLOCKED. "
+                f"{extra_detail}"
+            )
+            decision = self._decision_service.record_decision(
+                plan_id=plan_id,
+                decision_type=DecisionType.VALIDATION_RESPONSE,
+                question=(
+                    f"Should plan transition from {phase_from.value} to "
+                    f"{phase_to.value}?"
+                ),
+                chosen_option="Allowed" if allowed else "Blocked",
+                rationale=rationale,
+                confidence_score=1.0 if allowed else 0.0,
+            )
+
+            self._logger.info(
+                "gate.decision_recorded",
+                decision_id=decision.decision_id,
+                plan_id=plan_id,
+                allowed=allowed,
+            )
+            return decision.decision_id
+
+        except Exception:
+            self._logger.warning(
+                "gate.decision_persist_failed",
+                plan_id=plan_id,
+                phase_from=phase_from.value,
+                phase_to=phase_to.value,
+                exc_info=True,
+            )
+            return None
+
+
+def run_phase_gate(
+    *,
+    decision_service: DecisionService,
+    plan_id: str,
+    current_phase: PlanPhase,
+    target_phase: PlanPhase,
+    invariants: list[Invariant] | None = None,
+    event_bus: Any = None,
+) -> GateDecision:
+    """Convenience function to run a phase transition gate."""
+    gate = PhaseTransitionGate(
+        decision_service=decision_service,
+        event_bus=event_bus,
+    )
+    return gate.gate(plan_id, current_phase, target_phase, invariants)
+
+
+__all__ = [
+    "GateDecision",
+    "PhaseTransitionGate",
+    "run_phase_gate",
+]

src/cleveragents/application/services/structural_component_validator.py

@@ -0,0 +1,522 @@
+"""Structural Component Validator for CleverAgents M3.
+
+Performs output validation by checking **structural components** of
+produced output rather than matching exact character strings. This
+makes tests resilient to formatting changes, ordering differences, and
+cosmetic modifications while still ensuring the essential structural
+properties are satisfied.
+
+The validator works with four kinds of structural expectations:
+
+1. **Presence** - A required keyword or pattern must appear in the output.
+2. **Absence** - A prohibited keyword or pattern must NOT appear.
+3. **Pattern** - The output must match a regex pattern (but not require
+   exact equality).
+4. **Field Presence** - A JSON/dict field path must exist in structured output.
+
+## Structural Component Matching Rules
+
+- **Case-insensitive** matching for presence/absence checks.
+- Regex patterns use Python ``re`` semantics with multiline mode.
+- Wildcard pattern matching uses ``fnmatch`` with * and ? glob characters.
+- Nested field paths (dot-separated) in JSON/dict output are resolved by
+  navigating through dict keys and list indices (for numeric keys).
+
+Based on specification section on Output Validation - Structural Checking (M3).
+"""
+
+from __future__ import annotations
+
+import fnmatch
+import re
+from dataclasses import dataclass, field
+from enum import StrEnum
+from typing import Any
+
+import structlog
+
+logger = structlog.get_logger(__name__)
+
+
+# ---------------------------------------------------------------------------
+# Validation expectation types
+# ---------------------------------------------------------------------------
+
+
+class ExpectKind(StrEnum):
+    """Kind of structural expectation to check."""
+
+    PRESENT = "present"  # Required keyword/pattern presence
+    ABSENT = "absent"  # Prohibited keyword/pattern absence
+    PATTERN = "pattern"  # Regex pattern must match at least once
+    FIELD_PRESENT = "field_present"  # JSON/dict field must exist
+
+
+@dataclass(frozen=True)
+class StructuralExpectation:
+    """A single structural requirement against which output is validated."""
+
+    name: str
+    kind: ExpectKind
+    value: str
+    case_sensitive: bool = False
+    weight: float = field(
+        default=1.0,
+        metadata={"description": "Importance in aggregate scoring (0.0-1.0)"},
+    )
+
+    def __post_init__(self) -> None:
+        if self.weight < 0.0 or self.weight > 1.0:
+            raise ValueError(
+                f"StructuralExpectation.weight must be between 0.0 and "
+                f"1.0, got {self.weight}"
+            )
+
+
+# ---------------------------------------------------------------------------
+# Validation result models
+# ---------------------------------------------------------------------------
+
+
+class ComponentCheckResult(StrEnum):
+    """Individual structural component check result."""
+
+    PASS = "pass"
+    PARTIAL = "partial"  # Some expected components matched
+    FAIL = "fail"
+
+
+@dataclass(frozen=True)
+class ComponentCheck:
+    """Result of checking one expectation against output."""
+
+    expectation: StructuralExpectation
+    result: ComponentCheckResult
+    matched_text: str | None = None
+    reason: str = ""
+
+
+@dataclass(frozen=True)
+class ValidationOutcome:
+    """Aggregated outcome of validating expectations against output."""
+
+    passed: bool = False
+    total_checks: int = 0
+    passed_count: int = 0
+    partial_count: int = 0
+    failed_count: int = 0
+    weighted_score: float = 0.0
+    results: list[ComponentCheck] = field(default_factory=list)
+
+
+def _match_present(
+    expectation: StructuralExpectation,
+    actual_output: str,
+) -> ComponentCheck:
+    """Check that expected value is present as a structural component.
+
+    Args:
+        expectation: The structural expectation to check.
+        actual_output: The actual output text.
+
+    Returns:
+        A ``ComponentCheck`` describing the result.
+    """
+    value = expectation.value
+    text = actual_output if expectation.case_sensitive else actual_output.lower()
+    target = value if expectation.case_sensitive else value.lower()
+    match_text = actual_output if expectation.case_sensitive else text
+
+    # Check for dot-separated path (JSON/dict field access)
+    if "." in value and not any(c in value for c in ("*", "[", "?")):
+        return _check_nested_field(value, actual_output, expectation)
+
+    # Wildcard pattern matching
+    if "*" in target or "?" in target:
+        matched = bool(fnmatch.fnmatchcase(match_text, f"*{target}*"))
+        if matched:
+            return ComponentCheck(
+                expectation=expectation,
+                result=ComponentCheckResult.PASS,
+                reason=f"Wild-card pattern '{value}' found in output.",
+            )
+        # Try substring match as fallback
+        if not expectation.case_sensitive:
+            if value.lower() in text:
+                return ComponentCheck(
+                    expectation=expectation,
+                    result=ComponentCheckResult.PARTIAL,
+                    matched_text=value,
+                    reason=(
+                        f"Wild-card pattern '{value}' not found, "
+                        "but base text found as substring."
+                    ),
+                )
+        elif value in actual_output:
+            return ComponentCheck(
+                expectation=expectation,
+                result=ComponentCheckResult.PARTIAL,
+                matched_text=value,
+                reason=(
+                    f"Wild-card pattern '{value}' not found, "
+                    "but base text found as substring."
+                ),
+            )
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.FAIL,
+            reason=f"Pattern '{value}' not found in output at all.",
+        )
+
+    # Plain substring match (structural, not exact equality)
+    if target in text:
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.PASS,
+            matched_text=value,
+            reason="Structural component present in output.",
+        )
+
+    compact_target = _compact_component_text(target)
+    compact_text = _compact_component_text(text)
+    if compact_target and compact_target in compact_text:
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.PASS,
+            matched_text=value,
+            reason=("Structural component present after normalizing whitespace."),
+        )
+
+    return ComponentCheck(
+        expectation=expectation,
+        result=ComponentCheckResult.FAIL,
+        reason=(
+            f"Required component '{value}' not found in output. "
+            "Output does not contain this structural element."
+        ),
+    )
+
+
+def _compact_component_text(value: str) -> str:
+    """Normalize separators for structural component matching."""
+    return re.sub(r"\s+", "", value)
+
+
+def _check_absent(
+    expectation: StructuralExpectation,
+    actual_output: str,
+) -> ComponentCheck:
+    """Check that prohibited value is NOT present in output.
+
+    Occurrences preceded by a negation word (``no``, ``without``, ``not``,
+    ``never``, ``none``) are treated as the output asserting the absence
+    of the prohibited component, not as a violation.
+    """
+    text = actual_output if expectation.case_sensitive else actual_output.lower()
+    target = (
+        expectation.value if expectation.case_sensitive else expectation.value.lower()
+    )
+
+    if target not in text:
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.PASS,
+            reason=f"Prohibited component '{expectation.value}' not found.",
+        )
+
+    # Found at least one occurrence: a violation unless every occurrence is
+    # preceded by a negation word.
+    flags = 0 if expectation.case_sensitive else re.IGNORECASE
+    negation_re = re.compile(
+        rf"\b(no|without|not|never|none)\s+{re.escape(target)}",
+        flags=flags,
+    )
+    plain_re = re.compile(re.escape(target), flags=flags)
+    plain_count = len(plain_re.findall(actual_output))
+    negated_count = len(negation_re.findall(actual_output))
+
+    if plain_count > 0 and plain_count == negated_count:
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.PASS,
+            reason=(
+                f"All occurrences of '{expectation.value}' are in a negated "
+                "context (e.g., 'no X' / 'without X')."
+            ),
+        )
+
+    return ComponentCheck(
+        expectation=expectation,
+        result=ComponentCheckResult.FAIL,
+        reason=(
+            f"Prohibited component '{expectation.value}' found in output. "
+            "This violates the absence constraint."
+        ),
+    )
+
+
+def _check_pattern(
+    expectation: StructuralExpectation,
+    actual_output: str,
+) -> ComponentCheck:
+    """Check that output matches a regex pattern at least once.
+
+    Args:
+        expectation: The pattern expectation.
+        actual_output: The output to check against.
+    """
+    flags = 0 if expectation.case_sensitive else re.IGNORECASE
+
+    try:
+        match = re.search(expectation.value, actual_output, flags)
+    except re.error:
+        bad_pat = f"Invalid regex pattern '{expectation.value}'"
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.FAIL,
+            reason=f"{bad_pat}: not valid regex.",
+        )
+
+    if match:
+        matched = match.group(0)
+        short = matched[:80] + ("..." if len(matched) > 80 else "")
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.PASS,
+            matched_text=short,
+            reason=f"Regex pattern '{expectation.value}' matched in output.",
+        )
+
+    return ComponentCheck(
+        expectation=expectation,
+        result=ComponentCheckResult.FAIL,
+        reason=(
+            f"Regex pattern '{expectation.value}' did not match. "
+            "No section of the output satisfies this structural requirement."
+        ),
+    )
+
+
+def _check_nested_field(
+    field_path: str,
+    actual_output: str,
+    expectation: StructuralExpectation,
+) -> ComponentCheck:
+    """Check for nested field paths in JSON/dict-like output."""
+    import json
+
+    try:
+        parsed = json.loads(actual_output)
+    except (TypeError, ValueError):
+        return ComponentCheck(
+            expectation=expectation,
+            result=(
+                ComponentCheckResult.PASS
+                if expectation.kind == ExpectKind.ABSENT
+                else ComponentCheckResult.FAIL
+            ),
+            reason=(
+                "Output is not valid JSON; cannot check nested field "
+                f"path '{field_path}' directly. Treating as absent/pass."
+            ),
+        )
+
+    keys = field_path.split(".")
+    obj: Any = parsed
+    all_keys_resolved = True
+    broken_key = "?"
+
+    for key in keys:
+        if isinstance(obj, dict) and key in obj:
+            obj = obj[key]
+        elif isinstance(obj, list) and key.isdigit():
+            idx = int(key)
+            if idx < len(obj):
+                obj = obj[idx]
+            else:
+                all_keys_resolved = False
+                broken_key = key
+                break
+        else:
+            all_keys_resolved = False
+            broken_key = key
+            break
+
+    # When keys are missing, treat any non-ABSENT check as a FAIL rather than
+    # returning PASS on stale intermediate values of ``obj``.
+    if not all_keys_resolved:
+        if expectation.kind == ExpectKind.ABSENT:
+            return ComponentCheck(
+                expectation=expectation,
+                result=ComponentCheckResult.PASS,
+                reason=f"Nested field '{field_path}' does not exist.",
+            )
+        # Any other kind expects the field to be present; it is not found.
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.FAIL,
+            reason=(
+                f"Nested field '{field_path}' could not be resolved: "
+                f"path terminated early at key '{broken_key}'."
+            ),
+        )
+
+    str_obj = (str(obj) if obj else "") if isinstance(obj, (dict, list)) else str(obj)
+
+    if expectation.kind == ExpectKind.FIELD_PRESENT:
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.PASS,
+            matched_text=str_obj[:80],
+            reason=f"Nested field '{field_path}' exists.",
+        )
+
+    # Check value match if specified
+    check_lower = str_obj.lower() if not expectation.case_sensitive else str_obj
+    target_lower = (
+        expectation.value.lower()
+        if not expectation.case_sensitive
+        else expectation.value
+    )
+
+    if target_lower in check_lower:
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.PASS,
+            matched_text=str_obj[:80],
+            reason=f"Nested field '{field_path}' value matches.",
+        )
+
+    return ComponentCheck(
+        expectation=expectation,
+        result=ComponentCheckResult.FAIL,
+        reason=(
+            f"Nested field '{field_path}' exists but value does not match. "
+            f"Expected: '{expectation.value}'. Got: {str_obj[:80]}"
+        ),
+    )
+
+
+def _execute_check(
+    expectation: StructuralExpectation,
+    actual_output: str,
+) -> ComponentCheck:
+    """Dispatch to the appropriate matching function."""
+    if expectation.kind == ExpectKind.PRESENT:
+        return _match_present(expectation, actual_output)
+    elif expectation.kind == ExpectKind.ABSENT:
+        return _check_absent(expectation, actual_output)
+    elif expectation.kind == ExpectKind.PATTERN:
+        return _check_pattern(expectation, actual_output)
+    elif expectation.kind == ExpectKind.FIELD_PRESENT:
+        return _match_present(expectation, actual_output)
+    else:
+        return ComponentCheck(
+            expectation=expectation,
+            result=ComponentCheckResult.FAIL,
+            reason=f"Unknown kind: {expectation.kind}",
+        )
+
+
+# ---------------------------------------------------------------------------
+# Public validation API
+# ---------------------------------------------------------------------------
+
+
+def validate_structural_components(
+    expectations: list[StructuralExpectation],
+    actual_outputs: list[str],
+) -> ValidationOutcome:
+    """Validate multiple outputs against structural expectations.
+
+    Args:
+        expectations: Structural requirements to check against.
+        actual_outputs: Output strings produced during plan execution.
+
+    Returns:
+        A ``ValidationOutcome`` with aggregated results.
+    """
+    if not expectations:
+        return ValidationOutcome(
+            passed=True,
+            total_checks=0,
+            passed_count=0,
+            partial_count=0,
+            failed_count=0,
+            weighted_score=1.0,
+        )
+
+    all_results: list[ComponentCheck] = []
+    weighted_total = 0.0
+    weighted_achieved = 0.0
+
+    for output in actual_outputs:
+        for expectation in expectations:
+            result = _execute_check(expectation, output)
+            all_results.append(result)
+
+            if result.result == ComponentCheckResult.PASS:
+                weighted_achieved += expectation.weight
+            elif result.result == ComponentCheckResult.PARTIAL:
+                weighted_achieved += expectation.weight * 0.5
+            # FAIL contributes 0
+
+            weighted_total += expectation.weight
+
+    score = weighted_achieved / weighted_total if weighted_total > 0 else 0.0
+
+    passed_count = sum(1 for r in all_results if r.result == ComponentCheckResult.PASS)
+    partial_count = sum(
+        1 for r in all_results if r.result == ComponentCheckResult.PARTIAL
+    )
+    failed_count = sum(1 for r in all_results if r.result == ComponentCheckResult.FAIL)
+
+    max_weight_fail = any(
+        r.result == ComponentCheckResult.FAIL and r.expectation.weight >= 1.0
+        for r in all_results
+    )
+    passed = not max_weight_fail and score >= 0.5
+
+    return ValidationOutcome(
+        passed=passed,
+        total_checks=len(all_results),
+        passed_count=passed_count,
+        partial_count=partial_count,
+        failed_count=failed_count,
+        weighted_score=round(score, 4),
+        results=all_results,
+    )
+
+
+def validate_single(
+    expectation: StructuralExpectation,
+    actual_output: str,
+) -> ComponentCheckResult:
+    """Validate a single expectation against a single output."""
+    outcome = validate_structural_components(
+        expectations=[expectation],
+        actual_outputs=[actual_output],
+    )
+
+    if not outcome.results:
+        return ComponentCheckResult.PASS
+
+    has_fail = any(r.result == ComponentCheckResult.FAIL for r in outcome.results)
+    has_partial = any(r.result == ComponentCheckResult.PARTIAL for r in outcome.results)
+
+    if has_fail:
+        return ComponentCheckResult.FAIL
+    if has_partial:
+        return ComponentCheckResult.PARTIAL
+    return ComponentCheckResult.PASS
+
+
+__all__ = [
+    "ComponentCheck",
+    "ComponentCheckResult",
+    "ExpectKind",
+    "StructuralExpectation",
+    "ValidationOutcome",
+    "validate_single",
+    "validate_structural_components",
+]

tests/actor/test_config_actor_combined_format.py

@@ -7,8 +7,6 @@ combined-format actor and the nested-dict form.
 
 from __future__ import annotations
 
-import pytest
-
 from cleveragents.actor.schema import (
     _detect_nested_config_actor,
     _flatten_config_actor,