CHANGELOG.md

@@ -2,6 +2,14 @@
 
 ## Unreleased
 
+- Wired Invariant Reconciliation Actor into the Strategize phase of the
+  plan lifecycle.  When `PlanLifecycleService.start_strategize()` is called,
+  invariants from all scopes (global, project, action, plan) are collected,
+  reconciled using plan > project > global precedence, and recorded as
+  `invariant_enforced` decisions in the plan's decision tree.  Added
+  `invariant_service` parameter to `PlanLifecycleService` and registered
+  `InvariantService` as a singleton in the DI container.  Includes 6 Behave
+  BDD scenarios and 4 Robot integration tests. (#843)
 - Added TDD bug-capture E2E tests for bug #1028 — ACMS indexing pipeline not
   wired into CLI.  Four Robot Framework E2E tests prove ContextTierService starts
   empty on every CLI invocation.  Tests use ``@tdd_expected_fail`` until the bug

features/invariant_enforcement_strategize.feature

@@ -0,0 +1,73 @@
+Feature: Invariant Enforcement During Strategize Phase
+  As a plan strategizer
+  I want invariants to be automatically enforced when a plan enters Strategize
+  So that the decision tree reflects invariant constraints as first-class decisions
+
+  Background:
+    Given a lifecycle service with invariant enforcement wired
+
+  # === Basic enforcement ===
+
+  Scenario: Invariants are enforced when strategize starts
+    Given an action "local/test-action" with action invariants
+      | text                        | source |
+      | All tests must pass         | action |
+    And a plan created from invariant-action "local/test-action"
+    And a global invariant "Never delete production data" in the invariant service
+    When I begin strategize with invariant enforcement on the plan
+    Then the enforced plan should be in processing state
+    And invariant_enforced decisions should exist in the decision tree
+    And the decision tree should have an invariant_enforced for "All tests must pass"
+    And the decision tree should have an invariant_enforced for "Never delete production data"
+
+  Scenario: Strategize succeeds with no invariants
+    Given an action "local/no-invariants" with no action invariants
+    And a plan created from invariant-action "local/no-invariants"
+    When I begin strategize with invariant enforcement on the plan
+    Then the enforced plan should be in processing state
+    And 0 invariant_enforced decisions should be in the decision tree
+
+  Scenario: Strategize succeeds when invariant service is not available
+    Given a lifecycle service without invariant enforcement wired
+    And an action "local/test-action" created on the bare service
+    And a plan created from bare-service action "local/test-action"
+    When I begin strategize on the bare-service plan
+    Then the bare-service plan should be in processing state
+
+  # === Multi-scope enforcement ===
+
+  Scenario: Invariants from multiple scopes are reconciled during strategize
+    Given an action "local/multi-scope" with action invariants
+      | text                           | source |
+      | Mock all network calls         | action |
+    And a plan created from invariant-action "local/multi-scope"
+    And a global invariant "Backward compatibility required" in the invariant service
+    And a project invariant "Use ORM for queries" in the invariant service for project "local/api-service"
+    And the enforced plan targets project "local/api-service"
+    When I begin strategize with invariant enforcement on the plan
+    Then the enforced plan should be in processing state
+    And invariant_enforced decisions should exist in the decision tree
+    And the invariant_enforced decision count should be at least 3
+
+  # === Conflict resolution during enforcement ===
+
+  Scenario: Duplicate invariants are de-duplicated during enforcement
+    Given an action "local/dedup-action" with action invariants
+      | text              | source |
+      | All tests must pass | action |
+    And a plan created from invariant-action "local/dedup-action"
+    And a global invariant "All tests must pass" in the invariant service
+    When I begin strategize with invariant enforcement on the plan
+    Then the enforced plan should be in processing state
+    And the invariant_enforced decision count should be at least 1
+
+  # === Failure resilience ===
+
+  Scenario: Invariant enforcement failure does not block strategize
+    Given an action "local/resilience-test" with action invariants
+      | text                  | source |
+      | Some constraint       | action |
+    And a plan created from invariant-action "local/resilience-test"
+    And the invariant service is rigged to fail
+    When I begin strategize with invariant enforcement on the plan
+    Then the enforced plan should be in processing state

features/steps/invariant_enforcement_strategize_steps.py

@@ -0,0 +1,250 @@
+"""Step definitions for invariant_enforcement_strategize.feature.
+
+Tests that the Invariant Reconciliation Actor is invoked during the
+Strategize phase via ``PlanLifecycleService.start_strategize``, recording
+``invariant_enforced`` decisions in the decision tree.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import MagicMock
+
+from behave import given, then, when  # type: ignore[import-untyped]
+from behave.runner import Context
+
+from cleveragents.application.services.decision_service import DecisionService
+from cleveragents.application.services.invariant_service import InvariantService
+from cleveragents.application.services.plan_lifecycle_service import (
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings
+from cleveragents.domain.models.core.decision import DecisionType
+from cleveragents.domain.models.core.invariant import InvariantScope
+from cleveragents.domain.models.core.plan import ProcessingState, ProjectLink
+
+# ================================================================
+# Background
+# ================================================================
+
+
+@given("a lifecycle service with invariant enforcement wired")
+def step_lifecycle_with_invariant_enforcement(context: Context) -> None:
+    """Create a PlanLifecycleService with InvariantService and DecisionService."""
+    settings = Settings()
+    context.inv_enforcement_svc = InvariantService()
+    context.dec_enforcement_svc = DecisionService()
+    context.enforcement_lifecycle = PlanLifecycleService(
+        settings=settings,
+        decision_service=context.dec_enforcement_svc,
+        invariant_service=context.inv_enforcement_svc,
+    )
+    context.error = None
+
+
+@given("a lifecycle service without invariant enforcement wired")
+def step_lifecycle_without_invariant_service(context: Context) -> None:
+    """Create a PlanLifecycleService without InvariantService."""
+    settings = Settings()
+    context.bare_dec_svc = DecisionService()
+    context.bare_lifecycle = PlanLifecycleService(
+        settings=settings,
+        decision_service=context.bare_dec_svc,
+    )
+
+
+# ================================================================
+# Action and plan setup
+# ================================================================
+
+
+@given('an action "{name}" with action invariants')
+def step_create_action_with_invariants(context: Context, name: str) -> None:
+    """Create an action with invariants from the table."""
+    invariant_texts: list[str] = []
+    for row in context.table:
+        invariant_texts.append(row["text"])
+
+    context.enforcement_action = context.enforcement_lifecycle.create_action(
+        name=name,
+        description=f"Action {name}",
+        definition_of_done="All checks pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        invariants=invariant_texts,
+    )
+
+
+@given('an action "{name}" with no action invariants')
+def step_create_action_without_invariants(context: Context, name: str) -> None:
+    """Create an action without invariants."""
+    context.enforcement_action = context.enforcement_lifecycle.create_action(
+        name=name,
+        description=f"Action {name}",
+        definition_of_done="All checks pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+    )
+
+
+@given('an action "{name}" created on the bare service')
+def step_create_action_bare(context: Context, name: str) -> None:
+    """Create an action on the bare lifecycle service."""
+    context.bare_action = context.bare_lifecycle.create_action(
+        name=name,
+        description=f"Action {name}",
+        definition_of_done="All checks pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+    )
+
+
+@given('a plan created from invariant-action "{name}"')
+def step_create_plan_from_action(context: Context, name: str) -> None:
+    """Use the action to create a plan."""
+    context.enforced_plan = context.enforcement_lifecycle.use_action(
+        action_name=name,
+    )
+    context.enforced_plan_id = context.enforced_plan.identity.plan_id
+
+
+@given('a plan created from bare-service action "{name}"')
+def step_create_plan_from_bare_action(context: Context, name: str) -> None:
+    """Use the bare-service action to create a plan."""
+    context.bare_plan = context.bare_lifecycle.use_action(
+        action_name=name,
+    )
+    context.bare_plan_id = context.bare_plan.identity.plan_id
+
+
+@given('a global invariant "{text}" in the invariant service')
+def step_register_global_invariant(context: Context, text: str) -> None:
+    """Register a global invariant in the invariant service."""
+    context.inv_enforcement_svc.add_invariant(
+        text=text,
+        scope=InvariantScope.GLOBAL,
+        source_name="system",
+    )
+
+
+@given('a project invariant "{text}" in the invariant service for project "{project}"')
+def step_register_project_invariant(context: Context, text: str, project: str) -> None:
+    """Register a project invariant in the invariant service."""
+    context.inv_enforcement_svc.add_invariant(
+        text=text,
+        scope=InvariantScope.PROJECT,
+        source_name=project,
+    )
+
+
+@given('the enforced plan targets project "{project}"')
+def step_plan_targets_project(context: Context, project: str) -> None:
+    """Add a project link to the current plan."""
+    link = ProjectLink(project_name=project)
+    context.enforced_plan.project_links = [link]
+
+
+@given("the invariant service is rigged to fail")
+def step_invariant_service_fails(context: Context) -> None:
+    """Replace invariant service with a mock that raises."""
+    failing_svc = MagicMock(spec=InvariantService)
+    failing_svc.list_invariants.side_effect = RuntimeError(
+        "Simulated invariant service failure"
+    )
+    failing_svc.add_invariant.side_effect = RuntimeError(
+        "Simulated invariant service failure"
+    )
+    context.enforcement_lifecycle._invariant_service = failing_svc
+
+
+# ================================================================
+# When steps
+# ================================================================
+
+
+@when("I begin strategize with invariant enforcement on the plan")
+def step_start_strategize(context: Context) -> None:
+    """Start the strategize phase on the current plan."""
+    context.enforced_plan = context.enforcement_lifecycle.start_strategize(
+        context.enforced_plan_id,
+    )
+
+
+@when("I begin strategize on the bare-service plan")
+def step_start_strategize_bare(context: Context) -> None:
+    """Start strategize on the bare-service plan."""
+    context.bare_plan = context.bare_lifecycle.start_strategize(
+        context.bare_plan_id,
+    )
+
+
+# ================================================================
+# Then steps
+# ================================================================
+
+
+@then("the enforced plan should be in processing state")
+def step_plan_processing(context: Context) -> None:
+    """Assert the plan is in PROCESSING state."""
+    assert context.enforced_plan.processing_state == ProcessingState.PROCESSING, (
+        f"Expected PROCESSING, got {context.enforced_plan.processing_state}"
+    )
+
+
+@then("the bare-service plan should be in processing state")
+def step_bare_plan_processing(context: Context) -> None:
+    """Assert the bare plan is in PROCESSING state."""
+    assert context.bare_plan.processing_state == ProcessingState.PROCESSING, (
+        f"Expected PROCESSING, got {context.bare_plan.processing_state}"
+    )
+
+
+@then("invariant_enforced decisions should exist in the decision tree")
+def step_invariant_decisions_recorded(context: Context) -> None:
+    """Assert that at least one invariant_enforced decision exists."""
+    decisions = context.dec_enforcement_svc.list_decisions(context.enforced_plan_id)
+    invariant_decisions = [
+        d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED
+    ]
+    assert len(invariant_decisions) > 0, (
+        "Expected at least one invariant_enforced decision, found none"
+    )
+
+
+@then('the decision tree should have an invariant_enforced for "{text}"')
+def step_decision_contains_invariant_text(context: Context, text: str) -> None:
+    """Assert an invariant_enforced decision exists mentioning the text."""
+    decisions = context.dec_enforcement_svc.list_decisions(context.enforced_plan_id)
+    invariant_decisions = [
+        d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED
+    ]
+    found = any(text in d.chosen_option for d in invariant_decisions)
+    assert found, (
+        f"No invariant_enforced decision found for '{text}'. "
+        f"Decisions: {[d.chosen_option for d in invariant_decisions]}"
+    )
+
+
+@then("{count:d} invariant_enforced decisions should be in the decision tree")
+def step_invariant_decision_count(context: Context, count: int) -> None:
+    """Assert the exact number of invariant_enforced decisions."""
+    decisions = context.dec_enforcement_svc.list_decisions(context.enforced_plan_id)
+    invariant_decisions = [
+        d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED
+    ]
+    actual = len(invariant_decisions)
+    assert actual == count, (
+        f"Expected {count} invariant_enforced decisions, got {actual}"
+    )
+
+
+@then("the invariant_enforced decision count should be at least {count:d}")
+def step_decision_count_at_least(context: Context, count: int) -> None:
+    """Assert the plan has at least the given number of invariant decisions."""
+    decisions = context.dec_enforcement_svc.list_decisions(context.enforced_plan_id)
+    invariant_decisions = [
+        d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED
+    ]
+    actual = len(invariant_decisions)
+    assert actual >= count, (
+        f"Expected at least {count} invariant_enforced decisions, got {actual}"
+    )

robot/helper_invariant_enforcement_strategize.py

@@ -0,0 +1,224 @@
+"""Robot Framework helper for invariant enforcement during Strategize.
+
+Exercises PlanLifecycleService.start_strategize with invariant enforcement
+wired, verifying that invariant_enforced decisions appear in the decision
+tree after strategize starts.
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock
+
+# Ensure source tree is importable
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+from cleveragents.application.services.decision_service import (  # noqa: E402
+    DecisionService,
+)
+from cleveragents.application.services.invariant_service import (  # noqa: E402
+    InvariantService,
+)
+from cleveragents.application.services.plan_lifecycle_service import (  # noqa: E402
+    PlanLifecycleService,
+)
+from cleveragents.config.settings import Settings  # noqa: E402
+from cleveragents.domain.models.core.decision import DecisionType  # noqa: E402
+from cleveragents.domain.models.core.invariant import InvariantScope  # noqa: E402
+
+
+def _enforce_on_strategize() -> dict[str, object]:
+    """Test that invariants are enforced when strategize starts."""
+    settings = Settings()
+    inv_svc = InvariantService()
+    dec_svc = DecisionService()
+    svc = PlanLifecycleService(
+        settings=settings,
+        decision_service=dec_svc,
+        invariant_service=inv_svc,
+    )
+
+    # Create action with an invariant
+    svc.create_action(
+        name="local/test-enforce",
+        description="Test enforcement",
+        definition_of_done="All checks pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        invariants=["All tests must pass"],
+    )
+
+    # Add a global invariant
+    inv_svc.add_invariant(
+        "Never delete production data",
+        InvariantScope.GLOBAL,
+        "system",
+    )
+
+    # Create plan and start strategize
+    plan = svc.use_action(action_name="local/test-enforce")
+    plan = svc.start_strategize(plan.identity.plan_id)
+
+    # Count invariant_enforced decisions
+    decisions = dec_svc.list_decisions(plan.identity.plan_id)
+    inv_decisions = [
+        d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED
+    ]
+
+    return {
+        "plan_state": plan.processing_state.value
+        if plan.processing_state
+        else "unknown",
+        "decision_count": len(inv_decisions),
+    }
+
+
+def _no_invariant_service() -> dict[str, object]:
+    """Test strategize works without invariant service."""
+    settings = Settings()
+    dec_svc = DecisionService()
+    svc = PlanLifecycleService(
+        settings=settings,
+        decision_service=dec_svc,
+    )
+
+    svc.create_action(
+        name="local/no-inv",
+        description="No invariants",
+        definition_of_done="All checks pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+    )
+
+    plan = svc.use_action(action_name="local/no-inv")
+    plan = svc.start_strategize(plan.identity.plan_id)
+
+    decisions = dec_svc.list_decisions(plan.identity.plan_id)
+    inv_decisions = [
+        d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED
+    ]
+
+    return {
+        "plan_state": plan.processing_state.value
+        if plan.processing_state
+        else "unknown",
+        "decision_count": len(inv_decisions),
+    }
+
+
+def _multi_scope() -> dict[str, object]:
+    """Test multi-scope invariant reconciliation during strategize."""
+    settings = Settings()
+    inv_svc = InvariantService()
+    dec_svc = DecisionService()
+    svc = PlanLifecycleService(
+        settings=settings,
+        decision_service=dec_svc,
+        invariant_service=inv_svc,
+    )
+
+    # Add invariants from multiple scopes
+    inv_svc.add_invariant(
+        "Backward compatibility required",
+        InvariantScope.GLOBAL,
+        "system",
+    )
+    inv_svc.add_invariant(
+        "Use ORM for queries",
+        InvariantScope.PROJECT,
+        "local/api-service",
+    )
+
+    svc.create_action(
+        name="local/multi-scope-enforce",
+        description="Multi-scope test",
+        definition_of_done="All checks pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+        invariants=["Mock all network calls"],
+    )
+
+    from cleveragents.domain.models.core.plan import ProjectLink
+
+    plan = svc.use_action(
+        action_name="local/multi-scope-enforce",
+        project_links=[ProjectLink(project_name="local/api-service")],
+    )
+    plan = svc.start_strategize(plan.identity.plan_id)
+
+    decisions = dec_svc.list_decisions(plan.identity.plan_id)
+    inv_decisions = [
+        d for d in decisions if d.decision_type == DecisionType.INVARIANT_ENFORCED
+    ]
+
+    return {
+        "plan_state": plan.processing_state.value
+        if plan.processing_state
+        else "unknown",
+        "decision_count": len(inv_decisions),
+    }
+
+
+def _resilient() -> dict[str, object]:
+    """Test strategize proceeds despite invariant enforcement failure."""
+    settings = Settings()
+    failing_inv_svc = MagicMock(spec=InvariantService)
+    failing_inv_svc.list_invariants.side_effect = RuntimeError("Simulated failure")
+    failing_inv_svc.add_invariant.side_effect = RuntimeError("Simulated failure")
+
+    dec_svc = DecisionService()
+    svc = PlanLifecycleService(
+        settings=settings,
+        decision_service=dec_svc,
+        invariant_service=failing_inv_svc,
+    )
+
+    svc.create_action(
+        name="local/resilient",
+        description="Resilience test",
+        definition_of_done="All checks pass",
+        strategy_actor="openai/gpt-4",
+        execution_actor="openai/gpt-4",
+    )
+
+    plan = svc.use_action(action_name="local/resilient")
+    plan = svc.start_strategize(plan.identity.plan_id)
+
+    return {
+        "plan_state": plan.processing_state.value
+        if plan.processing_state
+        else "unknown",
+    }
+
+
+def main() -> None:
+    """Entry point for Robot Framework helper."""
+    if len(sys.argv) < 2:
+        print("Usage: helper_invariant_enforcement_strategize.py <command>")
+        sys.exit(1)
+
+    command = sys.argv[1]
+    handlers: dict[str, object] = {
+        "enforce-on-strategize": _enforce_on_strategize,
+        "no-invariant-service": _no_invariant_service,
+        "multi-scope": _multi_scope,
+        "resilient": _resilient,
+    }
+
+    handler = handlers.get(command)
+    if handler is None:
+        print(f"Unknown command: {command}")
+        sys.exit(1)
+
+    if callable(handler):
+        result = handler()
+        print(json.dumps(result))
+        print("enforcement-ok")
+
+
+if __name__ == "__main__":
+    main()

robot/invariant_enforcement_strategize.robot

@@ -0,0 +1,49 @@
+*** Settings ***
+Documentation    Integration tests for invariant enforcement during Strategize phase
+...              Verifies that PlanLifecycleService invokes the Invariant Reconciliation Actor
+...              at the start of Strategize, recording invariant_enforced decisions
+Resource         ${CURDIR}/common.resource
+Suite Setup      Setup Test Environment
+Suite Teardown   Cleanup Test Environment
+
+*** Variables ***
+${HELPER}    ${CURDIR}/helper_invariant_enforcement_strategize.py
+
+*** Test Cases ***
+Invariants Enforced During Strategize Start
+    [Documentation]    When strategize starts, invariants from action and global scope are reconciled
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    enforce-on-strategize    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    enforcement-ok
+    Should Contain    ${result.stdout}    "decision_count": 2
+    Should Contain    ${result.stdout}    "plan_state": "processing"
+
+Strategize Without Invariant Service
+    [Documentation]    Strategize succeeds when no InvariantService is wired
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    no-invariant-service    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    enforcement-ok
+    Should Contain    ${result.stdout}    "plan_state": "processing"
+    Should Contain    ${result.stdout}    "decision_count": 0
+
+Strategize With Multi-Scope Invariants
+    [Documentation]    Strategize reconciles invariants from global, project, and plan scopes
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    multi-scope    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    enforcement-ok
+    Should Contain    ${result.stdout}    "plan_state": "processing"
+
+Strategize Resilient To Invariant Failure
+    [Documentation]    Strategize proceeds even if invariant enforcement fails
+    ${result}=    Run Process    ${PYTHON}    ${HELPER}    resilient    cwd=${WORKSPACE}
+    Log    ${result.stdout}
+    Log    ${result.stderr}
+    Should Be Equal As Integers    ${result.rc}    0
+    Should Contain    ${result.stdout}    enforcement-ok
+    Should Contain    ${result.stdout}    "plan_state": "processing"

src/cleveragents/application/container.py

@@ -45,6 +45,7 @@ from cleveragents.application.services.execution_environment_resolver import (
 from cleveragents.application.services.fix_then_revalidate import (
     FixThenRevalidateOrchestrator,
 )
+from cleveragents.application.services.invariant_service import InvariantService
 from cleveragents.application.services.multi_project_service import (
     MultiProjectService,
 )
@@ -581,6 +582,11 @@ class Container(containers.DeclarativeContainer):
         event_bus=event_bus,
     )
 
+    # Invariant Service - Singleton (in-memory state shared across callers)
+    invariant_service = providers.Singleton(
+        InvariantService,
+    )
+
     # Plan Lifecycle Service - Factory (v3 four-phase lifecycle)
     plan_lifecycle_service = providers.Factory(
         PlanLifecycleService,
@@ -588,6 +594,7 @@ class Container(containers.DeclarativeContainer):
         unit_of_work=unit_of_work,
         decision_service=decision_service,
         event_bus=event_bus,
+        invariant_service=invariant_service,
     )
 
     # Checkpoint Service - database-backed via CheckpointRepository

src/cleveragents/application/services/plan_lifecycle_service.py

@@ -96,6 +96,7 @@ if TYPE_CHECKING:
     from cleveragents.application.services.error_pattern_service import (
         ErrorPatternService,
     )
+    from cleveragents.application.services.invariant_service import InvariantService
     from cleveragents.config.settings import Settings
     from cleveragents.infrastructure.database.unit_of_work import UnitOfWork
     from cleveragents.infrastructure.events.protocol import EventBus
@@ -166,6 +167,7 @@ class PlanLifecycleService:
         event_bus: EventBus | None = None,
         job_store: InMemoryJobStore | None = None,
         error_pattern_service: ErrorPatternService | None = None,
+        invariant_service: InvariantService | None = None,
     ):
         """Initialize the plan lifecycle service.
 
@@ -194,6 +196,14 @@ class PlanLifecycleService:
                 consulted before the Execute phase to inject preventive
                 guidance.  When ``None``, predictive prevention is
                 silently skipped.
+            invariant_service: Optional
+                :class:`InvariantService` for invariant enforcement
+                during the Strategize phase.  When provided, the
+                Invariant Reconciliation Actor is invoked at the start
+                of Strategize to collect, reconcile, and enforce
+                invariants as ``invariant_enforced`` decisions.
+                When ``None``, invariant enforcement is silently
+                skipped.
         """
         self.settings = settings
         self.unit_of_work = unit_of_work
@@ -201,6 +211,7 @@ class PlanLifecycleService:
         self.event_bus = event_bus
         self._job_store = job_store
         self.error_pattern_service = error_pattern_service
+        self._invariant_service = invariant_service
         self._logger = logger.bind(service="plan_lifecycle")
         self.preflight_guardrail = PlanPreflightGuardrail()
 
@@ -281,6 +292,73 @@ class PlanLifecycleService:
                 exc_info=True,
             )
 
