fix: wire runtime parsing and template round-tripping

Author: lan17

engine/src/agent_control_engine/core.py

@@ -16,6 +16,7 @@
 from agent_control_models import (
     ConditionNode,
     ControlDefinition,
+    ControlDefinitionRuntime,
     ControlMatch,
     EvaluationRequest,
     EvaluationResponse,
@@ -47,7 +48,7 @@ class ControlWithIdentity(Protocol):
 
     id: int
     name: str
-    control: ControlDefinition
+    control: ControlDefinition | ControlDefinitionRuntime
 
 
 @dataclass

models/src/agent_control_models/controls.py

@@ -448,6 +448,24 @@ class ControlAction(BaseModel):
 MAX_CONDITION_DEPTH = 6
 
 
+def _validate_common_control_constraints(
+    condition: ConditionNode,
+    action: ControlAction,
+) -> None:
+    """Validate control constraints shared by authoring and runtime models."""
+    if condition.max_depth() > MAX_CONDITION_DEPTH:
+        raise ValueError(
+            f"Condition nesting depth exceeds maximum of {MAX_CONDITION_DEPTH}"
+        )
+
+    if (
+        action.decision == "steer"
+        and not condition.is_leaf()
+        and action.steering_context is None
+    ):
+        raise ValueError("Composite steer controls require action.steering_context")
+
+
 class ConditionNode(BaseModel):
     """Recursive boolean condition tree for control evaluation."""
 
@@ -689,19 +707,7 @@ def canonicalize_legacy_condition_shape(cls, data: Any) -> Any:
     @model_validator(mode="after")
     def validate_condition_constraints(self) -> Self:
         """Validate cross-field control constraints."""
-        if self.condition.max_depth() > MAX_CONDITION_DEPTH:
-            raise ValueError(
-                f"Condition nesting depth exceeds maximum of {MAX_CONDITION_DEPTH}"
-            )
-
-        if (
-            self.action.decision == "steer"
-            and not self.condition.is_leaf()
-            and self.action.steering_context is None
-        ):
-            raise ValueError(
-                "Composite steer controls require action.steering_context"
-            )
+        _validate_common_control_constraints(self.condition, self.action)
         has_template = self.template is not None
         has_template_values = self.template_values is not None
         if has_template != has_template_values:
@@ -710,6 +716,15 @@ def validate_condition_constraints(self) -> Self:
             )
         return self
 
+    def to_template_control_input(self) -> TemplateControlInput:
+        """Extract template-backed authoring input from a stored control definition."""
+        if self.template is None or self.template_values is None:
+            raise ValueError("Control definition is not template-backed")
+        return TemplateControlInput(
+            template=self.template,
+            template_values=dict(self.template_values),
+        )
+
     def iter_condition_leaves(self) -> Iterator[ConditionNode]:
         """Yield leaf conditions in evaluation order."""
         yield from self.condition.iter_leaves()
@@ -808,6 +823,54 @@ class ControlDefinitionRuntime(BaseModel):
     action: ControlAction = Field(..., description="What action to take when control matches")
     tags: list[str] = Field(default_factory=list, description="Tags for categorization")
 
+    @model_validator(mode="before")
+    @classmethod
+    def canonicalize_legacy_condition_shape(cls, data: Any) -> Any:
+        """Accept legacy flat leaf payloads during runtime parsing."""
+        return ControlDefinition.canonicalize_payload(data)
+
+    @model_validator(mode="after")
+    def validate_condition_constraints(self) -> Self:
+        """Validate runtime-relevant control constraints."""
+        _validate_common_control_constraints(self.condition, self.action)
+        return self
+
+    def iter_condition_leaves(self) -> Iterator[ConditionNode]:
+        """Yield leaf conditions in evaluation order."""
+        yield from self.condition.iter_leaves()
+
+    def iter_condition_leaf_parts(self) -> Iterator[ConditionLeafParts]:
+        """Yield leaf selector/evaluator pairs in evaluation order."""
+        yield from self.condition.iter_leaf_parts()
+
+    def observability_identity(self) -> ControlObservabilityIdentity:
+        """Return a deterministic representative identity for observability."""
+        all_evaluators: list[str] = []
+        all_selector_paths: list[str] = []
+        seen_evaluators: set[str] = set()
+        seen_selector_paths: set[str] = set()
+        leaf_count = 0
+
+        for selector, evaluator in self.iter_condition_leaf_parts():
+            leaf_count += 1
+            selector_path = selector.path or "*"
+
+            if evaluator.name not in seen_evaluators:
+                seen_evaluators.add(evaluator.name)
+                all_evaluators.append(evaluator.name)
+
+            if selector_path not in seen_selector_paths:
+                seen_selector_paths.add(selector_path)
+                all_selector_paths.append(selector_path)
+
+        return ControlObservabilityIdentity(
+            selector_path=all_selector_paths[0] if all_selector_paths else None,
+            evaluator_name=all_evaluators[0] if all_evaluators else None,
+            leaf_count=leaf_count,
+            all_evaluators=all_evaluators,
+            all_selector_paths=all_selector_paths,
+        )
+
     def to_control_definition(self) -> ControlDefinition:
         """Promote a runtime control to the full public model when needed."""
         return ControlDefinition.model_validate(

models/tests/test_control_templates.py

@@ -3,7 +3,12 @@
 from __future__ import annotations
 
 import pytest
-from agent_control_models import ControlDefinition, TemplateControlInput, TemplateDefinition
+from agent_control_models import (
+    ControlDefinition,
+    ControlDefinitionRuntime,
+    TemplateControlInput,
+    TemplateDefinition,
+)
 from agent_control_models.server import CreateControlRequest
 from pydantic import ValidationError
 
@@ -97,3 +102,51 @@ def test_create_control_request_rejects_mixed_raw_and_template_payload() -> None
                 },
             }
         )
