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..7c44c037e --- /dev/null +++ b/features/steps/invariant_action_scope_effective_steps.py @@ -0,0 +1,425 @@ +"""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}'" + ) + + +# ================================================================ +# Thens — InvariantSet.merge() assertions +# ================================================================ + + +@then("the invariant set should have {count:d} invariant") +def step_invariant_set_count_singular(context: object, count: int) -> None: + """Assert the InvariantSet has exactly the given number of invariants (singular).""" + actual = len(context.invariant_set.invariants) + assert actual == count, f"Expected {count} invariant in set, got {actual}" diff --git a/src/cleveragents/application/services/invariant_service.py b/src/cleveragents/application/services/invariant_service.py index 9427a2c40..dfda6cba3 100644 --- a/src/cleveragents/application/services/invariant_service.py +++ b/src/cleveragents/application/services/invariant_service.py @@ -11,8 +11,8 @@ a dict keyed by invariant ID. ## Merge Precedence -Effective invariants are computed using plan > action > project > global -order. See ``merge_invariants`` for de-duplication semantics. +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. """ @@ -180,10 +180,7 @@ class InvariantService: plan_id: Optional plan identifier to collect plan-scoped invariants. action_name: Optional action name to collect action-scoped - invariants (promoted to plan-level during reconciliation). - When ``None``, the action tier is omitted for backward - compatibility. Pass ``"*"`` to include all action-scoped - invariants regardless of source name. + invariants. project_name: Optional project name to collect project-scoped invariants. @@ -202,8 +199,7 @@ class InvariantService: inv for inv in active if inv.scope == InvariantScope.ACTION - and action_name is not None # Only include when explicitly requested - and (inv.source_name == action_name or action_name == "*") + and (action_name is None or inv.source_name == action_name) ] project_invs = [ inv 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 feddf0f13..3cd642297 100644 --- a/src/cleveragents/domain/models/core/invariant.py +++ b/src/cleveragents/domain/models/core/invariant.py @@ -1,8 +1,8 @@ """Invariant domain models for CleverAgents. 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 and participate in the merge. +global, project, action, or plan level. ACTION invariants participate +directly in the 4-tier merge chain rather than being promoted to plan-level. The runtime precedence chain is: plan > action > project > global @@ -16,14 +16,17 @@ of Strategize and recorded as ``invariant_enforced`` decisions. |------------|----------------------------------------------------| | ``GLOBAL`` | Applies to every plan in the system | | ``PROJECT``| Applies to plans targeting a specific project | -| ``ACTION`` | Defined in an action template; promoted on ``use`` | +| ``ACTION`` | Defined in an action template; between PLAN and PROJECT in precedence | | ``PLAN`` | Attached directly to a specific plan | ## Merge Precedence When computing the effective set of invariants for a plan, the merge -order is **plan > action > 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. """ @@ -41,9 +44,8 @@ class InvariantScope(StrEnum): """Scope at which an invariant applies. Precedence (highest to lowest): PLAN > ACTION > PROJECT > GLOBAL. - ACTION invariants are promoted to plan-level and participate in the - merge during reconciliation, sitting between PLAN and PROJECT in the - precedence chain. + ACTION invariants participate directly in the 4-tier merge chain, + sitting between PLAN and PROJECT in precedence order. """ GLOBAL = "global" @@ -140,9 +142,9 @@ class InvariantSet(BaseModel): def merge( cls, plan_invariants: list[Invariant], - action_invariants: list[Invariant] | None = None, - project_invariants: list[Invariant] | None = None, - global_invariants: list[Invariant] | None = None, + action_invariants: list[Invariant], + project_invariants: list[Invariant], + global_invariants: list[Invariant], ) -> InvariantSet: """Merge invariants respecting plan > action > project > global precedence. @@ -152,10 +154,8 @@ class InvariantSet(BaseModel): Args: plan_invariants: Plan-level invariants (highest precedence). - action_invariants: Action-scoped invariants (second-highest - precedence; promoted to plan-level when an - action is used). - 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: @@ -165,9 +165,9 @@ class InvariantSet(BaseModel): invariants=tuple( merge_invariants( plan_invariants, - action_invariants or [], - project_invariants or [], - global_invariants or [], + action_invariants, + project_invariants, + global_invariants, ) ) ) @@ -177,9 +177,9 @@ class InvariantSet(BaseModel): def merge_invariants( plan_invariants: list[Invariant], - action_invariants: list[Invariant] | None = None, - project_invariants: list[Invariant] | None = None, - global_invariants: list[Invariant] | None = None, + action_invariants: list[Invariant], + project_invariants: list[Invariant], + global_invariants: list[Invariant], ) -> list[Invariant]: """Merge invariants implementing plan > action > project > global precedence. @@ -189,13 +189,9 @@ def merge_invariants( Args: plan_invariants: Plan-level invariants (highest precedence). - action_invariants: Action-scoped invariants (second-highest - precedence; promoted to plan level when an action is used). - Defaults to empty list when ``None``. - project_invariants: Project-level invariants. Defaults to empty - list when ``None``. + action_invariants: Action-level invariants (second precedence). + project_invariants: Project-level invariants (third precedence). global_invariants: Global-level invariants (lowest precedence). - Defaults to empty list when ``None``. Returns: A de-duplicated list of invariants in precedence order. @@ -203,12 +199,13 @@ def merge_invariants( seen: set[str] = set() result: list[Invariant] = [] - for inv_list in ( + tiers = ( plan_invariants, - action_invariants or [], - project_invariants or [], - global_invariants or [], - ): + action_invariants, + project_invariants, + global_invariants, + ) + for inv_list in tiers: for inv in inv_list: if not inv.active: continue