+    def _enforce_invariants(self, plan: Plan) -> None:
+        """Invoke the Invariant Reconciliation Actor at Strategize start.
+
+        Collects invariants from all scopes (global, project, action,
+        plan), reconciles conflicts using plan > project > global
+        precedence, and records ``invariant_enforced`` decisions in the
+        plan's decision tree.
+
+        Requires both ``_invariant_service`` and ``decision_service``
+        to be wired.  When either is ``None``, invariant enforcement is
+        silently skipped.
+
+        Failures are logged but never propagated — invariant enforcement
+        must not block lifecycle transitions.
+        """
+        if self._invariant_service is None or self.decision_service is None:
+            return
+
+        try:
+            from cleveragents.actor.reconciliation import (
+                InvariantReconciliationActor,
+            )
+
+            actor = InvariantReconciliationActor(
+                invariant_service=self._invariant_service,
+                decision_service=self.decision_service,
+            )
+
+            # Extract project and action names from the plan
+            project_name: str | None = None
+            if plan.project_links:
+                project_name = plan.project_links[0].project_name
+
+            action_name: str | None = plan.action_name
+
+            # Also register any plan-level invariants from the plan
+            # model into the invariant service so the actor can
+            # discover them during collection.
+            from cleveragents.domain.models.core.invariant import InvariantScope
+
+            for inv in plan.invariants or []:
+                self._invariant_service.add_invariant(
+                    text=inv.text,
+                    scope=InvariantScope.PLAN,
+                    source_name=plan.identity.plan_id,
+                )
+
+            result = actor.run(
+                plan_id=plan.identity.plan_id,
+                project_name=project_name,
+                action_name=action_name,
+            )
+
+            self._logger.info(
+                "Invariant enforcement completed",
+                plan_id=plan.identity.plan_id,
+                effective_count=len(result.reconciled_set.invariants),
+                conflict_count=len(result.conflicts),
+                decision_count=len(result.enforced_decision_ids),
+            )
+        except Exception:
+            self._logger.warning(
+                "invariant_enforcement_failed",
+                plan_id=plan.identity.plan_id,
+                exc_info=True,
+            )
+
     def _run_estimation(self, plan: Plan) -> None:
         """Run the estimation actor if configured on the plan.
 
@@ -999,6 +1077,13 @@ class PlanLifecycleService:
             chosen_option=f"Begin strategize phase for plan {plan_id}",
         )
 
+        # -- Invariant enforcement ----------------------------------------
+        # Per spec §19554-19610: the Invariant Reconciliation Actor is
+        # invoked at the start of Strategize to compute effective
+        # invariants and record invariant_enforced decisions.
+        self._enforce_invariants(plan)
+        # -- End invariant enforcement ------------------------------------
+
         return plan
 
     def complete_strategize(self, plan_id: str) -> Plan: