fix(subplan): propagate invariant_enforced decisions to child plans on spawn

SubplanService.spawn() now re-records all invariant_enforced decisions from
the parent plan decision tree onto each child plan decision tree. This
satisfies the spec requirement: 'recorded as invariant_enforced decisions that
propagate to child plans.' (Glossary → Invariant)

Previously, child plans started Strategize with a completely empty invariant
set, violating the spec's propagation requirement. non_overridable global
invariants enforced on the parent were not guaranteed to be enforced on child
plans.

The fix adds _propagate_invariant_decisions() to SubplanService which queries
the parent plan's invariant_enforced decisions and re-records each one on the
child plan using DecisionService.record_decision(). The DecisionService is
already injected into SubplanService, so no new dependencies are required.

BDD regression coverage added in features/tdd_invariant_propagation_subplan.feature
with 4 scenarios covering: single invariant propagation, multiple invariant
propagation, non_overridable invariant propagation, and clean spawn with no
parent invariants.

ISSUES CLOSED: #9131

CHANGELOG.md

@@ -309,6 +309,16 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 
 ### Fixed
 
+- **`invariant_enforced` decisions not propagated to child plans on subplan spawn** (#9131):
+  Fixed `SubplanService.spawn()` to propagate all `invariant_enforced` decisions from the
+  parent plan's decision tree to each child plan's decision tree. Previously, child plans
+  started Strategize with a completely empty invariant set, violating the spec requirement:
+  "recorded as `invariant_enforced` decisions that propagate to child plans." The fix adds
+  a `_propagate_invariant_decisions()` helper that re-records each parent
+  `invariant_enforced` decision on the child plan, including `non_overridable` global
+  invariants. BDD regression coverage added in
+  `features/tdd_invariant_propagation_subplan.feature`.
+
 - **fix(repositories): derive PlanResult.success from result_success column instead of error_message** (#7501):
   Fixed a critical bug in `PlanRepository._to_domain` where `PlanResult.success` was incorrectly
   derived from `error_message is None`. Because `error_message` is shared between the build phase

CONTRIBUTORS.md

@@ -5,6 +5,7 @@
 * HAL 9000 <hal9000@cleverthis.com>
 * Hamza Khyari <hamza.khyari@cleverthis.com>
 * Jeffrey Phillips Freeman <jeffrey.freeman@syncleus.com>
+* Jeffrey Phillips Freeman <the@jeffreyfreeman.me>
 * Luis Mendes <luis.p.mendes@gmail.com>
 * Rui Hu <rui.hu@cleverthis.com>
 
@@ -13,6 +14,7 @@
 Below are some of the specific details of various contributions.
 
 * Jeffrey Phillips Freeman has acted as Lead Developer, daily contributor, and Project Owner.
+* Jeffrey Phillips Freeman has contributed an implementation for the invariant propagation fix (PR #10881 / issue #9131): added `_propagate_invariant_decisions()` to `SubplanService` to propagate all `invariant_enforced` decisions from parent plans to child plan decision trees during subplan spawn, satisfying the specification requirement for invariant propagation across hierarchical plan execution.
 * Brent E. Edwards has contributed quality assurance, test coverage, and CI pipeline improvements.
 * HAL 9000 has contributed automated implementation, bug fixes, and feature development as part of the CleverAgents automation pool.
 * HAL 9000 has contributed concurrency safety improvements, including thread-safe context tier management (issue #7547) for parallel plan execution.

features/steps/tdd_invariant_propagation_subplan_steps.py

@@ -0,0 +1,331 @@
+"""Step definitions for TDD Issue #9131 — invariant_enforced decisions not
+propagated to child plans on subplan spawn.
+
+These steps exercise ``SubplanService.spawn()`` and verify that it propagates
+``invariant_enforced`` decisions from the parent plan to each child plan's
+decision tree.
+
+On ``master`` (before the fix), ``spawn()`` creates child Plan objects but
+does NOT record ``invariant_enforced`` decisions on them.  The assertions in
+these steps will **fail** until the bug is fixed.
+"""
+
+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.subplan_service import (
+    SpawnEntry,
+    SpawnResult,
+    SubplanService,
+)
+from cleveragents.domain.models.core.decision import (
+    ContextSnapshot,
+    Decision,
+    DecisionType,
+)
+from cleveragents.domain.models.core.plan import (
+    ExecutionMode,
+    NamespacedName,
+    Plan,
+    PlanIdentity,
+    SubplanConfig,
+)
+
+_PARENT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6PN00"
+_ROOT_PLAN_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6RF00"
+_SPAWN_DEC_ID: str = "01HGZ6FE0AQDYTR4BXVQZ6DA00"
+
+
+def _make_parent_plan() -> Plan:
+    """Create a minimal parent plan for testing."""
+    return Plan(
+        identity=PlanIdentity(
+            plan_id=_PARENT_PLAN_ID,
+            root_plan_id=_ROOT_PLAN_ID,
+        ),
+        namespaced_name=NamespacedName(namespace="local", name="parent-plan"),
+        description="Parent plan for invariant propagation test",
+        action_name="local/parent-action",
+    )
+
+
+def _make_spawn_decision(
+    plan_id: str = _PARENT_PLAN_ID,
+) -> Decision:
+    """Create a subplan_spawn Decision."""
+    return Decision(
+        decision_id=_SPAWN_DEC_ID,
+        plan_id=plan_id,
+        decision_type=DecisionType.SUBPLAN_SPAWN,
+        sequence_number=10,
+        question="Should we spawn a child plan?",
+        chosen_option="local/sub-action",
+        context_snapshot=ContextSnapshot(),
+    )
+
+
+def _make_spawn_entry(plan_id: str = _PARENT_PLAN_ID) -> SpawnEntry:
+    """Create a single spawn entry."""
+    return SpawnEntry(
+        decision=_make_spawn_decision(plan_id),
+        action_name="local/sub-action",
+        description="Child plan for invariant propagation test",
+    )
+
+
+# ---------------------------------------------------------------------------
+# Given steps
+# ---------------------------------------------------------------------------
+
+
+@given("a parent plan with invariant_enforced decisions recorded")
+def step_parent_with_invariant_decisions(context: Context) -> None:
+    """Set up a parent plan with one invariant_enforced decision."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+
+    # Record one invariant_enforced decision on the parent plan
+    inv_dec = context.decision_service.record_decision(
+        plan_id=_PARENT_PLAN_ID,
+        decision_type=DecisionType.INVARIANT_ENFORCED,
+        question="Is this invariant enforced?",
+        chosen_option="invariant enforced: do not modify production data",
+    )
+    context.parent_invariant_decisions = [inv_dec]
+    context.expected_invariant_count = 1
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a parent plan with {count:d} invariant_enforced decisions recorded")
+def step_parent_with_n_invariant_decisions(context: Context, count: int) -> None:
+    """Set up a parent plan with N invariant_enforced decisions."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+
+    inv_decisions: list[Decision] = []
+    for i in range(count):
+        inv_dec = context.decision_service.record_decision(
+            plan_id=_PARENT_PLAN_ID,
+            decision_type=DecisionType.INVARIANT_ENFORCED,
+            question=f"Is invariant {i} enforced?",
+            chosen_option=f"invariant enforced: constraint {i}",
+        )
+        inv_decisions.append(inv_dec)
+
+    context.parent_invariant_decisions = inv_decisions
+    context.expected_invariant_count = count
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a parent plan with a non_overridable invariant_enforced decision")
+def step_parent_with_non_overridable_invariant(context: Context) -> None:
+    """Set up a parent plan with a non_overridable invariant_enforced decision."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+
+    # Record a non_overridable invariant decision
+    inv_dec = context.decision_service.record_decision(
+        plan_id=_PARENT_PLAN_ID,
+        decision_type=DecisionType.INVARIANT_ENFORCED,
+        question="Is this non-overridable global invariant enforced?",
+        chosen_option="non_overridable invariant enforced: never delete user data",
+        rationale="Global non-overridable invariant must propagate to all child plans",
+    )
+    context.parent_invariant_decisions = [inv_dec]
+    context.non_overridable_chosen_option = inv_dec.chosen_option
+    context.expected_invariant_count = 1
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a parent plan with no invariant_enforced decisions")
+def step_parent_with_no_invariant_decisions(context: Context) -> None:
+    """Set up a parent plan with no invariant_enforced decisions."""
+    context.parent_plan = _make_parent_plan()
+    context.decision_service = DecisionService()
+    context.parent_invariant_decisions = []
+    context.expected_invariant_count = 0
+
+    context.subplan_service = SubplanService(
+        decision_service=context.decision_service,
+    )
+    context.subplan_config = SubplanConfig(
+        execution_mode=ExecutionMode.SEQUENTIAL,
+    )
+
+
+@given("a valid spawn entry for the parent plan")
+def step_valid_spawn_entry(context: Context) -> None:
+    """Create a single valid spawn entry."""
+    context.spawn_entries = [_make_spawn_entry()]
+
+
+# ---------------------------------------------------------------------------
+# When steps
+# ---------------------------------------------------------------------------
+
+
+@when("I spawn a child plan via SubplanService")
+def step_spawn_child_plan(context: Context) -> None:
+    """Call SubplanService.spawn() and capture the result."""
+    context.spawn_error = None
+    try:
+        context.spawn_result = context.subplan_service.spawn(
+            parent_plan=context.parent_plan,
+            config=context.subplan_config,
+            spawn_entries=context.spawn_entries,
+        )
+    except Exception as exc:
+        context.spawn_error = exc
+        context.spawn_result = None
+
+
+# ---------------------------------------------------------------------------
+# Then steps
+# ---------------------------------------------------------------------------
+
+
+@then(
+    "the decision service should have recorded invariant_enforced decisions"
+    " for the child plan"
+)
+def step_child_has_invariant_decisions(context: Context) -> None:
+    """Assert that invariant_enforced decisions were recorded for the child plan.
+
+    Bug #9131: spawn() does not propagate invariant_enforced decisions to
+    child plans.  This assertion will fail until the bug is fixed.
+    """
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert len(result.child_plans) > 0, "No child plans were created"
+
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    # Query the decision service for invariant_enforced decisions on the child plan
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) > 0, (
+        f"Child plan {child_plan_id!r} has no invariant_enforced decisions. "
+        f"Bug #9131: SubplanService.spawn() does not propagate "
+        f"invariant_enforced decisions from parent plan to child plans."
+    )
+
+
+@then("the child plan should have {count:d} invariant_enforced decisions recorded")
+def step_child_has_n_invariant_decisions(context: Context, count: int) -> None:
+    """Assert that the child plan has exactly N invariant_enforced decisions.
+
+    Bug #9131: spawn() does not propagate invariant_enforced decisions.
+    """
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert len(result.child_plans) > 0, "No child plans were created"
+
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) == count, (
+        f"Child plan {child_plan_id!r} has {len(child_invariant_decisions)} "
+        f"invariant_enforced decisions, expected {count}. "
+        f"Bug #9131: SubplanService.spawn() does not propagate "
+        f"invariant_enforced decisions from parent plan to child plans."
+    )
+
+
+@then("the child plan should have the non_overridable invariant decision propagated")
+def step_child_has_non_overridable_invariant(context: Context) -> None:
+    """Assert that the non_overridable invariant decision was propagated.
+
+    Bug #9131: spawn() does not propagate invariant_enforced decisions.
+    """
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert len(result.child_plans) > 0, "No child plans were created"
+
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) > 0, (
+        f"Child plan {child_plan_id!r} has no invariant_enforced decisions. "
+        f"Bug #9131: non_overridable invariant was not propagated."
+    )
+
+    # Verify the non_overridable invariant text was propagated
+    expected_option: str = context.non_overridable_chosen_option
+    propagated_options: list[str] = [d.chosen_option for d in child_invariant_decisions]
+    assert expected_option in propagated_options, (
+        f"Non-overridable invariant chosen_option {expected_option!r} not found "
+        f"in child plan decisions: {propagated_options}. "
+        f"Bug #9131: non_overridable invariant was not propagated."
+    )
+
+
+@then("the spawn should succeed with no invariant decisions recorded for the child")
+def step_spawn_succeeds_no_invariants(context: Context) -> None:
+    """Assert that spawn succeeds cleanly when parent has no invariant decisions."""
+    assert context.spawn_error is None, (
+        f"Spawn raised an unexpected error: {context.spawn_error}"
+    )
+    result: SpawnResult = context.spawn_result
+    assert result is not None, "Spawn result is None"
+    assert result.total_spawned == 1, (
+        f"Expected 1 spawned child, got {result.total_spawned}"
+    )
+
+    # When parent has no invariant decisions, child should also have none
+    assert len(result.child_plans) > 0, "No child plans were created"
+    child_plan: Plan = result.child_plans[0]
+    child_plan_id: str = child_plan.identity.plan_id
+
+    child_invariant_decisions: list[Decision] = context.decision_service.list_by_type(
+        child_plan_id,
+        DecisionType.INVARIANT_ENFORCED,
+    )
+
+    assert len(child_invariant_decisions) == 0, (
+        f"Child plan {child_plan_id!r} has {len(child_invariant_decisions)} "
+        f"invariant_enforced decisions, expected 0 (parent had none)."
+    )

features/tdd_invariant_propagation_subplan.feature

@@ -0,0 +1,38 @@
+@tdd_issue @tdd_issue_9131 @mock_only
+Feature: TDD Issue #9131 — invariant_enforced decisions not propagated to child plans on subplan spawn
+  As a developer
+  I want to verify that SubplanService.spawn() propagates invariant_enforced decisions
+  from the parent plan to each child plan's decision tree
+  So that child plans start Strategize with the parent's enforced invariant set
+
+  The spec states: "recorded as invariant_enforced decisions that propagate to child plans."
+  SubplanService.spawn() currently creates child Plan objects with no invariant_enforced
+  decisions, violating the spec's propagation requirement.
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Child plan receives parent invariant_enforced decisions on spawn
+    Given a parent plan with invariant_enforced decisions recorded
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the decision service should have recorded invariant_enforced decisions for the child plan
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Child plan receives all parent invariant_enforced decisions
+    Given a parent plan with 3 invariant_enforced decisions recorded
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the child plan should have 3 invariant_enforced decisions recorded
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Non-overridable global invariants are propagated to child plans
+    Given a parent plan with a non_overridable invariant_enforced decision
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the child plan should have the non_overridable invariant decision propagated
+
+  @tdd_issue @tdd_issue_9131
+  Scenario: Child plan with no parent invariant_enforced decisions spawns cleanly
+    Given a parent plan with no invariant_enforced decisions
+    And a valid spawn entry for the parent plan
+    When I spawn a child plan via SubplanService
+    Then the spawn should succeed with no invariant decisions recorded for the child

src/cleveragents/application/services/subplan_service.py

@@ -7,6 +7,17 @@ presence, and ``max_parallel`` bounds before creating child plans.
 Spawn metadata (spawn decision ID, parent/root plan IDs, execution mode) is
 persisted alongside each child plan for status output and provenance tracking.
 
+Invariant propagation
+---------------------
+
+Per the specification (Glossary → Invariant): "recorded as
+``invariant_enforced`` decisions that propagate to child plans."
+
+When a child plan is spawned, all ``invariant_enforced`` decisions from the
+parent plan's decision tree are re-recorded on the child plan's decision tree.
+This ensures that child plans start Strategize with the full invariant set
+enforced on the parent, including ``non_overridable`` global invariants.
+
 Design decisions:
     - Dependency injection: ``DecisionService`` and ``UnitOfWork`` are
       injected via the constructor (consistent with existing service patterns).
@@ -19,6 +30,7 @@ Based on:
     - docs/specification.md L18170-L18295 (subplan spawning)
     - ADR-006 (Plan Lifecycle)
     - Forgejo issue #197
+    - Forgejo issue #9131 (invariant_enforced propagation fix)
 """
 
 from __future__ import annotations
@@ -154,6 +166,8 @@ class SubplanService:
     2. Validating each entry (resource scopes, merge strategy, parallelism).
     3. Building ``SubplanStatus`` objects for the parent plan.
     4. Persisting ``SpawnMetadata`` for status queries.
+    5. Propagating ``invariant_enforced`` decisions from the parent plan to
+       each child plan's decision tree (spec: Glossary → Invariant).
 
     Args:
         decision_service: Service for querying decision records.
@@ -193,6 +207,12 @@ class SubplanService:
         and ``SpawnMetadata`` for each entry.  The caller is responsible for
         persisting the updated parent plan.
 
+        Invariant propagation: all ``invariant_enforced`` decisions from the
+        parent plan are re-recorded on each child plan's decision tree so that
+        child plans start Strategize with the parent's full invariant set.
+        This satisfies the spec requirement: "recorded as ``invariant_enforced``
+        decisions that propagate to child plans."
+
         Args:
             parent_plan: The parent plan that will own the child plans.
             config: Subplan execution configuration.
@@ -228,8 +248,17 @@ class SubplanService:
         if not validation.valid:
             raise SpawnValidationError(validation.errors)
 
-        # Build statuses, metadata, and real child Plan objects
+        # Retrieve parent plan's invariant_enforced decisions once, before
+        # creating child plans.  These will be propagated to every child.
         parent_id: str = parent_plan.identity.plan_id
+        parent_invariant_decisions: list[Decision] = (
+            self._decision_service.list_by_type(
+                parent_id,
+                DecisionType.INVARIANT_ENFORCED,
+            )
+        )
+
+        # Build statuses, metadata, and real child Plan objects
         root_id: str = parent_plan.identity.root_plan_id or parent_id
         mode: str = config.execution_mode.value
 
@@ -283,6 +312,16 @@ class SubplanService:
             )
             child_plans.append(child_plan)
 
+            # Propagate invariant_enforced decisions from parent to child.
+            # Each parent invariant decision is re-recorded on the child plan
+            # so that the child starts Strategize with the same invariant set.
+            # Spec: Glossary → Invariant: "recorded as invariant_enforced
+            # decisions that propagate to child plans."
+            self._propagate_invariant_decisions(
+                child_plan_id=subplan_id,
+                parent_invariant_decisions=parent_invariant_decisions,
+            )
+
         # Attach subplan statuses to the parent plan for lifecycle tracking
         parent_plan.subplan_statuses = list(parent_plan.subplan_statuses) + statuses
 
@@ -293,6 +332,7 @@ class SubplanService:
                 "root_plan_id": root_id,
                 "count": len(statuses),
                 "execution_mode": mode,
+                "invariant_decisions_propagated": len(parent_invariant_decisions),
             },
         )
 
@@ -304,6 +344,40 @@ class SubplanService:
             child_plans=child_plans,
         )
 
+    # ------------------------------------------------------------------
+    # _propagate_invariant_decisions
+    # ------------------------------------------------------------------
+
+    def _propagate_invariant_decisions(
+        self,
+        child_plan_id: str,
+        parent_invariant_decisions: list[Decision],
+    ) -> None:
+        """Propagate invariant_enforced decisions from parent to child plan.
+
+        Re-records each parent ``invariant_enforced`` decision on the child
+        plan's decision tree.  The child plan starts Strategize with the same
+        invariant set as the parent, satisfying the spec requirement:
+        "recorded as ``invariant_enforced`` decisions that propagate to child
+        plans."
+
+        Args:
+            child_plan_id: ULID of the child plan to record decisions on.
+            parent_invariant_decisions: The parent plan's ``invariant_enforced``
+                decisions to propagate.
+        """
+        for parent_decision in parent_invariant_decisions:
+            self._decision_service.record_decision(
+                plan_id=child_plan_id,
+                decision_type=DecisionType.INVARIANT_ENFORCED,
+                question=parent_decision.question,
+                chosen_option=parent_decision.chosen_option,
+                rationale=parent_decision.rationale,
+                context_snapshot=parent_decision.context_snapshot,
+                confidence_score=parent_decision.confidence_score,
+                alternatives_considered=list(parent_decision.alternatives_considered),
+            )
+
     # ------------------------------------------------------------------
     # validate_spawn
     # ------------------------------------------------------------------