diff --git a/benchmarks/invariant_merge_bench.py b/benchmarks/invariant_merge_bench.py index d68db24cf..87e5e853b 100644 --- a/benchmarks/invariant_merge_bench.py +++ b/benchmarks/invariant_merge_bench.py @@ -74,7 +74,7 @@ class MergeSmallSuite: def time_merge_small(self) -> None: """Benchmark merge with ~13 invariants.""" - merge_invariants(self.plan, self.project, self.global_invs) + merge_invariants(self.plan, [], self.project, self.global_invs) class MergeMediumSuite: @@ -88,7 +88,7 @@ class MergeMediumSuite: def time_merge_medium(self) -> None: """Benchmark merge with ~50 invariants.""" - merge_invariants(self.plan, self.project, self.global_invs) + merge_invariants(self.plan, [], self.project, self.global_invs) class MergeLargeSuite: @@ -102,7 +102,7 @@ class MergeLargeSuite: def time_merge_large(self) -> None: """Benchmark merge with ~250 invariants.""" - merge_invariants(self.plan, self.project, self.global_invs) + merge_invariants(self.plan, [], self.project, self.global_invs) class MergeDeduplicationSuite: @@ -137,7 +137,7 @@ class MergeDeduplicationSuite: def time_merge_dedup(self) -> None: """Benchmark merge with 60 invariants, all duplicates.""" - merge_invariants(self.plan, self.project, self.global_invs) + merge_invariants(self.plan, [], self.project, self.global_invs) class InvariantSetMergeSuite: @@ -151,7 +151,7 @@ class InvariantSetMergeSuite: def time_invariant_set_merge(self) -> None: """Benchmark InvariantSet.merge().""" - InvariantSet.merge(self.plan, self.project, self.global_invs) + InvariantSet.merge(self.plan, [], self.project, self.global_invs) class ServiceEffectiveSuite: diff --git a/features/invariant_action_scope_effective.feature b/features/invariant_action_scope_effective.feature new file mode 100644 index 000000000..2fa2a5a78 --- /dev/null +++ b/features/invariant_action_scope_effective.feature @@ -0,0 +1,109 @@ +# BDD scenarios for issue #3128 — action-scoped invariants in get_effective_invariants() +# +# Verifies that InvariantService.get_effective_invariants() correctly includes +# action-scoped invariants in the 4-tier precedence merge: +# plan > action > project > global +# +# Also verifies that merge_invariants() and InvariantSet.merge() accept and +# correctly process the action_invariants parameter. +# +# See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/3128 + +Feature: Action-scoped invariants included in effective invariant computation + As a developer using InvariantService + I want get_effective_invariants() to include action-scoped invariants + So that the 4-tier precedence chain (plan > action > project > global) is fully enforced + + # ================================================================ + # get_effective_invariants() — action_name parameter + # ================================================================ + + Scenario: get_effective_invariants returns action-scoped invariants when action_name provided + Given an invariant service with action and global invariants + When I get effective invariants with action_name "deploy-service" + Then the effective set should contain the action invariant "Do not modify public API" + And the effective set should contain the global invariant "Log all changes" + And the effective set should have 2 invariants + + Scenario: get_effective_invariants includes all action invariants when action_name is None + Given an invariant service with action and global invariants + When I get effective invariants with no action_name + Then the effective set should contain the action invariant "Do not modify public API" + And the effective set should contain the global invariant "Log all changes" + And the effective set should have 2 invariants + + Scenario: get_effective_invariants with all four scopes returns all invariants + Given an invariant service with invariants at all four scopes + When I get effective invariants for plan "plan-001" action "deploy-service" and project "myapp" + Then the effective set should have 4 invariants + And the effective set should contain the plan invariant "Plan-level rule" + And the effective set should contain the action invariant "Action-level rule" + And the effective set should contain the project invariant "Project-level rule" + And the effective set should contain the global invariant "Global-level rule" + + Scenario: Action invariants have higher precedence than project invariants + Given an invariant service with action and project invariants having the same text + When I get effective invariants for plan "plan-001" action "deploy-service" and project "myapp" + Then the effective set should have 1 invariant + And the first effective invariant should have scope "action" + + Scenario: Plan invariants have higher precedence than action invariants + Given an invariant service with plan and action invariants having the same text + When I get effective invariants for plan "plan-001" action "deploy-service" and project "myapp" + Then the effective set should have 1 invariant + And the first effective invariant should have scope "plan" + + Scenario: get_effective_invariants filters action invariants by action_name + Given an invariant service with two different action invariants + When I get effective invariants with action_name "deploy-service" + Then the effective set should contain the action invariant "Deploy constraint" + And the effective set should not contain the action invariant "Build constraint" + And the effective set should have 1 invariant + + # ================================================================ + # list_invariants(effective=True) — action scope passthrough + # ================================================================ + + Scenario: list_invariants with effective=True and action scope passes action_name + Given an invariant service with action and global invariants + When I list invariants with effective True and action scope "deploy-service" + Then the listed effective set should contain the action invariant "Do not modify public API" + And the listed effective set should contain the global invariant "Log all changes" + + # ================================================================ + # merge_invariants() — 4-tier signature + # ================================================================ + + Scenario: merge_invariants accepts action_invariants as second parameter + Given I have plan action project and global invariant lists + When I merge all four invariant tiers + Then the merged result should have 4 invariants + And the merged result should contain "Plan rule" + And the merged result should contain "Action rule" + And the merged result should contain "Project rule" + And the merged result should contain "Global rule" + + Scenario: merge_invariants deduplicates action invariants by text + Given I have overlapping action and project invariant lists + When I merge all four invariant tiers + Then the merged result should have 1 invariant + And the first merged invariant should have scope "action" + + Scenario: merge_invariants with empty action list behaves like 3-tier merge + Given I have plan project and global invariant lists with no action invariants + When I merge all four invariant tiers with empty action list + Then the merged result should have 3 invariants + + # ================================================================ + # InvariantSet.merge() — 4-tier signature + # ================================================================ + + Scenario: InvariantSet.merge accepts action_invariants as second parameter + Given I have plan action project and global invariant lists + When I merge all four tiers using InvariantSet + Then the invariant set should have 4 invariants + + Scenario: InvariantSet.merge with empty action list produces correct result + Given I have plan project and global invariant lists with no action invariants + When I merge all four tiers using InvariantSet with empty action list + Then the invariant set should have 3 invariants diff --git a/features/steps/invariant_action_scope_effective_steps.py b/features/steps/invariant_action_scope_effective_steps.py new file mode 100644 index 000000000..48bdff80d --- /dev/null +++ b/features/steps/invariant_action_scope_effective_steps.py @@ -0,0 +1,413 @@ +"""Step definitions for invariant_action_scope_effective.feature. + +Tests that InvariantService.get_effective_invariants() correctly includes +action-scoped invariants in the 4-tier precedence merge: + plan > action > project > global + +Also tests that merge_invariants() and InvariantSet.merge() accept and +correctly process the action_invariants parameter. + +See: https://git.cleverthis.com/cleveragents/cleveragents-core/issues/3128 +""" + +from __future__ import annotations + +from behave import given, then, when # type: ignore[import-untyped] + +from cleveragents.application.services.invariant_service import InvariantService +from cleveragents.domain.models.core.invariant import ( + Invariant, + InvariantScope, + InvariantSet, + merge_invariants, +) + +# ================================================================ +# Givens — service setup +# ================================================================ + + +@given("an invariant service with action and global invariants") +def step_service_action_global(context: object) -> None: + """Set up a service with one action-scoped and one global invariant.""" + context.service = InvariantService() + context.service.add_invariant( + text="Do not modify public API", + scope=InvariantScope.ACTION, + source_name="deploy-service", + ) + context.service.add_invariant( + text="Log all changes", + scope=InvariantScope.GLOBAL, + source_name="system", + ) + + +@given("an invariant service with invariants at all four scopes") +def step_service_all_four_scopes(context: object) -> None: + """Set up a service with one invariant at each of the four scopes.""" + context.service = InvariantService() + context.service.add_invariant( + text="Plan-level rule", + scope=InvariantScope.PLAN, + source_name="plan-001", + ) + context.service.add_invariant( + text="Action-level rule", + scope=InvariantScope.ACTION, + source_name="deploy-service", + ) + context.service.add_invariant( + text="Project-level rule", + scope=InvariantScope.PROJECT, + source_name="myapp", + ) + context.service.add_invariant( + text="Global-level rule", + scope=InvariantScope.GLOBAL, + source_name="system", + ) + + +@given("an invariant service with action and project invariants having the same text") +def step_service_action_project_same_text(context: object) -> None: + """Set up a service where action and project invariants share the same text.""" + context.service = InvariantService() + context.service.add_invariant( + text="Shared constraint text", + scope=InvariantScope.ACTION, + source_name="deploy-service", + ) + context.service.add_invariant( + text="Shared constraint text", + scope=InvariantScope.PROJECT, + source_name="myapp", + ) + + +@given("an invariant service with plan and action invariants having the same text") +def step_service_plan_action_same_text(context: object) -> None: + """Set up a service where plan and action invariants share the same text.""" + context.service = InvariantService() + context.service.add_invariant( + text="Shared constraint text", + scope=InvariantScope.PLAN, + source_name="plan-001", + ) + context.service.add_invariant( + text="Shared constraint text", + scope=InvariantScope.ACTION, + source_name="deploy-service", + ) + + +@given("an invariant service with two different action invariants") +def step_service_two_action_invariants(context: object) -> None: + """Set up a service with two action invariants from different actions.""" + context.service = InvariantService() + context.service.add_invariant( + text="Deploy constraint", + scope=InvariantScope.ACTION, + source_name="deploy-service", + ) + context.service.add_invariant( + text="Build constraint", + scope=InvariantScope.ACTION, + source_name="build-service", + ) + + +# ================================================================ +# Whens — get_effective_invariants calls +# ================================================================ + + +@when('I get effective invariants with action_name "{action_name}"') +def step_get_effective_with_action(context: object, action_name: str) -> None: + """Call get_effective_invariants with the given action_name.""" + context.effective = context.service.get_effective_invariants( + action_name=action_name, + ) + + +@when("I get effective invariants with no action_name") +def step_get_effective_no_action(context: object) -> None: + """Call get_effective_invariants without an action_name.""" + context.effective = context.service.get_effective_invariants() + + +@when( + 'I get effective invariants for plan "{plan_id}" action "{action_name}" and project "{project_name}"' +) +def step_get_effective_all_four( + context: object, plan_id: str, action_name: str, project_name: str +) -> None: + """Call get_effective_invariants with all four scope parameters.""" + context.effective = context.service.get_effective_invariants( + plan_id=plan_id, + action_name=action_name, + project_name=project_name, + ) + + +@when('I get effective invariants with only action_name "{action_name}"') +def step_get_effective_only_action(context: object, action_name: str) -> None: + """Call get_effective_invariants with only action_name (no plan or project).""" + context.effective = context.service.get_effective_invariants( + action_name=action_name, + ) + + +@when('I get effective invariants with action_name "{action_name}" for filtering') +def step_get_effective_action_filter(context: object, action_name: str) -> None: + """Call get_effective_invariants to filter by specific action_name.""" + context.effective = context.service.get_effective_invariants( + action_name=action_name, + ) + + +# ================================================================ +# Whens — list_invariants with effective=True +# ================================================================ + + +@when('I list invariants with effective True and action scope "{action_name}"') +def step_list_effective_action(context: object, action_name: str) -> None: + """Call list_invariants with effective=True and action scope.""" + context.listed_effective = context.service.list_invariants( + scope=InvariantScope.ACTION, + source_name=action_name, + effective=True, + ) + + +# ================================================================ +# Whens — merge_invariants() direct calls +# ================================================================ + + +@given("I have plan action project and global invariant lists") +def step_four_tier_lists(context: object) -> None: + """Create four separate invariant lists for merge testing.""" + context.plan_invs = [ + Invariant(text="Plan rule", scope=InvariantScope.PLAN, source_name="plan-001") + ] + context.action_invs = [ + Invariant( + text="Action rule", + scope=InvariantScope.ACTION, + source_name="deploy-service", + ) + ] + context.project_invs = [ + Invariant( + text="Project rule", scope=InvariantScope.PROJECT, source_name="myapp" + ) + ] + context.global_invs = [ + Invariant(text="Global rule", scope=InvariantScope.GLOBAL, source_name="system") + ] + + +@given("I have overlapping action and project invariant lists") +def step_overlapping_action_project(context: object) -> None: + """Create action and project invariant lists with the same text.""" + context.plan_invs: list[Invariant] = [] + context.action_invs = [ + Invariant( + text="Shared rule", + scope=InvariantScope.ACTION, + source_name="deploy-service", + ) + ] + context.project_invs = [ + Invariant(text="Shared rule", scope=InvariantScope.PROJECT, source_name="myapp") + ] + context.global_invs: list[Invariant] = [] + + +@given("I have plan project and global invariant lists with no action invariants") +def step_three_tier_lists(context: object) -> None: + """Create three invariant lists (no action tier) for merge testing.""" + context.plan_invs = [ + Invariant(text="Plan rule", scope=InvariantScope.PLAN, source_name="plan-001") + ] + context.action_invs: list[Invariant] = [] + context.project_invs = [ + Invariant( + text="Project rule", scope=InvariantScope.PROJECT, source_name="myapp" + ) + ] + context.global_invs = [ + Invariant(text="Global rule", scope=InvariantScope.GLOBAL, source_name="system") + ] + + +@when("I merge all four invariant tiers") +def step_merge_four_tiers(context: object) -> None: + """Call merge_invariants with all four tiers.""" + context.merged = merge_invariants( + context.plan_invs, + context.action_invs, + context.project_invs, + context.global_invs, + ) + + +@when("I merge all four invariant tiers with empty action list") +def step_merge_four_tiers_empty_action(context: object) -> None: + """Call merge_invariants with an empty action list.""" + context.merged = merge_invariants( + context.plan_invs, + [], + context.project_invs, + context.global_invs, + ) + + +@when("I merge all four tiers using InvariantSet") +def step_merge_invariant_set_four(context: object) -> None: + """Call InvariantSet.merge with all four tiers.""" + context.invariant_set = InvariantSet.merge( + context.plan_invs, + context.action_invs, + context.project_invs, + context.global_invs, + ) + + +@when("I merge all four tiers using InvariantSet with empty action list") +def step_merge_invariant_set_empty_action(context: object) -> None: + """Call InvariantSet.merge with an empty action list.""" + context.invariant_set = InvariantSet.merge( + context.plan_invs, + [], + context.project_invs, + context.global_invs, + ) + + +# ================================================================ +# Thens — effective set assertions +# ================================================================ + + +@then('the effective set should contain the action invariant "{text}"') +def step_effective_contains_action(context: object, text: str) -> None: + """Assert the effective set contains an action invariant with the given text.""" + texts = [inv.text for inv in context.effective] + assert text in texts, ( + f"Expected action invariant '{text}' in effective set, got: {texts}" + ) + + +@then('the effective set should not contain the action invariant "{text}"') +def step_effective_not_contains_action(context: object, text: str) -> None: + """Assert the effective set does NOT contain an invariant with the given text.""" + texts = [inv.text for inv in context.effective] + assert text not in texts, ( + f"Expected action invariant '{text}' to be absent from effective set, got: {texts}" + ) + + +@then('the effective set should contain the global invariant "{text}"') +def step_effective_contains_global(context: object, text: str) -> None: + """Assert the effective set contains a global invariant with the given text.""" + texts = [inv.text for inv in context.effective] + assert text in texts, ( + f"Expected global invariant '{text}' in effective set, got: {texts}" + ) + + +@then('the effective set should contain the plan invariant "{text}"') +def step_effective_contains_plan(context: object, text: str) -> None: + """Assert the effective set contains a plan invariant with the given text.""" + texts = [inv.text for inv in context.effective] + assert text in texts, ( + f"Expected plan invariant '{text}' in effective set, got: {texts}" + ) + + +@then('the effective set should contain the project invariant "{text}"') +def step_effective_contains_project(context: object, text: str) -> None: + """Assert the effective set contains a project invariant with the given text.""" + texts = [inv.text for inv in context.effective] + assert text in texts, ( + f"Expected project invariant '{text}' in effective set, got: {texts}" + ) + + +@then("the effective set should have {count:d} invariants") +def step_effective_count(context: object, count: int) -> None: + """Assert the effective set has exactly the given number of invariants.""" + actual = len(context.effective) + assert actual == count, f"Expected {count} effective invariants, got {actual}" + + +@then("the effective set should have {count:d} invariant") +def step_effective_count_singular(context: object, count: int) -> None: + """Assert the effective set has exactly the given number of invariants (singular).""" + actual = len(context.effective) + assert actual == count, f"Expected {count} effective invariant, got {actual}" + + +@then('the first effective invariant should have scope "{scope}"') +def step_first_effective_scope(context: object, scope: str) -> None: + """Assert the first invariant in the effective set has the given scope.""" + assert context.effective, "Effective set is empty" + actual = context.effective[0].scope.value + assert actual == scope, ( + f"Expected first effective invariant scope '{scope}', got '{actual}'" + ) + + +# ================================================================ +# Thens — list_invariants effective assertions +# ================================================================ + + +@then('the listed effective set should contain the action invariant "{text}"') +def step_listed_effective_contains_action(context: object, text: str) -> None: + """Assert the listed effective set contains an action invariant with the given text.""" + texts = [inv.text for inv in context.listed_effective] + assert text in texts, ( + f"Expected action invariant '{text}' in listed effective set, got: {texts}" + ) + + +@then('the listed effective set should contain the global invariant "{text}"') +def step_listed_effective_contains_global(context: object, text: str) -> None: + """Assert the listed effective set contains a global invariant with the given text.""" + texts = [inv.text for inv in context.listed_effective] + assert text in texts, ( + f"Expected global invariant '{text}' in listed effective set, got: {texts}" + ) + + +# ================================================================ +# Thens — merge_invariants() assertions +# ================================================================ + + +@then("the merged result should have {count:d} invariants") +def step_merged_count_four(context: object, count: int) -> None: + """Assert the merged result has exactly the given number of invariants.""" + actual = len(context.merged) + assert actual == count, f"Expected {count} merged invariants, got {actual}" + + +@then('the merged result should contain "{text}"') +def step_merged_contains_text(context: object, text: str) -> None: + """Assert the merged result contains an invariant with the given text.""" + texts = [inv.text for inv in context.merged] + assert text in texts, f"Expected '{text}' in merged result, got: {texts}" + + +@then('the first merged invariant should have scope "{scope}"') +def step_first_merged_scope(context: object, scope: str) -> None: + """Assert the first invariant in the merged result has the given scope.""" + assert context.merged, "Merged result is empty" + actual = context.merged[0].scope.value + assert actual == scope, ( + f"Expected first merged invariant scope '{scope}', got '{actual}'" + ) diff --git a/features/steps/invariant_models_steps.py b/features/steps/invariant_models_steps.py index b1909d647..27dffd375 100644 --- a/features/steps/invariant_models_steps.py +++ b/features/steps/invariant_models_steps.py @@ -160,6 +160,7 @@ def step_plan_invariants_inactive(context): def step_merge(context): context.merged = merge_invariants( getattr(context, "plan_invariants", []), + getattr(context, "action_invariants", []), getattr(context, "project_invariants", []), getattr(context, "global_invariants", []), ) @@ -189,6 +190,7 @@ def step_merged_scope(context, idx, scope): def step_merge_invariant_set(context): inv_set = InvariantSet.merge( getattr(context, "plan_invariants", []), + getattr(context, "action_invariants", []), getattr(context, "project_invariants", []), getattr(context, "global_invariants", []), ) diff --git a/robot/helper_m3_e2e_verification.py b/robot/helper_m3_e2e_verification.py index bfd06a911..84e9d78c1 100644 --- a/robot/helper_m3_e2e_verification.py +++ b/robot/helper_m3_e2e_verification.py @@ -867,6 +867,7 @@ def invariants_enforced_during_strategize() -> None: merged = merge_invariants( plan_invariants=[plan_inv], + action_invariants=[], project_invariants=[project_inv], global_invariants=[global_inv], ) @@ -890,6 +891,7 @@ def invariants_enforced_during_strategize() -> None: invariant_set = InvariantSet.merge( plan_invariants=[plan_inv], + action_invariants=[], project_invariants=[project_inv], global_invariants=[global_inv], ) diff --git a/src/cleveragents/application/services/invariant_service.py b/src/cleveragents/application/services/invariant_service.py index e941b999e..dfda6cba3 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/src/cleveragents/application/services/invariant_service.py @@ -11,7 +11,7 @@ a dict keyed by invariant ID. ## Merge Precedence -Effective invariants are computed using plan > project > global order. +Effective invariants are computed using plan > action > project > global order. See ``merge_invariants`` for de-duplication semantics. Based on ``docs/specification.md`` and implementation plan Stage M3.5. @@ -124,6 +124,7 @@ class InvariantService: if effective: return self.get_effective_invariants( plan_id=source_name if scope == InvariantScope.PLAN else None, + action_name=source_name if scope == InvariantScope.ACTION else None, project_name=source_name if scope == InvariantScope.PROJECT else None, ) @@ -167,16 +168,19 @@ class InvariantService: def get_effective_invariants( self, plan_id: str | None = None, + action_name: str | None = None, project_name: str | None = None, ) -> list[Invariant]: - """Return the merged precedence chain for a plan/project context. + """Return the merged precedence chain for a plan/action/project context. Collects active invariants from each scope tier and merges them - using plan > project > global precedence. + using plan > action > project > global precedence. Args: plan_id: Optional plan identifier to collect plan-scoped invariants. + action_name: Optional action name to collect action-scoped + invariants. project_name: Optional project name to collect project-scoped invariants. @@ -191,6 +195,12 @@ class InvariantService: if inv.scope == InvariantScope.PLAN and (plan_id is None or inv.source_name == plan_id) ] + action_invs = [ + inv + for inv in active + if inv.scope == InvariantScope.ACTION + and (action_name is None or inv.source_name == action_name) + ] project_invs = [ inv for inv in active @@ -199,7 +209,7 @@ class InvariantService: ] global_invs = [inv for inv in active if inv.scope == InvariantScope.GLOBAL] - return merge_invariants(plan_invs, project_invs, global_invs) + return merge_invariants(plan_invs, action_invs, project_invs, global_invs) def enforce_invariants( self, diff --git a/src/cleveragents/cli/commands/invariant.py b/src/cleveragents/cli/commands/invariant.py index 2376cb5bd..7c655d72e 100644 --- a/src/cleveragents/cli/commands/invariant.py +++ b/src/cleveragents/cli/commands/invariant.py @@ -176,6 +176,7 @@ def list_invariants( agents invariant list agents invariant list --global agents invariant list --effective --project myapp + agents invariant list --effective --action deploy-service agents invariant list "data.*safe" """ try: diff --git a/src/cleveragents/domain/models/core/invariant.py b/src/cleveragents/domain/models/core/invariant.py index 1880bb668..940187ab1 100644 --- a/src/cleveragents/domain/models/core/invariant.py +++ b/src/cleveragents/domain/models/core/invariant.py @@ -4,7 +4,7 @@ Invariants are natural-language constraints on plan execution, scoped at global, project, action, or plan level. When an action is used, its invariants are promoted to plan-level. The runtime precedence chain is: - plan > project > global + plan > action > project > global They are reconciled by the Invariant Reconciliation Actor at the start of Strategize and recorded as ``invariant_enforced`` decisions. @@ -21,8 +21,11 @@ of Strategize and recorded as ``invariant_enforced`` decisions. ## Merge Precedence When computing the effective set of invariants for a plan, the merge -order is **plan > project > global**. Duplicate texts (case-insensitive) -are de-duplicated, keeping the highest-precedence copy. +order is **plan > action > project > global**. Action-scoped invariants +sit between plan and project in the precedence chain — they override +project and global constraints but are themselves overridden by +plan-level constraints. Duplicate texts (case-insensitive) are +de-duplicated, keeping the highest-precedence copy. Based on ``docs/specification.md`` and implementation plan Stage M3.5. """ @@ -39,8 +42,9 @@ from ulid import ULID class InvariantScope(StrEnum): """Scope at which an invariant applies. - Precedence (highest to lowest): PLAN > PROJECT > GLOBAL. - ACTION invariants are promoted to PLAN scope at ``plan use`` time. + Precedence (highest to lowest): PLAN > ACTION > PROJECT > GLOBAL. + ACTION invariants participate directly in the 4-tier merge chain, + sitting between PLAN and PROJECT in precedence order. """ GLOBAL = "global" @@ -137,10 +141,11 @@ class InvariantSet(BaseModel): def merge( cls, plan_invariants: list[Invariant], + action_invariants: list[Invariant], project_invariants: list[Invariant], global_invariants: list[Invariant], ) -> InvariantSet: - """Merge invariants respecting plan > project > global precedence. + """Merge invariants respecting plan > action > project > global precedence. De-duplicates by text (case-insensitive), keeping the copy from the highest-precedence tier. Within each tier, source ordering @@ -148,7 +153,8 @@ class InvariantSet(BaseModel): Args: plan_invariants: Plan-level invariants (highest precedence). - project_invariants: Project-level invariants. + action_invariants: Action-level invariants (second precedence). + project_invariants: Project-level invariants (third precedence). global_invariants: Global-level invariants (lowest precedence). Returns: @@ -156,7 +162,12 @@ class InvariantSet(BaseModel): """ return cls( invariants=tuple( - merge_invariants(plan_invariants, project_invariants, global_invariants) + merge_invariants( + plan_invariants, + action_invariants, + project_invariants, + global_invariants, + ) ) ) @@ -165,10 +176,11 @@ class InvariantSet(BaseModel): def merge_invariants( plan_invariants: list[Invariant], + action_invariants: list[Invariant], project_invariants: list[Invariant], global_invariants: list[Invariant], ) -> list[Invariant]: - """Merge invariants implementing plan > project > global precedence. + """Merge invariants implementing plan > action > project > global precedence. De-duplicates by text (case-insensitive). The first occurrence (from the highest-precedence tier) wins. Within each tier, the @@ -176,7 +188,8 @@ def merge_invariants( Args: plan_invariants: Plan-level invariants (highest precedence). - project_invariants: Project-level invariants. + action_invariants: Action-level invariants (second precedence). + project_invariants: Project-level invariants (third precedence). global_invariants: Global-level invariants (lowest precedence). Returns: @@ -185,7 +198,13 @@ def merge_invariants( seen: set[str] = set() result: list[Invariant] = [] - for inv_list in (plan_invariants, project_invariants, global_invariants): + tiers = ( + plan_invariants, + action_invariants, + project_invariants, + global_invariants, + ) + for inv_list in tiers: for inv in inv_list: if not inv.active: continue