fix(invariant): add action_name param to get_effective_invariants() — include action-scoped invariants in 4-tier merge
CI / quality (pull_request) Successful in 19s
CI / helm (pull_request) Successful in 24s
CI / push-validation (pull_request) Successful in 26s
CI / lint (pull_request) Successful in 47s
CI / typecheck (pull_request) Successful in 51s
CI / security (pull_request) Successful in 52s
CI / e2e_tests (pull_request) Successful in 3m5s
CI / build (pull_request) Successful in 3m17s
CI / unit_tests (pull_request) Failing after 6m41s
CI / docker (pull_request) Has been skipped
CI / integration_tests (pull_request) Successful in 9m53s
CI / coverage (pull_request) Successful in 11m14s
CI / status-check (pull_request) Failing after 1s

- Add `action_name: str | None = None` parameter to `InvariantService.get_effective_invariants()`
- Implement collection of action-scoped invariants filtered by `action_name` within the method
- Update `merge_invariants()` and `InvariantSet.merge()` to accept 4 tiers (plan > action > project > global)
- Update `list_invariants(effective=True)` path to pass `action_name` context through to `get_effective_invariants()`
- Update benchmarks, robot helper, and BDD step definitions to use 4-parameter merge signature
- Add BDD feature file with 12 scenarios covering action-scoped effective invariant computation
- Update module docstrings to reflect 4-tier precedence chain
- Restore `frozen=True` on all Pydantic invariant domain models (Invariant, InvariantViolation, InvariantEnforcementRecord, InvariantSet)
- Restore `tuple[Invariant, ...]` for InvariantSet.invariants to maintain immutability
- Fix `remove_invariant()` to use `model_copy()` pattern instead of direct mutation
- Restore deleted immutability and hashability test step definitions

Fixes the silent data loss bug where action-scoped invariants were never collected
in `get_effective_invariants()` because the method lacked an `action_name` parameter.
The `--effective` CLI flag now correctly includes action-scoped invariants.

ISSUES CLOSED: #3128
This commit is contained in:
2026-04-14 23:28:51 +00:00
parent abbb830c60
commit 7aa28bb5f4
8 changed files with 576 additions and 20 deletions
+5 -5
View File
@@ -74,7 +74,7 @@ class MergeSmallSuite:
def time_merge_small(self) -> None: def time_merge_small(self) -> None:
"""Benchmark merge with ~13 invariants.""" """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: class MergeMediumSuite:
@@ -88,7 +88,7 @@ class MergeMediumSuite:
def time_merge_medium(self) -> None: def time_merge_medium(self) -> None:
"""Benchmark merge with ~50 invariants.""" """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: class MergeLargeSuite:
@@ -102,7 +102,7 @@ class MergeLargeSuite:
def time_merge_large(self) -> None: def time_merge_large(self) -> None:
"""Benchmark merge with ~250 invariants.""" """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: class MergeDeduplicationSuite:
@@ -137,7 +137,7 @@ class MergeDeduplicationSuite:
def time_merge_dedup(self) -> None: def time_merge_dedup(self) -> None:
"""Benchmark merge with 60 invariants, all duplicates.""" """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: class InvariantSetMergeSuite:
@@ -151,7 +151,7 @@ class InvariantSetMergeSuite:
def time_invariant_set_merge(self) -> None: def time_invariant_set_merge(self) -> None:
"""Benchmark InvariantSet.merge().""" """Benchmark InvariantSet.merge()."""
InvariantSet.merge(self.plan, self.project, self.global_invs) InvariantSet.merge(self.plan, [], self.project, self.global_invs)
class ServiceEffectiveSuite: class ServiceEffectiveSuite:
@@ -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,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}'"
)
+2
View File
@@ -160,6 +160,7 @@ def step_plan_invariants_inactive(context):
def step_merge(context): def step_merge(context):
context.merged = merge_invariants( context.merged = merge_invariants(
getattr(context, "plan_invariants", []), getattr(context, "plan_invariants", []),
getattr(context, "action_invariants", []),
getattr(context, "project_invariants", []), getattr(context, "project_invariants", []),
getattr(context, "global_invariants", []), getattr(context, "global_invariants", []),
) )
@@ -189,6 +190,7 @@ def step_merged_scope(context, idx, scope):
def step_merge_invariant_set(context): def step_merge_invariant_set(context):
inv_set = InvariantSet.merge( inv_set = InvariantSet.merge(
getattr(context, "plan_invariants", []), getattr(context, "plan_invariants", []),
getattr(context, "action_invariants", []),
getattr(context, "project_invariants", []), getattr(context, "project_invariants", []),
getattr(context, "global_invariants", []), getattr(context, "global_invariants", []),
) )
+2
View File
@@ -867,6 +867,7 @@ def invariants_enforced_during_strategize() -> None:
merged = merge_invariants( merged = merge_invariants(
plan_invariants=[plan_inv], plan_invariants=[plan_inv],
action_invariants=[],
project_invariants=[project_inv], project_invariants=[project_inv],
global_invariants=[global_inv], global_invariants=[global_inv],
) )
@@ -890,6 +891,7 @@ def invariants_enforced_during_strategize() -> None:
invariant_set = InvariantSet.merge( invariant_set = InvariantSet.merge(
plan_invariants=[plan_inv], plan_invariants=[plan_inv],
action_invariants=[],
project_invariants=[project_inv], project_invariants=[project_inv],
global_invariants=[global_inv], global_invariants=[global_inv],
) )
@@ -11,7 +11,7 @@ a dict keyed by invariant ID.
## Merge Precedence ## 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. See ``merge_invariants`` for de-duplication semantics.
Based on ``docs/specification.md`` and implementation plan Stage M3.5. Based on ``docs/specification.md`` and implementation plan Stage M3.5.
@@ -124,6 +124,7 @@ class InvariantService:
if effective: if effective:
return self.get_effective_invariants( return self.get_effective_invariants(
plan_id=source_name if scope == InvariantScope.PLAN else None, 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, project_name=source_name if scope == InvariantScope.PROJECT else None,
) )
@@ -167,16 +168,19 @@ class InvariantService:
def get_effective_invariants( def get_effective_invariants(
self, self,
plan_id: str | None = None, plan_id: str | None = None,
action_name: str | None = None,
project_name: str | None = None, project_name: str | None = None,
) -> list[Invariant]: ) -> 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 Collects active invariants from each scope tier and merges them
using plan > project > global precedence. using plan > action > project > global precedence.
Args: Args:
plan_id: Optional plan identifier to collect plan-scoped plan_id: Optional plan identifier to collect plan-scoped
invariants. invariants.
action_name: Optional action name to collect action-scoped
invariants.
project_name: Optional project name to collect project-scoped project_name: Optional project name to collect project-scoped
invariants. invariants.
@@ -191,6 +195,12 @@ class InvariantService:
if inv.scope == InvariantScope.PLAN if inv.scope == InvariantScope.PLAN
and (plan_id is None or inv.source_name == plan_id) 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 = [ project_invs = [
inv inv
for inv in active for inv in active
@@ -199,7 +209,7 @@ class InvariantService:
] ]
global_invs = [inv for inv in active if inv.scope == InvariantScope.GLOBAL] 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( def enforce_invariants(
self, self,
@@ -176,6 +176,7 @@ def list_invariants(
agents invariant list agents invariant list
agents invariant list --global agents invariant list --global
agents invariant list --effective --project myapp agents invariant list --effective --project myapp
agents invariant list --effective --action deploy-service
agents invariant list "data.*safe" agents invariant list "data.*safe"
""" """
try: try:
@@ -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 global, project, action, or plan level. When an action is used, its
invariants are promoted to plan-level. The runtime precedence chain is: 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 They are reconciled by the Invariant Reconciliation Actor at the start
of Strategize and recorded as ``invariant_enforced`` decisions. of Strategize and recorded as ``invariant_enforced`` decisions.
@@ -21,8 +21,11 @@ of Strategize and recorded as ``invariant_enforced`` decisions.
## Merge Precedence ## Merge Precedence
When computing the effective set of invariants for a plan, the merge When computing the effective set of invariants for a plan, the merge
order is **plan > project > global**. Duplicate texts (case-insensitive) order is **plan > action > project > global**. Action-scoped invariants
are de-duplicated, keeping the highest-precedence copy. 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. Based on ``docs/specification.md`` and implementation plan Stage M3.5.
""" """
@@ -39,8 +42,9 @@ from ulid import ULID
class InvariantScope(StrEnum): class InvariantScope(StrEnum):
"""Scope at which an invariant applies. """Scope at which an invariant applies.
Precedence (highest to lowest): PLAN > PROJECT > GLOBAL. Precedence (highest to lowest): PLAN > ACTION > PROJECT > GLOBAL.
ACTION invariants are promoted to PLAN scope at ``plan use`` time. ACTION invariants participate directly in the 4-tier merge chain,
sitting between PLAN and PROJECT in precedence order.
""" """
GLOBAL = "global" GLOBAL = "global"
@@ -137,10 +141,11 @@ class InvariantSet(BaseModel):
def merge( def merge(
cls, cls,
plan_invariants: list[Invariant], plan_invariants: list[Invariant],
action_invariants: list[Invariant],
project_invariants: list[Invariant], project_invariants: list[Invariant],
global_invariants: list[Invariant], global_invariants: list[Invariant],
) -> InvariantSet: ) -> 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 De-duplicates by text (case-insensitive), keeping the copy from
the highest-precedence tier. Within each tier, source ordering the highest-precedence tier. Within each tier, source ordering
@@ -148,7 +153,8 @@ class InvariantSet(BaseModel):
Args: Args:
plan_invariants: Plan-level invariants (highest precedence). 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). global_invariants: Global-level invariants (lowest precedence).
Returns: Returns:
@@ -156,7 +162,12 @@ class InvariantSet(BaseModel):
""" """
return cls( return cls(
invariants=tuple( 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( def merge_invariants(
plan_invariants: list[Invariant], plan_invariants: list[Invariant],
action_invariants: list[Invariant],
project_invariants: list[Invariant], project_invariants: list[Invariant],
global_invariants: list[Invariant], global_invariants: list[Invariant],
) -> 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 De-duplicates by text (case-insensitive). The first occurrence
(from the highest-precedence tier) wins. Within each tier, the (from the highest-precedence tier) wins. Within each tier, the
@@ -176,7 +188,8 @@ def merge_invariants(
Args: Args:
plan_invariants: Plan-level invariants (highest precedence). 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). global_invariants: Global-level invariants (lowest precedence).
Returns: Returns:
@@ -185,7 +198,13 @@ def merge_invariants(
seen: set[str] = set() seen: set[str] = set()
result: list[Invariant] = [] 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: for inv in inv_list:
if not inv.active: if not inv.active:
continue continue