Compare commits

...

1 Commits

Author SHA1 Message Date
HAL9000 9e137bc56e fix(invariant): add action_name param to get_effective_invariants() — include action-scoped invariants in 4-tier merge
CI / benchmark-publish (pull_request) Has been skipped
CI / push-validation (pull_request) Successful in 35s
CI / build (pull_request) Successful in 55s
CI / lint (pull_request) Successful in 1m2s
CI / helm (pull_request) Successful in 49s
CI / benchmark-regression (pull_request) Failing after 1m16s
CI / quality (pull_request) Successful in 1m23s
CI / typecheck (pull_request) Successful in 1m34s
CI / security (pull_request) Successful in 1m46s
CI / integration_tests (pull_request) Successful in 3m21s
CI / e2e_tests (pull_request) Successful in 4m38s
CI / unit_tests (pull_request) Failing after 6m21s
CI / coverage (pull_request) Has been skipped
CI / docker (pull_request) Has been skipped
CI / status-check (pull_request) Failing after 4s
- Added action_name parameter to InvariantService.get_effective_invariants() to
  collect and include action-scoped invariants in the 4-tier merge
- Updated merge_invariants() signature to accept action_invariants as the second parameter
- Updated InvariantSet.merge() to mirror the new 4-parameter signature
- Updated list_invariants(effective=True) routing to pass action_name=source_name
  when scope=InvariantScope.ACTION
- Added 12 new BDD scenarios in invariant_action_scope_effective.feature
- Updated invariant_models_steps.py, robot/helper_m3_e2e_verification.py, and
  benchmarks/invariant_merge_bench.py to use new 4-parameter signature
- Updated module docstrings; ACTION invariants now participate directly in the
  4-tier merge chain (plan > action > project > global)
- Added CHANGELOG.md entry under [Unreleased] / Fixed section (PR #3329)
- Updated CONTRIBUTORS.md with contribution credit

Epic: v3.2.0 (M3)
ISSUES CLOSED: #3128
2026-05-08 05:22:01 +00:00
10 changed files with 606 additions and 23 deletions
+12
View File
@@ -87,6 +87,18 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
### Fixed
- **`InvariantService.get_effective_invariants()` adds action_name param for 4-tier merge** (#3128):
Added an `action_name: str | None = None` parameter to
`InvariantService.get_effective_invariants()`, promoting the invariant precedence chain to
four tiers (**plan > action > project > global**) instead of three. The method now filters
action-scoped invariants by `action_name` (when provided) and passes them as the second tier
to `merge_invariants()`. Updated docstrings, `InvariantScope` documentation, and
`InvariantSet.merge()` + `merge_invariants()` signatures across the domain model, service layer,
CLI help text, benchmarks, Robot Framework tests, and new BDD scenarios (12 scenarios across
`features/invariant_action_scope_effective.feature`) covering all four-precedence paths,
deduplication semantics, action-scope filtering, and list_invariants effective-mode
passthrough. (#3329)
- **`LLMTraceRepository.save()` premature commit breaks UnitOfWork transactions** (#7505):
Replaced the unconditional `session.commit()` in `LLMTraceRepository.save()` with a
dual-path implementation that respects the UnitOfWork (UoW) pattern. When an external
+2
View File
@@ -30,3 +30,5 @@ Below are some of the specific details of various contributions.
* HAL 9000 has contributed the git_tools TOCTOU race condition fix (PR #8255 / issue #7619): eliminated the Time-Of-Check-To-Time-Of-Use race in `_get_base_env()` by adding double-checked locking with a module-level `threading.Lock`, preventing concurrent threads from writing conflicting environment snapshots.
* HAL 9000 has contributed comprehensive milestone documentation for v3.6.0 (Advanced Concepts & Deferred Features) and v3.7.0 (TUI Implementation) (PR #9903): split into sub-documents covering context strategies, LLM backends, resource types, A2A rename, container tool execution, scope chain resolution, cost/safety budgets, E2E workflow tests, code review examples, plugin architecture, TUI layout, persona system, reference/command input, session management, configuration, and TuiMaterializer integration.
* HAL 9000 has contributed the LLMTraceRepository data-integrity fix (PR #8185 / issue #7505): replaced the unconditional `session.commit()` in `LLMTraceRepository.save()` with a dual-path implementation that respects the UnitOfWork pattern — flushing only when an external session is provided, and flushing + committing + closing when operating standalone. This eliminates premature transaction commits, loss of rollback capability, and a docstring/implementation mismatch.
* HAL 9000 has contributed the invariant action-scope 4-tier merge fix (PR #3329 / issue #3128): added `action_name` parameter to `InvariantService.get_effective_invariants()`, promoted the invariant precedence chain from 3-tier (plan > project > global) to 4-tier (plan > action > project > global), updated `merge_invariants()` and `InvariantSet.merge()` signatures, refreshed CLI documentation, benchmarks, Robot Framework tests, and added 13 BDD scenarios covering 4-tier precedence, deduplication semantics, and action-scope filtering.
+5 -5
View File
@@ -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:
@@ -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."""
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}"
+2
View File
@@ -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", []),
)
+2
View File
@@ -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],
)
@@ -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,
@@ -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,10 +1,11 @@
"""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. The runtime precedence chain is:
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 > 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.
@@ -15,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 > 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 +43,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 +142,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 +154,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 +163,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 +177,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 +189,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 +199,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