CHANGELOG.md

@@ -170,6 +170,16 @@ ensuring data is stored with proper parameter values.
   code 1 when no actor is configured.
 
 ### Added
+- **ACMS Context Policy Configuration Loader and Plan Execution Integration**:
+  Implemented `ContextPolicyConfigurationLoader` for loading YAML/TOML policy
+  configurations with full schema validation, and `PlanExecutionACMSIntegration`
+  for wiring ACMS-assembled context into the plan execution engine. Enables
+  flexible, per-view context policy configuration with scope rules, priority
+  weights, and budget overrides. `PlanExecutor` now accepts an optional
+  `acms_integration` parameter (dependency injection) and passes it to
+  `RuntimeExecuteActor`, which uses it to assemble context via ACMS policies
+  before LLM calls instead of passing raw file dumps. (#9584)
+
 
 - **Automated CLI Docstring Example Validation** (#9106): Added `DocstringExampleValidator`
   in `src/cleveragents/cli/docstring_validator.py` that introspects Typer command signatures

CONTRIBUTORS.md

@@ -52,6 +52,7 @@ Below are some of the specific details of various contributions.
 * HAL 9000 has contributed the agent-evolution-pool-supervisor PR metadata assignment (#7888): the supervisor now automatically looks up the Type/Automation label and earliest open milestone before dispatching improvement PR creation workers, ensuring all generated improvement PRs have correct Type labels and milestone assignments.
 * HAL 9000 has contributed the decision recording hook for the Strategize phase (issue #8522): captures every decision point with question, chosen option, alternatives, confidence, rationale, and full context snapshot for replay and correction.
 * HAL 9000 has contributed the ContextStrategy protocol and StrategyRegistry plugin registration system (PR #10590 / issue #8616): implemented the pluggable context assembly strategy protocol with proper type-safe method signatures, created the central thread-safe StrategyRegistry supporting registration, lookup, entry-point discovery, and per-strategy configuration (timeout, fragments limits, workers, circuit breaker threshold). Six built-in strategies implemented and documented: simple-keyword, semantic-embedding, breadth-depth-navigator, arce, temporal-archaeology, and plan-decision-context. Full BDD test coverage including thread safety, boundary validation, and error handling tests. (Part of Epic #8505)
+* HAL9000 has contributed automated implementation of ACMS context policy configuration loader and plan execution integration.
 * This project was made possible thanks to considerable donation of time, money, and resources by CleverThis, Inc.
 * HAL 9000 has contributed automated bug fixes, CLI output formatting improvements, and ongoing maintenance as part of the CleverAgents automation system.
 * HAL 9000 has contributed the pr-review-pool-supervisor tracking prefix documentation fix (#7891): aligned all documentation references from the outdated `AUTO-REV-POOL` prefix to the correct `AUTO-REV-SUP` prefix used in production.

benchmarks/acms_context_policy_bench.py

@@ -0,0 +1,172 @@
+"""ASV benchmarks for ACMS context policy loader and plan execution integration.
+
+Measures the performance of:
+- ContextPolicyConfigurationLoader.load_from_string (YAML and TOML)
+- ACMSContextAssembler.assemble_context with varying policy counts
+- PlanExecutionACMSIntegration.prepare_llm_context with and without policies
+- PolicyScope.matches with scalar and list values
+"""
+
+from __future__ import annotations
+
+import importlib
+import sys
+from pathlib import Path
+
+_SRC = str(Path(__file__).resolve().parents[1] / "src")
+if _SRC not in sys.path:
+    sys.path.insert(0, _SRC)
+
+import cleveragents  # noqa: E402
+
+importlib.reload(cleveragents)
+
+from cleveragents.acms.context_policy_loader import (  # noqa: E402
+    ContextPolicyConfigurationLoader,
+    PolicyScope,
+)
+from cleveragents.acms.plan_execution_integration import (  # noqa: E402
+    ACMSContextAssembler,
+    PlanExecutionACMSIntegration,
+)
+
+_YAML_CONFIG_1 = """
+view_name: bench_view
+policies:
+  - name: policy1
+    priority_weight: 1.0
+    scopes:
+      - name: file_type
+        value: python
+"""
+
+_YAML_CONFIG_10 = """
+view_name: bench_view
+policies:
+""" + "\n".join(f"  - name: policy{i}\n    priority_weight: {i}.0" for i in range(10))
+
+_YAML_CONFIG_100 = """
+view_name: bench_view
+policies:
+""" + "\n".join(f"  - name: policy{i}\n    priority_weight: {i}.0" for i in range(100))
+
+_TOML_CONFIG_1 = """
+view_name = "bench_view"
+[[policies]]
+name = "policy1"
+priority_weight = 1.0
+"""
+
+
+class LoaderSuite:
+    """Benchmark ContextPolicyConfigurationLoader throughput."""
+
+    def setup(self) -> None:
+        """Set up loader instance."""
+        self._loader = ContextPolicyConfigurationLoader()
+
+    def time_load_yaml_1_policy(self) -> None:
+        """Benchmark loading a YAML config with 1 policy."""
+        self._loader.load_from_string(_YAML_CONFIG_1, "yaml")
+
+    def time_load_yaml_10_policies(self) -> None:
+        """Benchmark loading a YAML config with 10 policies."""
+        self._loader.load_from_string(_YAML_CONFIG_10, "yaml")
+
+    def time_load_yaml_100_policies(self) -> None:
+        """Benchmark loading a YAML config with 100 policies."""
+        self._loader.load_from_string(_YAML_CONFIG_100, "yaml")
+
+    def time_load_toml_1_policy(self) -> None:
+        """Benchmark loading a TOML config with 1 policy."""
+        self._loader.load_from_string(_TOML_CONFIG_1, "toml")
+
+
+class AssemblerSuite:
+    """Benchmark ACMSContextAssembler.assemble_context throughput."""
+
+    def setup(self) -> None:
+        """Set up assembler instances with varying policy counts."""
+        loader = ContextPolicyConfigurationLoader()
+
+        config_1 = loader.load_from_string(_YAML_CONFIG_1, "yaml")
+        config_10 = loader.load_from_string(_YAML_CONFIG_10, "yaml")
+        config_100 = loader.load_from_string(_YAML_CONFIG_100, "yaml")
+
+        self._assembler_1 = ACMSContextAssembler(config_1)
+        self._assembler_10 = ACMSContextAssembler(config_10)
+        self._assembler_100 = ACMSContextAssembler(config_100)
+
+        self._raw_context = {
+            "file_type": "python",
+            "path": "src/module.py",
+            "content": "def hello(): pass",
+        }
+
+    def time_assemble_1_policy(self) -> None:
+        """Benchmark assembling context with 1 policy."""
+        self._assembler_1.assemble_context(self._raw_context)
+
+    def time_assemble_10_policies(self) -> None:
+        """Benchmark assembling context with 10 policies."""
+        self._assembler_10.assemble_context(self._raw_context)
+
+    def time_assemble_100_policies(self) -> None:
+        """Benchmark assembling context with 100 policies."""
+        self._assembler_100.assemble_context(self._raw_context)
+
+
+class IntegrationSuite:
+    """Benchmark PlanExecutionACMSIntegration.prepare_llm_context throughput."""
+
+    def setup(self) -> None:
+        """Set up integration instances."""
+        loader = ContextPolicyConfigurationLoader()
+        config = loader.load_from_string(_YAML_CONFIG_10, "yaml")
+
+        self._integration_no_policy = PlanExecutionACMSIntegration()
+        self._integration_with_policy = PlanExecutionACMSIntegration(
+            policy_config=config
+        )
+        self._raw_context = {
+            "file_type": "python",
+            "path": "src/module.py",
+        }
+
+    def time_prepare_context_no_policy(self) -> None:
+        """Benchmark prepare_llm_context with no policy (passthrough)."""
+        self._integration_no_policy.prepare_llm_context(self._raw_context)
+
+    def time_prepare_context_with_policy(self) -> None:
+        """Benchmark prepare_llm_context with ACMS policy assembly."""
+        self._integration_with_policy.prepare_llm_context(self._raw_context)
+
+
+class PolicyScopeSuite:
+    """Benchmark PolicyScope.matches throughput."""
+
+    def setup(self) -> None:
+        """Set up scope instances."""
+        self._scope_scalar = PolicyScope(name="file_type", value="python")
+        self._scope_list = PolicyScope(
+            name="file_type", value=["python", "javascript", "typescript"]
+        )
+        self._context_match = {"file_type": "python"}
+        self._context_no_match = {"file_type": "java"}
+        self._context_list = {"file_type": ["python", "rust"]}
+
+    def time_scope_scalar_match(self) -> None:
+        """Benchmark scalar scope match."""
+        self._scope_scalar.matches(self._context_match)
+
+    def time_scope_scalar_no_match(self) -> None:
+        """Benchmark scalar scope no-match."""
+        self._scope_scalar.matches(self._context_no_match)
+
+    def time_scope_list_match(self) -> None:
+        """Benchmark list scope match."""
+        self._scope_list.matches(self._context_match)
+
+    def time_scope_context_list_match(self) -> None:
+        """Benchmark scope match against list context value."""
+        self._scope_scalar.matches(self._context_list)

features/acms_context_policy_loader.feature

@@ -0,0 +1,171 @@
+Feature: ACMS Context Policy Configuration Loader
+  As a developer
+  I want to load and validate context policy configurations from YAML/TOML files
+  So that context policies can be flexibly configured per view
+
+  Background:
+    Given I have a context policy configuration loader
+
+  Scenario: Load valid YAML configuration
+    Given I have a YAML configuration file with:
+      """
+      view_name: test_view
+      default_priority_weight: 1.5
+      default_budget: 1000
+      policies:
+        - name: policy1
+          description: Test policy
+          priority_weight: 2.0
+          budget_override: 500
+          enabled: true
+          scopes:
+            - name: file_type
+              value: python
+      """
+    When I load the configuration from the YAML file
+    Then the configuration should have view_name "test_view"
+    And the configuration should have 1 policy
+    And the first policy should have name "policy1"
+    And the first policy should have priority_weight 2.0
+    And the first policy should have budget_override 500
+
+  Scenario: Load valid TOML configuration
+    Given I have a TOML configuration file with:
+      """
+      view_name = "test_view"
+      default_priority_weight = 1.5
+      default_budget = 1000
+
+      [[policies]]
+      name = "policy1"
+      description = "Test policy"
+      priority_weight = 2.0
+      budget_override = 500
+      enabled = true
+
+      [[policies.scopes]]
+      name = "file_type"
+      value = "python"
+      """
+    When I load the configuration from the TOML file
+    Then the configuration should have view_name "test_view"
+    And the configuration should have 1 policy
+
+  Scenario: Validate schema - missing required fields
+    Given I have a YAML configuration file with:
+      """
+      policies:
+        - description: Missing name field
+      """
+    When I try to load the configuration from the YAML file
+    Then I should get a validation error about missing name field
+
+  Scenario: Validate schema - invalid policy type
+    Given I have a YAML configuration file with:
+      """
+      policies:
+        - "invalid_policy_string"
+      """
+    When I try to load the configuration from the YAML file
+    Then I should get a validation error about policy type
+
+  Scenario: Validate schema - invalid numeric fields
+    Given I have a YAML configuration file with:
+      """
+      default_priority_weight: "not_a_number"
+      policies: []
+      """
+    When I try to load the configuration from the YAML file
+    Then I should get a validation error about numeric field
+
+  Scenario: Apply per-view policy with scope rules
+    Given I have a context policy configuration with:
+      | view_name | test_view |
+      | policies  | 1         |
+    And the policy has scope rules:
+      | name      | value  |
+      | file_type | python |
+    When I apply the policy to context with file_type "python"
+    Then the policy should match the context
+
+  Scenario: Apply per-view policy with priority weights
+    Given I have a context policy configuration with multiple policies
+    And policy1 has priority_weight 1.0
+    And policy2 has priority_weight 2.0
+    When I assemble context with both policies
+    Then policy2 should be applied before policy1
+
+  Scenario: Apply per-view policy with budget overrides
+    Given I have a context policy configuration with:
+      | view_name | test_view |
+      | policies  | 1         |
+    And the policy has budget_override 500
+    When I apply the policy to context
+    Then the assembled context should have budget 500
+
+  Scenario: Load configuration from non-existent file
+    Given I have a configuration file path that does not exist
+    When I try to load the configuration from the file
+    Then I should get a FileNotFoundError
+
+  Scenario: Load configuration with unsupported format
+    Given I have a configuration file with unsupported format ".json"
+    When I try to load the configuration from the file
+    Then I should get a ValueError about unsupported format
+
+  Scenario: Load configuration from string - YAML
+    Given I have a YAML configuration string:
+      """
+      view_name: string_view
+      policies:
+        - name: policy1
+      """
+    When I load the configuration from the YAML string
+    Then the configuration should have view_name "string_view"
+
+  Scenario: Load configuration from string - TOML
+    Given I have a TOML configuration string:
+      """
+      view_name = "string_view"
+      [[policies]]
+      name = "policy1"
+      """
+    When I load the configuration from the TOML string
+    Then the configuration should have view_name "string_view"
+
+  Scenario: Multiple scopes in a policy
+    Given I have a context policy configuration with:
+      | view_name | test_view |
+      | policies  | 1         |
+    And the policy has multiple scope rules:
+      | name      | value  |
+      | file_type | python |
+      | path      | src    |
+    When I apply the policy to context with multiple scopes: file_type "python" path "src"
+    Then the policy should match the context
+
+  Scenario: Policy with list values in scope
+    Given I have a context policy configuration with:
+      | view_name | test_view |
+      | policies  | 1         |
+    And the policy has scope with name "file_type" and values ["python", "javascript"]
+    When I apply the policy to context with file_type "python"
+    Then the policy should match the context
+
+  Scenario: Disabled policy should not be applied
+    Given I have a context policy configuration with:
+      | view_name | test_view |
+      | policies  | 1         |
+    And the policy is disabled
+    When I assemble context
+    Then the policy should not be applied
+
+  Scenario: Policy metadata is preserved
+    Given I have a context policy configuration with:
+      | view_name | test_view |
+      | policies  | 1         |
+    And the policy has metadata:
+      | key1 | value1 |
+      | key2 | value2 |
+    When I apply the policy to context
+    Then the assembled context should include the policy metadata

features/acms_plan_execution_integration.feature

@@ -0,0 +1,72 @@
+Feature: Plan Execution ACMS Integration
+  As a developer
+  I want the plan execution engine to use ACMS-assembled context for LLM calls
+  So that LLM calls receive properly assembled context instead of raw file dumps
+
+  Background:
+    Given I have a plan execution ACMS integration
+
+  Scenario: Prepare LLM context without policy configuration
+    Given I have no policy configuration
+    When I prepare LLM context with raw context data
+    Then the LLM context should be the same as the raw context
+
+  Scenario: Prepare LLM context with policy configuration
+    Given I have a policy configuration with 1 policy
+    When I prepare LLM context with raw context data
+    Then the LLM context should be assembled using ACMS policies
+
+  Scenario: Load policy configuration from file
+    Given I have a policy configuration file
+    When I load the policy configuration from the file
+    Then the integration should have the policy configuration loaded
+
+  Scenario: Load policy configuration from YAML string
+    Given I have a YAML policy configuration string
+    When I load the policy configuration from the YAML string
+    Then the integration should have the policy configuration loaded
+
+  Scenario: Load policy configuration from TOML string
+    Given I have a TOML policy configuration string
+    When I load the policy configuration from the TOML string
+    Then the integration should have the policy configuration loaded
+
+  Scenario: End-to-end: Plan execution with ACMS context
+    Given I have a plan execution ACMS integration
+    And I have a policy configuration with scope rules
+    And I have raw context data from file analysis
+    When I prepare LLM context for plan execution
+    Then the LLM context should include applied policies
+    And the LLM context should include assembled data
+    And the LLM context should have the correct view name
+
+  Scenario: ACMS context assembly respects priority weights
+    Given I have a policy configuration with multiple policies
+    And policy1 has priority_weight 1.0
+    And policy2 has priority_weight 2.0
+    When I prepare LLM context
+    Then policy2 should be applied before policy1 in the assembled context
+
+  Scenario: ACMS context assembly applies budget overrides
+    Given I have a policy configuration with 1 policy
+    And the policy has budget_override 500
+    When I prepare LLM context
+    Then the assembled context should have budget 500
+
+  Scenario: ACMS context assembly filters by scope
+    Given I have a policy configuration with scope rules
+    And the scope rule is file_type equals python
+    When I prepare LLM context with file_type "python"
+    Then the policy should be applied
+    When I prepare LLM context with file_type "javascript"
+    Then the policy should not be applied to the LLM context
+
+  Scenario: PlanExecutor accepts ACMS integration via dependency injection
+    Given I have a plan execution ACMS integration with a policy configuration
+    When I wire the ACMS integration into a PlanExecutor
+    Then the PlanExecutor should have the ACMS integration configured
+
+  Scenario: RuntimeExecuteActor uses ACMS integration for context assembly
+    Given I have a RuntimeExecuteActor with ACMS integration
+    When I execute decisions through the RuntimeExecuteActor
+    Then the ACMS integration should have been used for context assembly

features/steps/acms_context_policy_loader_steps.py

@@ -0,0 +1,406 @@
+"""Step definitions for ACMS context policy loader tests."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+from typing import Any
+
+from behave import given, then, when
+
+from cleveragents.acms.context_policy_loader import (
+    ContextPolicyConfig,
+    ContextPolicyConfigurationLoader,
+    PolicyScope,
+    ViewPolicyConfiguration,
+)
+from cleveragents.acms.plan_execution_integration import (
+    ACMSContextAssembler,
+    PlanExecutionACMSIntegration,
+)
+
+
+@given("I have a context policy configuration loader")
+def step_have_loader(context: Any) -> None:
+    """Initialize a context policy configuration loader."""
+    context.loader = ContextPolicyConfigurationLoader()
+    context.loaded_config = None
+    context.error = None
+
+
+@given("I have a YAML configuration file with:")
+def step_have_yaml_file(context: Any) -> None:
+    """Create a temporary YAML configuration file."""
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp:
+        tmp.write(context.text)
+        context.config_path = tmp.name
+
+
+@given("I have a TOML configuration file with:")
+def step_have_toml_file(context: Any) -> None:
+    """Create a temporary TOML configuration file."""
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".toml", delete=False) as tmp:
+        tmp.write(context.text)
+        context.config_path = tmp.name
+
+
+@when("I load the configuration from the YAML file")
+def step_load_yaml_file(context: Any) -> None:
+    """Load configuration from the YAML file."""
+    try:
+        context.loaded_config = context.loader.load(context.config_path)
+    except Exception as e:
+        context.error = e
+
+
+@when("I load the configuration from the TOML file")
+def step_load_toml_file(context: Any) -> None:
+    """Load configuration from the TOML file."""
+    try:
+        context.loaded_config = context.loader.load(context.config_path)
+    except Exception as e:
+        context.error = e
+
+
+@when("I try to load the configuration from the YAML file")
+def step_try_load_yaml_file(context: Any) -> None:
+    """Try to load configuration from the YAML file."""
+    try:
+        context.loaded_config = context.loader.load(context.config_path)
+    except Exception as e:
+        context.error = e
+
+
+@when("I try to load the configuration from the TOML file")
+def step_try_load_toml_file(context: Any) -> None:
+    """Try to load configuration from the TOML file."""
+    try:
+        context.loaded_config = context.loader.load(context.config_path)
+    except Exception as e:
+        context.error = e
+
+
+@then("the configuration should have view_name {view_name}")
+def step_check_view_name(context: Any, view_name: str) -> None:
+    """Check the view name in the configuration."""
+    assert context.loaded_config is not None
+    assert context.loaded_config.view_name == view_name
+
+
+@then(r"the configuration should have (\d+) policy")
+def step_check_policy_count(context: Any, count: str) -> None:
+    """Check the number of policies in the configuration."""
+    assert context.loaded_config is not None
+    assert len(context.loaded_config.policies) == int(count)
+
+
+@then("the first policy should have name {name}")
+def step_check_first_policy_name(context: Any, name: str) -> None:
+    """Check the name of the first policy."""
+    assert context.loaded_config is not None
+    assert len(context.loaded_config.policies) > 0
+    assert context.loaded_config.policies[0].name == name
+
+
+@then(r"the first policy should have priority_weight ([\d.]+)")
+def step_check_first_policy_priority(context: Any, weight_str: str) -> None:
+    """Check the priority weight of the first policy."""
+    assert context.loaded_config is not None
+    assert len(context.loaded_config.policies) > 0
+    assert context.loaded_config.policies[0].priority_weight == float(weight_str)
+
+
+@then(r"the first policy should have budget_override (\d+)")
+def step_check_first_policy_budget(context: Any, budget: str) -> None:
+    """Check the budget override of the first policy."""
+    assert context.loaded_config is not None
+    assert len(context.loaded_config.policies) > 0
+    assert context.loaded_config.policies[0].budget_override == int(budget)
+
+
+@then("I should get a validation error about missing name field")
+def step_check_missing_name_error(context: Any) -> None:
+    """Check for validation error about missing name field."""
+    assert context.error is not None
+    assert isinstance(context.error, ValueError)
+    assert "name" in str(context.error).lower()
+
+
+@then("I should get a validation error about policy type")
+def step_check_policy_type_error(context: Any) -> None:
+    """Check for validation error about policy type."""
+    assert context.error is not None
+    assert isinstance(context.error, ValueError)
+
+
+@then("I should get a validation error about numeric field")
+def step_check_numeric_field_error(context: Any) -> None:
+    """Check for validation error about numeric field."""
+    assert context.error is not None
+    assert isinstance(context.error, ValueError)
+
+
+@given("I have a configuration file path that does not exist")
+def step_have_nonexistent_file(context: Any) -> None:
+    """Set a non-existent file path."""
+    context.config_path = "/nonexistent/path/config.yaml"
+
+
+@when("I try to load the configuration from the file")
+def step_try_load_file(context: Any) -> None:
+    """Try to load configuration from the file."""
+    try:
+        context.loaded_config = context.loader.load(context.config_path)
+    except Exception as e:
+        context.error = e
+
+
+@then("I should get a FileNotFoundError")
+def step_check_file_not_found_error(context: Any) -> None:
+    """Check for FileNotFoundError."""
+    assert context.error is not None
+    assert isinstance(context.error, FileNotFoundError)
+
+
+@given("I have a configuration file with unsupported format {fmt}")
+def step_have_unsupported_format(context: Any, fmt: str) -> None:
+    """Create a file with unsupported format."""
+    with tempfile.NamedTemporaryFile(mode="w", suffix=fmt, delete=False) as tmp:
+        tmp.write("{}")
+        context.config_path = tmp.name
+
+
+@then("I should get a ValueError about unsupported format")
+def step_check_unsupported_format_error(context: Any) -> None:
+    """Check for ValueError about unsupported format."""
+    assert context.error is not None
+    assert isinstance(context.error, ValueError)
+    assert "unsupported" in str(context.error).lower()
+
+
+@given("I have a YAML configuration string:")
+def step_have_yaml_string(context: Any) -> None:
+    """Store a YAML configuration string."""
+    context.config_string = context.text
+    context.config_format = "yaml"
+
+
+@given("I have a TOML configuration string:")
+def step_have_toml_string(context: Any) -> None:
+    """Store a TOML configuration string."""
+    context.config_string = context.text
+    context.config_format = "toml"
+
+
+@when("I load the configuration from the YAML string")
+def step_load_yaml_string(context: Any) -> None:
+    """Load configuration from the YAML string."""
+    try:
+        context.loaded_config = context.loader.load_from_string(
+            context.config_string, "yaml"
+        )
+    except Exception as e:
+        context.error = e
+
+
+@when("I load the configuration from the TOML string")
+def step_load_toml_string(context: Any) -> None:
+    """Load configuration from the TOML string."""
+    try:
+        context.loaded_config = context.loader.load_from_string(
+            context.config_string, "toml"
+        )
+    except Exception as e:
+        context.error = e
+
+
+@given("I have a context policy configuration with:")
+def step_have_policy_config(context: Any) -> None:
+    """Create a context policy configuration from table."""
+    config_dict: dict[str, Any] = {}
+
+    for row in context.table:
+        key = row["view_name"] if "view_name" in row else row.get("key")
+        value = row.get("test_view") or row.get("value")
+
+        if key == "view_name":
+            config_dict["view_name"] = value
+        elif key == "policies":
+            config_dict["policies"] = [
+                {"name": f"policy{i}"} for i in range(int(value))
+            ]
+
+    context.policy_config = ViewPolicyConfiguration(
+        view_name=config_dict.get("view_name", "default"),
+        policies=[
+            ContextPolicyConfig(name=p["name"]) for p in config_dict.get("policies", [])
+        ],
+    )
+
+
+@given("the policy has scope rules:")
+def step_policy_has_scopes(context: Any) -> None:
+    """Add scope rules to the first policy."""
+    if not context.policy_config.policies:
+        context.policy_config.policies.append(ContextPolicyConfig(name="policy1"))
+
+    for row in context.table:
+        scope = PolicyScope(name=row["name"], value=row["value"])
+        context.policy_config.policies[0].scopes.append(scope)
+
+
+@when("I apply the policy to context with file_type {file_type}")
+def step_apply_policy_with_file_type(context: Any, file_type: str) -> None:
+    """Apply policy to context with specific file type."""
+    context.test_context = {"file_type": file_type}
+    context.assembler = ACMSContextAssembler(context.policy_config)
+
+
+@then("the policy should match the context")
+def step_policy_matches_context(context: Any) -> None:
+    """Check if the policy matches the context."""
+    policy = context.policy_config.policies[0]
+    matches = all(scope.matches(context.test_context) for scope in policy.scopes)
+    assert matches
+
+
+@given("I have a context policy configuration with multiple policies")
+def step_have_multiple_policies(context: Any) -> None:
+    """Create a configuration with multiple policies."""
+    context.policy_config = ViewPolicyConfiguration(
+        view_name="test_view",
+        policies=[
+            ContextPolicyConfig(name="policy1", priority_weight=1.0),
+            ContextPolicyConfig(name="policy2", priority_weight=2.0),
+        ],
+    )
+
+
+@given(r"policy1 has priority_weight ([\d.]+)")
+def step_policy1_priority(context: Any, weight_str: str) -> None:
+    """Set priority weight for policy1."""
+    if context.policy_config.policies:
+        context.policy_config.policies[0].priority_weight = weight
+    # Reinitialize integration if present (for plan execution integration tests)
+    if hasattr(context, "integration"):
+        context.integration = PlanExecutionACMSIntegration(
+            policy_config=context.policy_config
+        )
+
+
+@given(r"policy2 has priority_weight ([\d.]+)")
+def step_policy2_priority(context: Any, weight_str: str) -> None:
+    """Set priority weight for policy2."""
+    if len(context.policy_config.policies) > 1:
+        context.policy_config.policies[1].priority_weight = weight
+    # Reinitialize integration if present (for plan execution integration tests)
+    if hasattr(context, "integration"):
+        context.integration = PlanExecutionACMSIntegration(
+            policy_config=context.policy_config
+        )
+
+
+@when("I assemble context with both policies")
+def step_assemble_context_both(context: Any) -> None:
+    """Assemble context with both policies."""
+    context.assembler = ACMSContextAssembler(context.policy_config)
+    context.assembled = context.assembler.assemble_context({})
+    context.applied_policies = context.assembled.get("policies_applied", [])
+
+
+@then("policy2 should be applied before policy1")
+def step_policy2_before_policy1(context: Any) -> None:
+    """Check that policy2 is applied before policy1."""
+    if context.applied_policies:
+        assert context.applied_policies[0] == "policy2"
+
+
+@when("I apply the policy to context")
+def step_apply_policy(context: Any) -> None:
+    """Apply policy to context."""
+    context.assembler = ACMSContextAssembler(context.policy_config)
+    context.assembled = context.assembler.assemble_context({})
+
+
+@then(r"the assembled context should have budget (\d+)")
+def step_check_assembled_budget(context: Any, budget: str) -> None:
+    """Check the budget in the assembled context."""
+    # Support both direct assembler context and integration context
+    assembled = getattr(context, "assembled", None) or getattr(
+        context, "llm_context", None
+    )
+    assert assembled is not None
+    assert assembled["assembled_data"].get("budget") == int(budget)
+
+
+@when("I assemble context")
+def step_assemble_context(context: Any) -> None:
+    """Assemble context."""
+    context.assembler = ACMSContextAssembler(context.policy_config)
+    context.assembled = context.assembler.assemble_context({})
+
+
+@given("the policy is disabled")
+def step_policy_disabled(context: Any) -> None:
+    """Disable the first policy."""
+    if context.policy_config.policies:
+        context.policy_config.policies[0].enabled = False
+
+
+@then("the policy should not be applied")
+def step_policy_not_applied(context: Any) -> None:
+    """Check that the policy was not applied."""
+    # Support both direct assembler context and integration context
+    assembled = getattr(context, "assembled", None) or getattr(
+        context, "llm_context", None
+    )
+    assert assembled is not None
+    assert "policy1" not in assembled.get("policies_applied", [])
+
+
+@given("the policy has metadata:")
+def step_policy_metadata(context: Any) -> None:
+    """Add metadata to the policy."""
+    metadata = {}
+    for row in context.table:
+        metadata[row["key1"]] = row["value1"]
+    if context.policy_config.policies:
+        context.policy_config.policies[0].metadata = metadata
+
+
+@then("the assembled context should include the policy metadata")
+def step_check_policy_metadata(context: Any) -> None:
+    """Check that policy metadata is included in assembled context."""
+    assert context.assembled is not None
+    assembled_data = context.assembled.get("assembled_data", {})
+    assert "metadata" in assembled_data
+
+
+@given("the policy has multiple scope rules:")
+def step_policy_multiple_scopes(context: Any) -> None:
+    """Add multiple scope rules to the policy."""
+    if not context.policy_config.policies:
+        context.policy_config.policies.append(ContextPolicyConfig(name="policy1"))
+
+    for row in context.table:
+        scope = PolicyScope(name=row["name"], value=row["value"])
+        context.policy_config.policies[0].scopes.append(scope)
+
+
+@when(
+    "I apply the policy to context with multiple scopes: file_type {file_type} path {path}"
+)
+def step_apply_policy_multiple_scopes(context: Any, file_type: str, path: str) -> None:
+    """Apply policy to context with multiple scope values."""
+    context.test_context = {"file_type": file_type, "path": path}
+
+
+@given("the policy has scope with name {name} and values {values}")
+def step_policy_scope_list_values(context: Any, name: str, values: str) -> None:
+    """Add scope with list values to the policy."""
+    if not context.policy_config.policies:
+        context.policy_config.policies.append(ContextPolicyConfig(name="policy1"))
+
+    value_list = json.loads(values)
+    scope = PolicyScope(name=name, value=value_list)
+    context.policy_config.policies[0].scopes.append(scope)

features/steps/acms_plan_execution_integration_steps.py

@@ -0,0 +1,381 @@
+"""Step definitions for ACMS plan execution integration tests."""
+
+from __future__ import annotations
+
+import tempfile
+from typing import Any
+from unittest.mock import MagicMock
+
+import yaml
+from behave import given, then, when
+from ulid import ULID
+
+from cleveragents.acms.context_policy_loader import (
+    ContextPolicyConfig,
+    PolicyScope,
+    ViewPolicyConfiguration,
+)
+from cleveragents.acms.plan_execution_integration import (
+    PlanExecutionACMSIntegration,
+)
+from cleveragents.application.services.plan_execution_context import (
+    PlanExecutionContext,
+    RuntimeExecuteActor,
+)
+from cleveragents.application.services.plan_executor import (
+    PlanExecutor,
+    StrategyDecision,
+)
+from cleveragents.tool.runner import ToolRunner
+
+
+@when(r"I prepare LLM context$")
+def step_prepare_llm_context_generic(context: Any) -> None:
+    """Prepare LLM context for a plain text step with no qualifiers."""
+    if hasattr(context, "integration") and context.integration is not None:
+        raw = getattr(context, "raw_context", {"file_type": "python"})
+        context.llm_context = context.integration.prepare_llm_context(raw)
+
+
+@given("I have a plan execution ACMS integration")
+def step_have_integration(context: Any) -> None:
+    """Initialize a plan execution ACMS integration."""
+    context.integration = PlanExecutionACMSIntegration()
+    context.llm_context = None
+    context.error = None
+
+
+@given("I have no policy configuration")
+def step_no_policy_config(context: Any) -> None:
+    """Ensure no policy configuration is set."""
+    context.integration = PlanExecutionACMSIntegration(policy_config=None)
+
+
+@given(r"I have a policy configuration with (\d+) policy")
+def step_have_policy_config(context: Any, count: str) -> None:
+    """Create a policy configuration with specified number of policies."""
+    policies = [ContextPolicyConfig(name=f"policy{i + 1}") for i in range(int(count))]
+    context.policy_config = ViewPolicyConfiguration(
+        view_name="test_view",
+        policies=policies,
+    )
+    context.integration = PlanExecutionACMSIntegration(
+        policy_config=context.policy_config
+    )
+
+
+@given("I have raw context data from file analysis")
+def step_have_raw_context(context: Any) -> None:
+    """Create raw context data."""
+    context.raw_context = {
+        "file_type": "python",
+        "path": "src/module.py",
+        "content": "def hello(): pass",
+    }
+
+
+@when("I prepare LLM context with raw context data")
+def step_prepare_llm_context(context: Any) -> None:
+    """Prepare LLM context from raw context."""
+    context.raw_context = {"file_type": "python", "path": "src/module.py"}
+    context.llm_context = context.integration.prepare_llm_context(context.raw_context)
+
+
+@then("the LLM context should be the same as the raw context")
+def step_check_llm_context_same(context: Any) -> None:
+    """Check that LLM context is the same as raw context."""
+    assert context.llm_context == context.raw_context
+
+
+@then("the LLM context should be assembled using ACMS policies")
+def step_check_llm_context_assembled(context: Any) -> None:
+    """Check that LLM context is assembled."""
+    assert context.llm_context is not None
+    assert "view" in context.llm_context
+    assert "policies_applied" in context.llm_context
+    assert "assembled_data" in context.llm_context
+
+
+@given("I have a policy configuration file")
+def step_have_policy_file(context: Any) -> None:
+    """Create a temporary policy configuration file."""
+    config_data = {
+        "view_name": "test_view",
+        "policies": [{"name": "policy1"}],
+    }
+    with tempfile.NamedTemporaryFile(mode="w", suffix=".yaml", delete=False) as tmp:
+        yaml.dump(config_data, tmp)
+        context.config_path = tmp.name
+
+
+@when("I load the policy configuration from the file")
+def step_load_policy_file(context: Any) -> None:
+    """Load policy configuration from file."""
+    try:
+        context.integration.load_policy_config(context.config_path)
+    except Exception as e:
+        context.error = e
+
+
+@then("the integration should have the policy configuration loaded")
+def step_check_policy_loaded(context: Any) -> None:
+    """Check that policy configuration is loaded."""
+    assert context.integration.policy_config is not None
+    assert context.integration.assembler is not None
+
+
+@given("I have a YAML policy configuration string")
+def step_have_yaml_policy_string(context: Any) -> None:
+    """Create a YAML policy configuration string."""
+    context.policy_string = """
+view_name: test_view
+policies:
+  - name: policy1
+"""
+
+
+@when("I load the policy configuration from the YAML string")
+def step_load_yaml_policy_string(context: Any) -> None:
+    """Load policy configuration from YAML string."""
+    try:
+        context.integration.load_policy_config_from_string(
+            context.policy_string, "yaml"
+        )
+    except Exception as e:
+        context.error = e
+
+
+@given("I have a TOML policy configuration string")
+def step_have_toml_policy_string(context: Any) -> None:
+    """Create a TOML policy configuration string."""
+    context.policy_string = """
+view_name = "test_view"
+[[policies]]
+name = "policy1"
+"""
+
+
+@when("I load the policy configuration from the TOML string")
+def step_load_toml_policy_string(context: Any) -> None:
+    """Load policy configuration from TOML string."""
+    try:
+        context.integration.load_policy_config_from_string(
+            context.policy_string, "toml"
+        )
+    except Exception as e:
+        context.error = e
+
+
+@when("I prepare LLM context for plan execution")
+def step_prepare_llm_context_plan(context: Any) -> None:
+    """Prepare LLM context for plan execution."""
+    context.llm_context = context.integration.prepare_llm_context(context.raw_context)
+
+
+@then("the LLM context should include applied policies")
+def step_check_applied_policies(context: Any) -> None:
+    """Check that LLM context includes applied policies."""
+    assert context.llm_context is not None
+    assert "policies_applied" in context.llm_context
+
+
+@then("the LLM context should include assembled data")
+def step_check_assembled_data(context: Any) -> None:
+    """Check that LLM context includes assembled data."""
+    assert context.llm_context is not None
+    assert "assembled_data" in context.llm_context
+
+
+@then("the LLM context should have the correct view name")
+def step_check_view_name(context: Any) -> None:
+    """Check that LLM context has the correct view name."""
+    assert context.llm_context is not None
+    assert context.llm_context.get("view") == "test_view"
+
+
+@given("I have a policy configuration with multiple policies")
+def step_have_multiple_policies(context: Any) -> None:
+    """Create a policy configuration with multiple policies."""
+    policies = [
+        ContextPolicyConfig(name="policy1", priority_weight=1.0),
+        ContextPolicyConfig(name="policy2", priority_weight=2.0),
+    ]
+    context.policy_config = ViewPolicyConfiguration(
+        view_name="test_view",
+        policies=policies,
+    )
+    context.integration = PlanExecutionACMSIntegration(
+        policy_config=context.policy_config
+    )
+
+
+@then("policy2 should be applied before policy1 in the assembled context")
+def step_check_policy_order(context: Any) -> None:
+    """Check that policy2 is applied before policy1."""
+    assert context.llm_context is not None
+    policies_applied = context.llm_context.get("policies_applied", [])
+    if len(policies_applied) >= 2:
+        assert policies_applied[0] == "policy2"
+
+
+@given("I have a policy configuration with scope rules")
+def step_have_scope_rules(context: Any) -> None:
+    """Create a policy configuration with scope rules."""
+    policy = ContextPolicyConfig(
+        name="policy1",
+        scopes=[PolicyScope(name="file_type", value="python")],
+    )
+    context.policy_config = ViewPolicyConfiguration(
+        view_name="test_view",
+        policies=[policy],
+    )
+    context.integration = PlanExecutionACMSIntegration(
+        policy_config=context.policy_config
+    )
+
+
+@given(r"policy(\d+) has priority_weight ([\d.]+)")
+def step_set_policy_priority(context: Any, count: str, weight: str) -> None:
+    """Set the priority weight for a specific policy by index."""
+    idx = int(count) - 1
+    if (
+        context.policy_config
+        and context.policy_config.policies
+        and 0 <= idx < len(context.policy_config.policies)
+    ):
+        context.policy_config.policies[idx].priority_weight = float(weight)
+        # Rebuild integration with updated config
+        context.integration = PlanExecutionACMSIntegration(
+            policy_config=context.policy_config,
+        )
+
+
+@given(r"the policy has budget_override (\d+)")
+def step_set_policy_budget(context: Any, amount: str) -> None:
+    """Set the budget override for the first policy in the configuration."""
+    if context.policy_config and context.policy_config.policies:
+        context.policy_config.policies[0].budget_override = int(amount)
+        # Rebuild integration with updated config
+        context.integration = PlanExecutionACMSIntegration(
+            policy_config=context.policy_config,
+        )
+
+
+@given("the scope rule is file_type equals {value}")
+def step_scope_rule(context: Any, value: str) -> None:
+    """Set the scope rule."""
+    if context.policy_config.policies:
+        context.policy_config.policies[0].scopes = [
+            PolicyScope(name="file_type", value=value)
+        ]
+    context.integration = PlanExecutionACMSIntegration(
+        policy_config=context.policy_config
+    )
+
+
+@when("I prepare LLM context with file_type {file_type}")
+def step_prepare_context_file_type(context: Any, file_type: str) -> None:
+    """Prepare LLM context with specific file type."""
+    context.raw_context = {"file_type": file_type}
+    context.llm_context = context.integration.prepare_llm_context(context.raw_context)
+
+
+@then("the policy should be applied")
+def step_policy_applied(context: Any) -> None:
+    """Check that the policy was applied."""
+    assert context.llm_context is not None
+    policies_applied = context.llm_context.get("policies_applied", [])
+    assert "policy1" in policies_applied
+
+
+@then("the policy should not be applied to the LLM context")
+def step_policy_not_applied(context: Any) -> None:
+    """Check that the policy was NOT applied."""
+    assert context.llm_context is not None
+    policies_applied = context.llm_context.get("policies_applied", [])
+    assert "policy1" not in policies_applied
+
+    # Also verify budget is absent when no matching policy applies
+    assembled_data = context.llm_context.get("assembled_data", {})
+    assert "budget" not in assembled_data, (
+        f"Expected no 'budget' key when scope doesn't match, got keys: {list(assembled_data.keys())}"
+    )
+
+
+@given("I have a plan execution ACMS integration with a policy configuration")
+def step_have_integration_with_policy(context: Any) -> None:
+    """Create a plan execution ACMS integration with a policy configuration."""
+    policy = ContextPolicyConfig(name="policy1", priority_weight=1.0)
+    context.policy_config = ViewPolicyConfiguration(
+        view_name="test_view",
+        policies=[policy],
+    )
+    context.integration = PlanExecutionACMSIntegration(
+        policy_config=context.policy_config
+    )
+
+
+@when("I wire the ACMS integration into a PlanExecutor")
+def step_wire_acms_into_plan_executor(context: Any) -> None:
+    """Wire the ACMS integration into a PlanExecutor via dependency injection."""
+    mock_lifecycle = MagicMock()
+    context.executor = PlanExecutor(
+        lifecycle_service=mock_lifecycle,
+        acms_integration=context.integration,
+    )
+
+
+@then("the PlanExecutor should have the ACMS integration configured")
+def step_check_plan_executor_has_acms(context: Any) -> None:
+    """Check that the PlanExecutor has the ACMS integration configured."""
+    assert context.executor is not None
+    assert context.executor.acms_integration is context.integration
+
+
+@given("I have a RuntimeExecuteActor with ACMS integration")
+def step_have_runtime_actor_with_acms(context: Any) -> None:
+    """Create a RuntimeExecuteActor with ACMS integration."""
+    policy = ContextPolicyConfig(name="policy1", priority_weight=1.0)
+    policy_config = ViewPolicyConfiguration(
+        view_name="test_view",
+        policies=[policy],
+    )
+    context.acms_integration = PlanExecutionACMSIntegration(policy_config=policy_config)
+
+    plan_id = str(ULID())
+    context.execution_context = PlanExecutionContext(plan_id=plan_id)
+
+    mock_tool_runner = MagicMock(spec=ToolRunner)
+    mock_tool_runner.discover.return_value = []
+
+    context.runtime_actor = RuntimeExecuteActor(
+        tool_runner=mock_tool_runner,
+        execution_context=context.execution_context,
+        acms_integration=context.acms_integration,
+    )
+    context.acms_calls: list[dict[str, Any]] = []
+
+
+@when("I execute decisions through the RuntimeExecuteActor")
+def step_execute_decisions_through_runtime_actor(context: Any) -> None:
+    """Execute decisions through the RuntimeExecuteActor."""
+    root_id = str(ULID())
+    decisions = [
+        StrategyDecision(
+            decision_id=root_id,
+            step_text="Step one",
+            sequence=0,
+            parent_id=None,
+        ),
+    ]
+    context.runtime_result = context.runtime_actor.execute(decisions=decisions)
+
+
+@then("the ACMS integration should have been used for context assembly")
+def step_check_acms_used_for_context(context: Any) -> None:
+    """Check that the ACMS integration was used for context assembly."""
+    assert context.runtime_result is not None
+    assert context.runtime_actor.acms_integration is context.acms_integration
+    # Verify the result has the expected structure
+    assert context.runtime_result.changeset_id is not None
+    assert context.runtime_result.tool_call_count >= 0

src/cleveragents/acms/__init__.py

@@ -6,8 +6,9 @@ inheritance mechanism for resolving named detail levels across the
 ontology hierarchy (Layer 3 -> Layer 2 -> Layer 1 -> Layer 0).
 
 Also provides the ACMS index data model and file traversal engine for
-indexing large projects, and the hot storage tier LRU cache
-implementation.
+indexing large projects, the hot storage tier LRU cache implementation,
+and context policy configuration loading and plan execution integration
+for flexible context policy management.
 
 Based on ``docs/specification.md`` ~lines 42333-42422, 44405-44420.
 """
@@ -15,6 +16,12 @@ Based on ``docs/specification.md`` ~lines 42333-42422, 44405-44420.
 from __future__ import annotations
 
 from cleveragents.acms import uko as _uko
+from cleveragents.acms.context_policy_loader import (
+    ContextPolicyConfig,
+    ContextPolicyConfigurationLoader,
+    PolicyScope,
+    ViewPolicyConfiguration,
+)
 from cleveragents.acms.index import (
     ACMSIndex,
     FileTraversalEngine,
@@ -22,6 +29,10 @@ from cleveragents.acms.index import (
     IndexEntry,
     TierLevel,
 )
+from cleveragents.acms.plan_execution_integration import (
+    ACMSContextAssembler,
+    PlanExecutionACMSIntegration,
+)
 from cleveragents.acms.storage.hot import HotStorageTier
 from cleveragents.acms.uko import (
     CODE_DETAIL_LEVEL_MAP,
@@ -74,7 +85,7 @@ from cleveragents.acms.uko import (
     resolve_detail_level,
 )
 
-# Combine exports from uko, index, and storage modules
+# Combine exports from uko, index, storage, and context-policy modules
 _uko_exports = list(_uko.__all__)
 _index_exports = [
     "ACMSIndex",
@@ -84,5 +95,15 @@ _index_exports = [
     "TierLevel",
 ]
 _storage_exports = ["HotStorageTier"]
+_context_policy_exports = [
+    "ContextPolicyConfig",
+    "ContextPolicyConfigurationLoader",
+    "PolicyScope",
+    "ViewPolicyConfiguration",
+    "ACMSContextAssembler",
+    "PlanExecutionACMSIntegration",
+]
 
-__all__: list[str] = _uko_exports + _index_exports + _storage_exports
+__all__: list[str] = (
+    _uko_exports + _index_exports + _storage_exports + _context_policy_exports
+)

src/cleveragents/acms/context_policy_loader.py

@@ -0,0 +1,369 @@
+"""Context policy configuration loader for ACMS.
+
+This module provides functionality to load and validate context policy
+configurations from YAML/TOML files, supporting per-view policy application
+with scope rules, priority weights, and budget overrides.
+"""
+
+from __future__ import annotations
+
+import tomllib
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Any, ClassVar
+
+import yaml
+
+
+@dataclass
+class PolicyScope:
+    """Represents a scope rule for context policy application."""
+
+    name: str
+    """Name of the scope (e.g., 'file_type', 'path_pattern')."""
+
+    value: str | list[str]
+    """Value or list of values for the scope."""
+
+    def matches(self, context: dict[str, Any]) -> bool:
+        """Check if this scope matches the given context.
+
+        Args:
+            context: Context dictionary to match against.
+
+        Returns:
+            True if the scope matches, False otherwise.
+        """
+        if self.name not in context:
+            return False
+
+        context_value = context[self.name]
+        scope_values = self.value if isinstance(self.value, list) else [self.value]
+
+        if isinstance(context_value, list):
+            return any(cv in scope_values for cv in context_value)
+        return context_value in scope_values
+
+
+@dataclass
+class ContextPolicyConfig:
+    """Configuration for a context policy."""
+
+    name: str
+    """Name of the policy."""
+
+    description: str | None = None
+    """Description of the policy."""
+
+    scopes: list[PolicyScope] = field(default_factory=list)
+    """List of scope rules for this policy."""
+
+    priority_weight: float = 1.0
+    """Priority weight for this policy (higher = more important)."""
+
+    budget_override: int | None = None
+    """Optional budget override in tokens."""
+
+    enabled: bool = True
+    """Whether this policy is enabled."""
+
+    metadata: dict[str, Any] = field(default_factory=dict)
+    """Additional metadata for the policy."""
+
+
+@dataclass
+class ViewPolicyConfiguration:
+    """Configuration for policies applied to a specific view."""
+
+    view_name: str
+    """Name of the view."""
+
+    policies: list[ContextPolicyConfig] = field(default_factory=list)
+    """List of policies for this view."""
+
+    default_priority_weight: float = 1.0
+    """Default priority weight for policies in this view."""
+
+    default_budget: int | None = None
+    """Default budget for this view."""
+
+    metadata: dict[str, Any] = field(default_factory=dict)
+    """Additional metadata for the view."""
+
+
+class ContextPolicyConfigurationLoader:
+    """Loader for context policy configurations from YAML/TOML files."""
+
+    SUPPORTED_FORMATS: ClassVar[set[str]] = {"yaml", "yml", "toml"}
+
+    def __init__(self) -> None:
+        """Initialize the configuration loader."""
+        pass
+
+    def load(self, config_path: str | Path) -> ViewPolicyConfiguration:
+        """Load context policy configuration from a file.
+
+        Args:
+            config_path: Path to the configuration file (YAML or TOML).
+
+        Returns:
+            ViewPolicyConfiguration instance.
+
+        Raises:
+            ValueError: If the file format is not supported or config is invalid.
+            FileNotFoundError: If the configuration file does not exist.
+        """
+        config_path = Path(config_path)
+
+        if not config_path.exists():
+            raise FileNotFoundError(f"Configuration file not found: {config_path}")
+
+        file_format = config_path.suffix.lstrip(".").lower()
+
+        if file_format not in self.SUPPORTED_FORMATS:
+            raise ValueError(
+                f"Unsupported file format: {file_format}. "
+                f"Supported formats: {', '.join(self.SUPPORTED_FORMATS)}"
+            )
+
+        if file_format in ("yaml", "yml"):
+            return self._load_yaml(config_path)
+        elif file_format == "toml":
+            return self._load_toml(config_path)
+
+        raise ValueError(f"Unsupported file format: {file_format}")
+
+    def _load_yaml(self, config_path: Path) -> ViewPolicyConfiguration:
+        """Load configuration from a YAML file.
+
+        Args:
+            config_path: Path to the YAML configuration file.
+
+        Returns:
+            ViewPolicyConfiguration instance.
+
+        Raises:
+            ValueError: If the configuration is invalid.
+        """
+        with config_path.open("r") as f:
+            data = yaml.safe_load(f)
+
+        if not isinstance(data, dict):
+            raise ValueError("Configuration must be a dictionary")
+
+        return self._parse_configuration(data)
+
+    def _load_toml(self, config_path: Path) -> ViewPolicyConfiguration:
+        """Load configuration from a TOML file.
+
+        Args:
+            config_path: Path to the TOML configuration file.
+
+        Returns:
+            ViewPolicyConfiguration instance.
+
+        Raises:
+            ValueError: If the configuration is invalid.
+        """
+        with config_path.open("rb") as f:
+            data = tomllib.load(f)
+
+        return self._parse_configuration(data)
+
+    def _parse_configuration(self, data: dict[str, Any]) -> ViewPolicyConfiguration:
+        """Parse configuration data into ViewPolicyConfiguration.
+
+        Args:
+            data: Configuration data dictionary.
+
+        Returns:
+            ViewPolicyConfiguration instance.
+
+        Raises:
+            ValueError: If the configuration is invalid.
+        """
+        self._validate_schema(data)
+
+        view_name = data.get("view_name", "default")
+        default_priority_weight = data.get("default_priority_weight", 1.0)
+        default_budget = data.get("default_budget")
+        metadata = data.get("metadata", {})
+
+        policies = []
+        for policy_data in data.get("policies", []):
+            policy = self._parse_policy(policy_data, default_priority_weight)
+            policies.append(policy)
+
+        return ViewPolicyConfiguration(
+            view_name=view_name,
+            policies=policies,
+            default_priority_weight=default_priority_weight,
+            default_budget=default_budget,
+            metadata=metadata,
+        )
+
+    def _parse_policy(
+        self, policy_data: dict[str, Any], default_priority_weight: float
+    ) -> ContextPolicyConfig:
+        """Parse a single policy configuration.
+
+        Args:
+            policy_data: Policy configuration data.
+            default_priority_weight: Default priority weight to use.
+
+        Returns:
+            ContextPolicyConfig instance.
+
+        Raises:
+            ValueError: If the policy configuration is invalid.
+        """
+        if not isinstance(policy_data, dict):
+            raise ValueError("Policy must be a dictionary")
+
+        if "name" not in policy_data:
+            raise ValueError("Policy must have a 'name' field")
+
+        name = policy_data["name"]
+        description = policy_data.get("description")
+        priority_weight = policy_data.get("priority_weight", default_priority_weight)
+        budget_override = policy_data.get("budget_override")
+        enabled = policy_data.get("enabled", True)
+        metadata = policy_data.get("metadata", {})
+
+        scopes = []
+        for scope_data in policy_data.get("scopes", []):
+            scope = self._parse_scope(scope_data)
+            scopes.append(scope)
+
+        return ContextPolicyConfig(
+            name=name,
+            description=description,
+            scopes=scopes,
+            priority_weight=priority_weight,
+            budget_override=budget_override,
+            enabled=enabled,
+            metadata=metadata,
+        )
+
+    def _parse_scope(self, scope_data: dict[str, Any]) -> PolicyScope:
+        """Parse a single scope configuration.
+
+        Args:
+            scope_data: Scope configuration data.
+
+        Returns:
+            PolicyScope instance.
+
+        Raises:
+            ValueError: If the scope configuration is invalid.
+        """
+        if not isinstance(scope_data, dict):
+            raise ValueError("Scope must be a dictionary")
+
+        if "name" not in scope_data:
+            raise ValueError("Scope must have a 'name' field")
+
+        if "value" not in scope_data:
+            raise ValueError("Scope must have a 'value' field")
+
+        return PolicyScope(
+            name=scope_data["name"],
+            value=scope_data["value"],
+        )
+
+    def _validate_schema(self, data: dict[str, Any]) -> None:
+        """Validate the configuration schema.
+
+        Args:
+            data: Configuration data to validate.
+
+        Raises:
+            ValueError: If the schema is invalid.
+        """
+        if not isinstance(data, dict):
+            raise ValueError("Configuration must be a dictionary")
+
+        # Validate top-level fields
+        allowed_fields = {
+            "view_name",
+            "policies",
+            "default_priority_weight",
+            "default_budget",
+            "metadata",
+        }
+        for field_name in data:
+            if field_name not in allowed_fields:
+                raise ValueError(f"Unknown field: {field_name}")
+
+        # Validate policies
+        policies = data.get("policies", [])
+        if not isinstance(policies, list):
+            raise ValueError("'policies' must be a list")
+
+        for i, policy in enumerate(policies):
+            if not isinstance(policy, dict):
+                raise ValueError(f"Policy {i} must be a dictionary")
+
+            if "name" not in policy:
+                raise ValueError(f"Policy {i} must have a 'name' field")
+
+            # Validate scopes
+            scopes = policy.get("scopes", [])
+            if not isinstance(scopes, list):
+                raise ValueError(f"Policy {i} 'scopes' must be a list")
+
+            for j, scope in enumerate(scopes):
+                if not isinstance(scope, dict):
+                    raise ValueError(f"Policy {i} scope {j} must be a dictionary")
+
+                if "name" not in scope:
+                    raise ValueError(f"Policy {i} scope {j} must have a 'name' field")
+
+                if "value" not in scope:
+                    raise ValueError(f"Policy {i} scope {j} must have a 'value' field")
+
+        # Validate numeric fields
+        if "default_priority_weight" in data and not isinstance(
+            data["default_priority_weight"], (int, float)
+        ):
+            raise ValueError("'default_priority_weight' must be a number")
+
+        if (
+            "default_budget" in data
+            and data["default_budget"] is not None
+            and not isinstance(data["default_budget"], int)
+        ):
+            raise ValueError("'default_budget' must be an integer or null")
+
+    def load_from_string(
+        self, config_string: str, fmt: str = "yaml"
+    ) -> ViewPolicyConfiguration:
+        """Load configuration from a string.
+
+        Args:
+            config_string: Configuration string.
+            fmt: Format of the string ('yaml' or 'toml').
+
+        Returns:
+            ViewPolicyConfiguration instance.
+
+        Raises:
+            ValueError: If the format is not supported or config is invalid.
+        """
+        if fmt not in self.SUPPORTED_FORMATS:
+            raise ValueError(
+                f"Unsupported format: {fmt}. "
+                f"Supported formats: {', '.join(self.SUPPORTED_FORMATS)}"
+            )
+
+        if fmt in ("yaml", "yml"):
+            data = yaml.safe_load(config_string)
+        elif fmt == "toml":
+            data = tomllib.loads(config_string)
+        else:
+            raise ValueError(f"Unsupported format: {fmt}")
+
+        if not isinstance(data, dict):
+            raise ValueError("Configuration must be a dictionary")
+
+        return self._parse_configuration(data)

src/cleveragents/acms/plan_execution_integration.py

@@ -0,0 +1,189 @@
+"""Plan execution integration with ACMS context assembly.
+
+This module integrates the ACMS context assembly pipeline with the plan
+execution engine, ensuring that LLM calls use ACMS-assembled context instead
+of raw file dumps.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from cleveragents.acms.context_policy_loader import (
+    ContextPolicyConfig,
+    ContextPolicyConfigurationLoader,
+    PolicyScope,
+    ViewPolicyConfiguration,
+)
+
+
+class ACMSContextAssembler:
+    """Assembles context using ACMS policies for plan execution."""
+
+    def __init__(self, policy_config: ViewPolicyConfiguration) -> None:
+        """Initialize the ACMS context assembler.
+
+        Args:
+            policy_config: View policy configuration.
+
+        Raises:
+            ValueError: If policy_config is None.
+        """
+        if policy_config is None:
+            raise ValueError("policy_config must not be None")
+        self.policy_config = policy_config
+        self.enabled_policies = [p for p in policy_config.policies if p.enabled]
+
+    def assemble_context(self, raw_context: dict[str, Any]) -> dict[str, Any]:
+        """Assemble context using ACMS policies.
+
+        Args:
+            raw_context: Raw context data (e.g., file dumps).
+
+        Returns:
+            ACMS-assembled context dictionary.
+        """
+        assembled_context: dict[str, Any] = {
+            "view": self.policy_config.view_name,
+            "policies_applied": [],
+            "assembled_data": {},
+        }
+
+        # Sort policies by priority weight (descending)
+        sorted_policies = sorted(
+            self.enabled_policies,
+            key=lambda p: p.priority_weight,
+            reverse=True,
+        )
+
+        for policy in sorted_policies:
+            # Check if policy scopes match the context
+            if self._scopes_match(policy.scopes, raw_context):
+                assembled_context["policies_applied"].append(policy.name)
+                # Apply policy transformations
+                policy_context = self._apply_policy(policy, raw_context)
+                assembled_context["assembled_data"].update(policy_context)
+
+        return assembled_context
+
+    def _scopes_match(self, scopes: list[PolicyScope], context: dict[str, Any]) -> bool:
+        """Check if all scopes match the given context.
+
+        Args:
+            scopes: List of policy scopes.
+            context: Context to match against.
+
+        Returns:
+            True if all scopes match, False otherwise.
+        """
+        if not scopes:
+            return True
+
+        return all(scope.matches(context) for scope in scopes)
+
+    def _apply_policy(
+        self, policy: ContextPolicyConfig, raw_context: dict[str, Any]
+    ) -> dict[str, Any]:
+        """Apply a policy to the raw context.
+
+        Args:
+            policy: Policy configuration.
+            raw_context: Raw context data.
+
+        Returns:
+            Transformed context data.
+        """
+        policy_context: dict[str, Any] = {}
+
+        # Apply budget override if specified
+        if policy.budget_override is not None:
+            policy_context["budget"] = policy.budget_override
+
+        # Apply priority weight
+        policy_context["priority"] = policy.priority_weight
+
+        # Include policy metadata
+        if policy.metadata:
+            policy_context["metadata"] = policy.metadata
+
+        # Include relevant raw context data
+        for key, value in raw_context.items():
+            if key not in policy_context:
+                policy_context[key] = value
+
+        return policy_context
+
+
+class PlanExecutionACMSIntegration:
+    """Integrates ACMS context assembly into plan execution.
+
+    This class wires the ACMS context assembly pipeline into the plan
+    execution engine. When a policy configuration is loaded, the
+    ``prepare_llm_context`` method assembles context using ACMS policies
+    instead of passing raw file dumps to LLM calls.
+
+    Usage with the plan execution engine::
+
+        integration = PlanExecutionACMSIntegration()
+        integration.load_policy_config("path/to/policy.yaml")
+
+        # In RuntimeExecuteActor.execute(), replace raw context with:
+        llm_context = integration.prepare_llm_context(raw_context)
+    """
+
+    def __init__(
+        self,
+        policy_config: ViewPolicyConfiguration | None = None,
+    ) -> None:
+        """Initialize the plan execution ACMS integration.
+
+        Args:
+            policy_config: Optional view policy configuration.
+        """
+        self.policy_config = policy_config
+        self.assembler: ACMSContextAssembler | None = None
+
+        if policy_config:
+            self.assembler = ACMSContextAssembler(policy_config)
+
+    def prepare_llm_context(self, raw_context: dict[str, Any]) -> dict[str, Any]:
+        """Prepare context for LLM calls using ACMS assembly.
+
+        This method is the integration point between the plan execution
+        engine and the ACMS context assembly pipeline. Call this method
+        in place of passing raw file dumps to LLM calls.
+
+        Args:
+            raw_context: Raw context data (e.g., file dumps).
+
+        Returns:
+            ACMS-assembled context ready for LLM calls.
+        """
+        if self.assembler is None:
+            # If no policy config, return raw context as-is
+            return raw_context
+
+        return self.assembler.assemble_context(raw_context)
+
+    def load_policy_config(self, config_path: str) -> None:
+        """Load policy configuration from a file.
+
+        Args:
+            config_path: Path to the policy configuration file.
+        """
+        loader = ContextPolicyConfigurationLoader()
+        self.policy_config = loader.load(config_path)
+        self.assembler = ACMSContextAssembler(self.policy_config)
+
+    def load_policy_config_from_string(
+        self, config_string: str, fmt: str = "yaml"
+    ) -> None:
+        """Load policy configuration from a string.
+
+        Args:
+            config_string: Configuration string.
+            fmt: Format of the string ('yaml' or 'toml').
+        """
+        loader = ContextPolicyConfigurationLoader()
+        self.policy_config = loader.load_from_string(config_string, fmt)
+        self.assembler = ACMSContextAssembler(self.policy_config)

src/cleveragents/application/services/plan_execution_context.py

@@ -10,14 +10,16 @@ for capturing tool-call mutations during plan execution.
   phase and delegates changeset operations to a ``ChangeSetStore``.
 - ``RuntimeExecuteResult`` -- structured output from the runtime actor.
 - ``RuntimeExecuteActor`` -- wraps ``ToolRunner`` to execute strategy
-  decisions with full changeset capture.
+  decisions with full changeset capture.  When an ACMS integration is
+  provided, context is assembled via policy-driven decisions before LLM
+  calls instead of using raw file dumps.
 """
 
 from __future__ import annotations
 
 import time
 from collections.abc import Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 import structlog
 from pydantic import BaseModel, ConfigDict, Field
@@ -35,6 +37,11 @@ from cleveragents.domain.models.core.change import (
 from cleveragents.tool.context import BoundResource
 from cleveragents.tool.runner import ToolRunner
 
+if TYPE_CHECKING:
+    from cleveragents.acms.plan_execution_integration import (
+        PlanExecutionACMSIntegration,
+    )
+
 logger = structlog.get_logger(__name__)
 
 # Type alias for streaming callbacks
@@ -288,6 +295,11 @@ class RuntimeExecuteActor:
     Wraps the tool-calling runtime to execute strategy decisions with
     full changeset capture via ``PlanExecutionContext``.
 
+    When an ``acms_integration`` is provided, each decision's raw context
+    is assembled via ACMS policy-driven decisions before being passed to
+    LLM calls.  This ensures LLM calls receive properly scoped, budget-
+    constrained context views instead of raw file dumps.
+
     Parameters
     ----------
     tool_runner:
@@ -295,6 +307,9 @@ class RuntimeExecuteActor:
     execution_context:
         The ``PlanExecutionContext`` carrying plan metadata and
         changeset store.
+    acms_integration:
+        Optional ACMS integration for assembling context using
+        policy-driven decisions.  When ``None``, raw context is used.
     """
 
     def __init__(
@@ -302,6 +317,7 @@ class RuntimeExecuteActor:
         *,
         tool_runner: ToolRunner,
         execution_context: PlanExecutionContext,
+        acms_integration: PlanExecutionACMSIntegration | None = None,
     ) -> None:
         if tool_runner is None:
             raise ValidationError("tool_runner must not be None")
@@ -310,6 +326,7 @@ class RuntimeExecuteActor:
 
         self._tool_runner = tool_runner
         self._execution_context = execution_context
+        self._acms_integration = acms_integration
         self._logger = logger.bind(
             plan_id=execution_context.plan_id,
             component="runtime_execute_actor",
@@ -325,6 +342,11 @@ class RuntimeExecuteActor:
         """The execution context."""
         return self._execution_context
 
+    @property
+    def acms_integration(self) -> PlanExecutionACMSIntegration | None:
+        """The ACMS integration, if configured."""
+        return self._acms_integration
+
     def execute(
         self,
         decisions: list[Any],
@@ -378,6 +400,20 @@ class RuntimeExecuteActor:
                     },
                 )
 
+            # Assemble context via ACMS if integration is configured.
+            # This replaces raw file dumps with policy-driven context views
+            # for LLM calls, ensuring scoped and budget-constrained context.
+            raw_context: dict[str, Any] = {
+                "plan_id": plan_id,
+                "decision_id": decision_id,
+                "step_text": step_text,
+                "sequence": sequence,
+            }
+            if self._acms_integration is not None:
+                llm_context = self._acms_integration.prepare_llm_context(raw_context)
+            else:
+                llm_context = raw_context
+
             # Discover available tools
             available_tools = self._tool_runner.discover()
             tool_call_count += len(available_tools)
@@ -387,7 +423,12 @@ class RuntimeExecuteActor:
                 plan_id=plan_id,
                 tool_name="stub/execute-step",
                 arguments={"step_text": step_text, "sequence": sequence},
-                result={"status": "stub_executed", "tools_found": len(available_tools)},
+                result={
+                    "status": "stub_executed",
+                    "tools_found": len(available_tools),
+                    "acms_context_assembled": self._acms_integration is not None,
+                    "llm_context_keys": list(llm_context.keys()),
+                },
                 success=True,
                 duration_ms=0.0,
                 sandbox_path=self._execution_context.sandbox_root,
@@ -409,6 +450,7 @@ class RuntimeExecuteActor:
                 decision_id=decision_id,
                 step_text=step_text,
                 invocation_id=invocation.invocation_id,
+                acms_context_assembled=self._acms_integration is not None,
             )
 
         elapsed_ms = (time.monotonic() - start_time) * 1000.0

src/cleveragents/application/services/plan_executor.py

@@ -11,6 +11,9 @@ into the Execute phase so that ``subplan_spawn`` and
 ``subplan_parallel_spawn`` decisions are realised as actual child plan
 executions.
 Updated in M6 to wire StrategyActor decisions through to Execute phase.
+Updated in M5 (ACMS) to wire ``PlanExecutionACMSIntegration`` into the
+Execute phase so that LLM calls use ACMS-assembled context instead of
+raw file dumps.
 """
 
 from __future__ import annotations
@@ -77,6 +80,9 @@ from cleveragents.tool.builtins.changeset import ChangeSet, ChangeSetCapture
 from cleveragents.tool.runner import ToolRunner
 
 if TYPE_CHECKING:
+    from cleveragents.acms.plan_execution_integration import (
+        PlanExecutionACMSIntegration,
+    )
     from cleveragents.application.services.error_recovery_service import (
         ErrorRecoveryService,
     )
@@ -352,6 +358,7 @@ class PlanExecutor:
         tier_service: ContextTierService | None = None,
         project_repository: NamespacedProjectRepository | None = None,
         resource_registry: ResourceRegistryService | None = None,
+        acms_integration: PlanExecutionACMSIntegration | None = None,
     ) -> None:
         """Initialize the plan executor.
 
@@ -391,6 +398,10 @@ class PlanExecutor:
                 up project links during strategize.
             resource_registry: Optional resource registry for resolving
                 resource locations during strategize.
+            acms_integration: Optional ACMS integration for assembling
+                context using policy-driven decisions before LLM calls.
+                When ``None``, raw context is passed to LLM calls
+                without ACMS assembly.
         """
         if lifecycle_service is None:
             raise ValidationError("lifecycle_service must not be None")
@@ -408,6 +419,7 @@ class PlanExecutor:
         self._tier_service = tier_service
         self._project_repository = project_repository
         self._resource_registry = resource_registry
+        self._acms_integration = acms_integration
         self._strategize_actor = strategize_actor or StrategizeStubActor()
         self._execute_actor = execute_actor or ExecuteStubActor()
         self._running_plan_ids: set[str] = set()
@@ -481,6 +493,11 @@ class PlanExecutor:
         """Return the subplan execution service, if configured."""
         return self._subplan_execution_service
 
+    @property
+    def acms_integration(self) -> PlanExecutionACMSIntegration | None:
+        """Return the ACMS integration, if configured."""
+        return self._acms_integration
+
     # ------------------------------------------------------------------
     # Subplan spawning helpers
     # ------------------------------------------------------------------
@@ -1126,6 +1143,7 @@ class PlanExecutor:
         runtime_actor = RuntimeExecuteActor(
             tool_runner=self._tool_runner,
             execution_context=self._execution_context,
+            acms_integration=self._acms_integration,
         )
         self._lifecycle.start_execute(plan_id)
         self._try_create_checkpoint(plan_id, "pre_execute")