+
+
+def test_control_definition_can_round_trip_to_template_control_input() -> None:
+    control = ControlDefinition.model_validate(
+        {
+            "execution": "server",
+            "scope": {"step_types": ["llm"], "stages": ["pre"]},
+            "condition": {
+                "selector": {"path": "input"},
+                "evaluator": {
+                    "name": "regex",
+                    "config": {"pattern": "hello"},
+                },
+            },
+            "action": {"decision": "deny"},
+            "template": VALID_TEMPLATE,
+            "template_values": {"pattern": "hello"},
+        }
+    )
+
+    template_input = control.to_template_control_input()
+
+    assert template_input.template.parameters["pattern"].label == "Pattern"
+    assert template_input.template.parameters["pattern"].type == "regex_re2"
+    assert template_input.template.definition_template == VALID_TEMPLATE["definition_template"]
+    assert template_input.template_values == {"pattern": "hello"}
+
+
+def test_control_definition_runtime_ignores_template_metadata() -> None:
+    runtime_control = ControlDefinitionRuntime.model_validate(
+        {
+            "execution": "server",
+            "scope": {"step_types": ["llm"], "stages": ["pre"]},
+            "condition": {
+                "selector": {"path": "input"},
+                "evaluator": {
+                    "name": "regex",
+                    "config": {"pattern": "hello"},
+                },
+            },
+            "action": {"decision": "deny"},
+            "template": VALID_TEMPLATE,
+            "template_values": {"pattern": "hello"},
+        }
+    )
+
+    assert runtime_control.execution == "server"
+    assert runtime_control.action.decision == "deny"

sdks/python/src/agent_control/__init__.py

@@ -164,6 +164,13 @@ def get_server_controls() -> list[dict[str, Any]] | None:
     return state.server_controls
 
 
+def to_template_control_input(
+    data: dict[str, Any] | ControlDefinition,
+) -> TemplateControlInput:
+    """Convert stored control data into template authoring input."""
+    return controls.to_template_control_input(data)
+
+
 def _publish_server_controls(
     controls: list[dict[str, Any]] | None,
 ) -> list[dict[str, Any]] | None:
@@ -1316,6 +1323,7 @@ async def main():
     "update_control",
     "validate_control_data",
     "render_control_template",
+    "to_template_control_input",
     # Decorator
     "control",
     "ControlViolationError",

sdks/python/src/agent_control/controls.py

@@ -265,6 +265,22 @@ async def render_control_template(
     return cast(dict[str, Any], response.json())
 
 
+def to_template_control_input(
+    data: dict[str, Any] | ControlDefinition,
+) -> TemplateControlInput:
+    """Convert stored control data into template authoring input.
+
+    This is the supported reshape path for template-backed controls returned by
+    ``GET /controls/{id}`` or ``GET /controls/{id}/data`` before submitting them
+    back to ``set_control_data``.
+    """
+    if isinstance(data, ControlDefinition):
+        control_def = data
+    else:
+        control_def = ControlDefinition.model_validate(data)
+    return control_def.to_template_control_input()
+
+
 async def add_rule_to_control(
     client: AgentControlClient,
     control_id: int,

sdks/python/src/agent_control/evaluation.py

@@ -8,6 +8,7 @@
 from agent_control_engine.core import ControlEngine
 from agent_control_models import (
     ControlDefinition,
+    ControlDefinitionRuntime,
     ControlExecutionEvent,
     ControlMatch,
     EvaluationRequest,
@@ -33,7 +34,7 @@
 
 
 def _observability_metadata(
-    control_def: ControlDefinition,
+    control_def: ControlDefinition | ControlDefinitionRuntime,
 ) -> tuple[str | None, str | None, dict[str, object]]:
     """Return representative event fields plus full composite context."""
     identity = control_def.observability_identity()
@@ -159,7 +160,7 @@ class _ControlAdapter:
 
     id: int
     name: str
-    control: "ControlDefinition"
+    control: "ControlDefinition | ControlDefinitionRuntime"
 
 
 def _get_applicable_controls(
@@ -193,7 +194,7 @@ def _has_applicable_prefiltered_server_controls(
 
     for control in server_control_payloads:
         try:
-            control_def = ControlDefinition.model_validate(control["control"])
+            control_def = ControlDefinitionRuntime.model_validate(control["control"])
             parsed_server_controls.append(
                 _ControlAdapter(
                     id=control["id"],
@@ -315,7 +316,7 @@ async def check_evaluation_with_local(
             continue
 
         try:
-            control_def = ControlDefinition.model_validate(control_data)
+            control_def = ControlDefinitionRuntime.model_validate(control_data)
             for _, evaluator_spec in control_def.iter_condition_leaf_parts():
                 evaluator_name = evaluator_spec.name
 

sdks/python/tests/test_controls_api.py

@@ -107,3 +107,44 @@ async def test_render_control_template_calls_preview_endpoint() -> None:
             "template_values": {},
         },
     )
+
+
+def test_to_template_control_input_reshapes_stored_control_data() -> None:
+    template_input = agent_control.controls.to_template_control_input(
+        {
+            "execution": "server",
+            "scope": {"step_types": ["llm"], "stages": ["pre"]},
+            "condition": {
+                "selector": {"path": "input"},
+                "evaluator": {
+                    "name": "regex",
+                    "config": {"pattern": "hello"},
+                },
+            },
+            "action": {"decision": "deny"},
+            "template": {
+                "parameters": {
+                    "pattern": {
+                        "type": "regex_re2",
+                        "label": "Pattern",
+                    }
+                },
+                "definition_template": {
+                    "execution": "server",
+                    "scope": {"step_types": ["llm"], "stages": ["pre"]},
+                    "condition": {
+                        "selector": {"path": "input"},
+                        "evaluator": {
+                            "name": "regex",
+                            "config": {"pattern": {"$param": "pattern"}},
+                        },
+                    },
+                    "action": {"decision": "deny"},
+                },
+            },
+            "template_values": {"pattern": "hello"},
+        }
+    )
+
+    assert isinstance(template_input, TemplateControlInput)
+    assert template_input.template_values == {"pattern": "hello"}

server/src/agent_control_server/endpoints/agents.py

@@ -1,11 +1,12 @@
 from collections.abc import Sequence
-from typing import Any, Protocol
+from typing import Any
 
 from agent_control_engine import list_evaluators
 from agent_control_models.agent import Agent as APIAgent
 from agent_control_models.agent import StepSchema
-from agent_control_models.controls import ControlDefinition
+from agent_control_models.controls import ControlDefinitionRuntime
 from agent_control_models.errors import ErrorCode, ValidationErrorItem
+from agent_control_models.policy import Control as APIControl
 from agent_control_models.server import (
     AgentControlsResponse,
     AgentSummary,
@@ -82,13 +83,6 @@
 type StepKeyTuple = tuple[str, str]
 
 
-class _ControlWithDefinition(Protocol):
-    """Minimal control shape needed for evaluator dependency scans."""
-
-    name: str
-    control: ControlDefinition
-
-
 # =============================================================================
 # List Agents Models
 # =============================================================================
@@ -119,7 +113,7 @@ def _validate_controls_for_agent(agent: Agent, controls: list[Control]) -> list[
             continue
 
         try:
-            control_definition = ControlDefinition.model_validate(control.data)
+            control_definition = ControlDefinitionRuntime.model_validate(control.data)
         except ValidationError:
             errors.append(f"Control '{control.name}' has corrupted data")
             continue
@@ -165,7 +159,7 @@ def _validate_controls_for_agent(agent: Agent, controls: list[Control]) -> list[
 
 
 def _find_referencing_controls_for_removed_evaluators(
-    controls: Sequence[_ControlWithDefinition],
+    controls: Sequence[APIControl],
     agent_name: str,
     remove_evaluator_set: set[str],
 ) -> list[tuple[str, str]]: