fix(invariant): add action_name param to get_effective_invariants() — include action-scoped invariants in 4-tier merge #3329

Open
freemo wants to merge 3 commits from fix/invariant-service-action-scope-effective into master
5 changed files with 568 additions and 40 deletions
@@ -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
@@ -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."""
Outdated
Review

Suggestion: step_get_effective_only_action and step_get_effective_action_filter are defined but not exercised by any scenario. Consider removing these dead-code step definitions.

Suggestion: `step_get_effective_only_action` and `step_get_effective_action_filter` are defined but not exercised by any scenario. Consider removing these dead-code step definitions.
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}"
@@ -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
1
@@ -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:
@@ -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
Outdated
Review

Specification alignment issue — Suggestion (non-blocking):

Lines 3-5 in the module docstring still describe: 'When an action is used, its invariants are promoted to plan-level.' This conflicts with the new 4-tier model where ACTION has its own intermediate precedence position.

Update this to remove the promotion language and instead state that ACTION invariants participate directly in the merge chain.

**Specification alignment issue — Suggestion (non-blocking):** Lines 3-5 in the module docstring still describe: 'When an action is used, its invariants are promoted to plan-level.' This conflicts with the new 4-tier model where ACTION has its own intermediate precedence position. Update this to remove the promotion language and instead state that ACTION invariants participate directly in the merge chain.
@@ -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