feat(sdk): out of box controls - part 2

engine/src/agent_control_engine/selectors.py

@@ -16,7 +16,12 @@ def select_data(step: Step, path: str) -> Any:
         The selected value, or None if the path doesn't exist.
     """
     if not path or path == "*":
-        return step.model_dump(mode="json")
+        # Preserve the original wildcard contract as Step gains opt-in fields.
+        # New fields must be selected explicitly so strict JSON controls written
+        # against the legacy full-step shape do not change behavior.
+        return step.model_dump(mode="json", exclude={"canonical_name"})
+    if path == "canonical_name":
+        return step.canonical_name or step.name
 
     parts = path.split(".")
     current: Any = step

engine/tests/test_selectors.py

@@ -31,6 +31,7 @@ def llm_step_payload() -> Step:
     "path,expected",
     [
         ("name", "search_database"),
+        ("canonical_name", "search_database"),
         ("input.query", "SELECT * FROM users"),
         ("input.limit", 10),
         ("input.nested.key", "value"),
@@ -85,6 +86,31 @@ def test_select_data_none_handling():
     assert result is None
 
 
+def test_select_data_prefers_explicit_canonical_name() -> None:
+    payload = Step(
+        type="tool",
+        name="writer.web_search",
+        canonical_name="web_search",
+        input={},
+    )
+
+    assert select_data(payload, "canonical_name") == "web_search"
+
+
+def test_wildcard_preserves_legacy_shape_when_canonical_name_is_present() -> None:
+    payload = Step(
+        type="tool",
+        name="writer.web_search",
+        canonical_name="web_search",
+        input={},
+    )
+
+    selected = select_data(payload, "*")
+
+    assert selected == payload.model_dump(mode="json", exclude={"canonical_name"})
+    assert "canonical_name" not in selected
+
+
 def test_list_selection():
     """Test that selecting a path pointing to a list returns the whole list."""
     # Given: a payload with a list in the output

evaluators/builtin/src/agent_control_evaluators/sql/evaluator.py

@@ -487,13 +487,49 @@ def _check_limits(
                 if limit_node:
                     limit_value = self._extract_limit_value(limit_node)
 
+                    if limit_value is None and (
+                        self.config.max_limit is not None
+                        or self.config.max_result_window is not None
+                    ):
+                        return EvaluatorResult(
+                            matched=True,
+                            confidence=1.0,
+                            message=(
+                                "LIMIT must be a numeric literal when a maximum "
+                                "result bound is configured"
+                            ),
+                            metadata={
+                                "query": query[:100],
+                                "violation": "indeterminate_limit",
+                            },
+                        )
+
                     # Extract OFFSET value if present (use args.get() for direct child only)
                     offset_node = select_node.args.get("offset")
                     offset_value = 0
                     if offset_node:
-                        offset_value = self._extract_offset_value(offset_node) or 0
+                        extracted_offset = self._extract_offset_value(offset_node)
+                        if (
+                            extracted_offset is None
+                            and self.config.max_result_window is not None
+                        ):
+                            return EvaluatorResult(
+                                matched=True,
+                                confidence=1.0,
+                                message=(
+                                    "OFFSET must be a numeric literal when a maximum "
+                                    "result window is configured"
+                                ),
+                                metadata={
+                                    "query": query[:100],
+                                    "violation": "indeterminate_offset",
+                                },
+                            )
+                        if extracted_offset is not None:
+                            offset_value = extracted_offset
 
-                    # Check LIMIT value (skip if indeterminate)
+                    # Check LIMIT value. Indeterminate values with configured
+                    # maxima returned a fail-closed result above.
                     if limit_value is not None:
                         # Check max_limit
                         if (
@@ -1283,6 +1319,9 @@ def _get_operation_name(self, stmt: exp.Expression) -> str | None:
             exp.Delete: "DELETE",
             exp.Merge: "MERGE",
             # DDL (Data Definition Language)
+            # PostgreSQL SELECT ... INTO creates a table but sqlglot represents
+            # it as a Select containing an Into node rather than a Create node.
+            exp.Into: "CREATE",
             exp.Create: "CREATE",
             exp.Drop: "DROP",
             exp.Alter: "ALTER",

evaluators/builtin/tests/sql/test_sql.py

@@ -262,6 +262,12 @@ async def test_allowlist_mode_select_only(self):
         assert result.error is None
         assert result.matched is True
 
+        # PostgreSQL SELECT INTO creates a table despite its SELECT root node.
+        result = await evaluator.evaluate("SELECT * INTO backup FROM users")
+        assert result.error is None
+        assert result.matched is True
+        assert "CREATE" in result.metadata["blocked"]
+
     @pytest.mark.asyncio
     async def test_block_ddl_flag(self):
         """Should block all DDL operations when block_ddl=True."""
@@ -856,15 +862,36 @@ async def test_limit_with_offset(self):
 
     @pytest.mark.asyncio
     async def test_limit_all_allowed(self):
-        """Should allow LIMIT ALL (indeterminate limits are allowed)."""
+        """Should allow LIMIT ALL when LIMIT presence is not required."""
         config = SQLEvaluatorConfig(max_limit=1000)
         evaluator = SQLEvaluator(config)
 
-        # LIMIT ALL should be allowed (indeterminate limits are skipped)
+        # PostgreSQL normalizes LIMIT ALL to an omitted LIMIT. max_limit alone
+        # only constrains a present numeric limit; require_limit blocks this form.
         result = await evaluator.evaluate("SELECT * FROM users LIMIT ALL")
         assert result.error is None
         assert result.matched is False
 
+    @pytest.mark.asyncio
+    async def test_indeterminate_limit_and_offset_fail_closed(self):
+        """Parameterized or computed bounds must not bypass configured maxima."""
+        config = SQLEvaluatorConfig(max_limit=1000, max_result_window=1000)
+        evaluator = SQLEvaluator(config)
+
+        results = [
+            await evaluator.evaluate("SELECT * FROM users LIMIT $1"),
+            await evaluator.evaluate("SELECT * FROM users LIMIT (1000 + 1)"),
+            await evaluator.evaluate("SELECT * FROM users LIMIT 1000 OFFSET $1"),
+        ]
+
+        assert all(result.error is None for result in results)
+        assert all(result.matched is True for result in results)
+        assert [result.metadata["violation"] for result in results] == [
+            "indeterminate_limit",
+            "indeterminate_limit",
+            "indeterminate_offset",
+        ]
+
     @pytest.mark.asyncio
     async def test_require_and_max_limit_combined(self):
         """Should enforce both require_limit and max_limit."""

models/src/agent_control_models/agent.py

@@ -150,6 +150,15 @@ class Step(BaseModel):
     name: str = Field(
         ..., min_length=1, description="Step name (tool name or model/chain id)"
     )
+    canonical_name: str | None = Field(
+        default=None,
+        min_length=1,
+        exclude_if=lambda value: value is None,
+        description=(
+            "Optional integration-independent identity for a qualified step name "
+            "(for example, 'web_search' for 'writer.web_search')."
+        ),
+    )
     input: JSONValue = Field(
         ..., description="Input content for this step"
     )

models/src/agent_control_models/controls.py

@@ -27,7 +27,8 @@ class ControlSelector(BaseModel):
         default="*",
         description=(
             "Path to data using dot notation. "
-            "Examples: 'input', 'output', 'context.user_id', 'name', 'type', '*'"
+            "Examples: 'input', 'output', 'context.user_id', 'name', "
+            "'canonical_name', 'type', '*'"
         ),
     )
 
@@ -43,7 +44,15 @@ def validate_path(cls, v: str | None) -> str:
             )
 
         # Valid root fields
-        valid_roots = {"input", "output", "name", "type", "context", "*"}
+        valid_roots = {
+            "input",
+            "output",
+            "name",
+            "canonical_name",
+            "type",
+            "context",
+            "*",
+        }
         root = v.split(".")[0]
 
         if root not in valid_roots:
@@ -61,6 +70,7 @@ def validate_path(cls, v: str | None) -> str:
                 {"path": "input"},
                 {"path": "*"},
                 {"path": "name"},
+                {"path": "canonical_name"},
                 {"path": "output"},
             ]
         }

sdks/python/src/agent_control/evaluation.py

@@ -517,6 +517,7 @@ def _with_parse_errors(result: EvaluationResult) -> EvaluationResult:
 async def evaluate_controls(
     step_name: str,
     *,
+    canonical_step_name: str | None = None,
     input: Any | None = None,
     output: Any | None = None,
     context: dict[str, Any] | None = None,
@@ -530,6 +531,10 @@ async def evaluate_controls(
 ) -> EvaluationResult:
     """Evaluate controls for a step.
 
+    ``canonical_step_name`` is an optional integration-independent identity for
+    qualified tool names, such as ``web_search`` for ``writer.web_search``.
+    Integrations should leave it unset when they cannot provide that identity.
+
     When ``target_type`` and ``target_id`` are both supplied, the request
     is target-bearing: the server merges target bindings into the
     effective control set. If they are omitted, the SDK falls back to the
@@ -547,6 +552,7 @@ async def evaluate_controls(
     step_dict: dict[str, Any] = {
         "type": step_type,
         "name": step_name,
+        "canonical_name": canonical_step_name,
         "input": input if input is not None else default_value,
         "output": output if output is not None else default_value,
     }

sdks/python/src/agent_control/integrations/_core.py

@@ -48,6 +48,7 @@ async def _evaluate_and_enforce(
     agent_name: str,
     step_name: str,
     *,
+    canonical_step_name: str | None = None,
     input: Any | None = None,
     output: Any | None = None,
     context: dict[str, Any] | None = None,
@@ -58,6 +59,7 @@ async def _evaluate_and_enforce(
 
     result = await agent_control.evaluate_controls(
         step_name=step_name,
+        canonical_step_name=canonical_step_name,
         input=input,
         output=output,
         context=context,

sdks/python/src/agent_control/integrations/google_adk/plugin.py

@@ -268,6 +268,7 @@ async def before_tool_callback(
             return None
 
         step_name = self._resolve_tool_step_name(tool, tool_context=tool_context)
+        canonical_step_name = resolve_tool_name(tool)
         self._ensure_step_known(self._build_tool_step_schema(tool, step_name))
         context = self._safe_context(
             step_type="tool",
@@ -281,6 +282,7 @@ async def before_tool_callback(
             await _evaluate_and_enforce(
                 self.agent_name,
                 step_name,
+                canonical_step_name=canonical_step_name,
                 input=tool_args,
                 context=context,
                 step_type="tool",
@@ -311,6 +313,7 @@ async def after_tool_callback(
             return None
 
         step_name = self._resolve_tool_step_name(tool, tool_context=tool_context)
+        canonical_step_name = resolve_tool_name(tool)
         self._ensure_step_known(self._build_tool_step_schema(tool, step_name))
         context = self._safe_context(
             step_type="tool",
@@ -325,6 +328,7 @@ async def after_tool_callback(
             await _evaluate_and_enforce(
                 self.agent_name,
                 step_name,
+                canonical_step_name=canonical_step_name,
                 input=tool_args,
                 output=result,
                 context=context,

sdks/python/src/agent_control/integrations/strands/plugin.py

@@ -115,6 +115,7 @@ async def _evaluate_and_enforce(
     ) -> None:
         result = await agent_control.evaluate_controls(
             step_name=step_name,
+            canonical_step_name=step_name if step_type == "tool" else None,
             input=input,
             output=output,
             context=context,

sdks/python/tests/test_evaluation.py

@@ -105,6 +105,30 @@ async def test_evaluate_controls_with_explicit_agent_name(monkeypatch):
     mock_check.assert_called_once()
 
 
+@pytest.mark.asyncio
+async def test_evaluate_controls_forwards_canonical_step_name(monkeypatch):
+    """Qualified integrations can send a stable tool identity separately."""
+    mock_result = EvaluationResult(is_safe=True, confidence=1.0)
+    mock_check = AsyncMock(return_value=mock_result)
+    monkeypatch.setattr(evaluation, "check_evaluation_with_local", mock_check)
+
+    with patch("agent_control.state.server_url", "http://localhost:8000"), patch(
+        "agent_control.state.api_key", None
+    ):
+        await evaluation.evaluate_controls(
+            step_name="writer.web_search",
+            canonical_step_name="web_search",
+            input={},
+            step_type="tool",
+            stage="pre",
+            agent_name="test-bot",
+        )
+
+    step = mock_check.await_args.kwargs["step"]
+    assert step.name == "writer.web_search"
+    assert step.canonical_name == "web_search"
+
+
 @pytest.mark.asyncio
 async def test_evaluate_controls_with_context(monkeypatch):
     """evaluate_controls should pass context through to evaluation."""

sdks/python/tests/test_google_adk_plugin.py

@@ -369,6 +369,7 @@ async def test_tool_callbacks_scope_step_name_by_agent(plugin_module):
         )
 
     assert mock_eval.await_args.args[1] == "writer.get_weather"
+    assert mock_eval.await_args.kwargs["canonical_step_name"] == "get_weather"
 
 
 @pytest.mark.asyncio

sdks/python/tests/test_strands_plugin.py

@@ -191,6 +191,29 @@ async def test_evaluate_and_enforce_safe_result(agent_control_hook):
             step_type="llm",
             stage="pre"
         )
+        assert mock_evaluate.await_args.kwargs["canonical_step_name"] is None
+
+
+@pytest.mark.asyncio
+async def test_evaluate_and_enforce_forwards_canonical_tool_name(agent_control_hook):
+    """Strands tool hooks retain the bare tool name as canonical identity."""
+    with patch(
+        "agent_control.integrations.strands.plugin.agent_control.evaluate_controls"
+    ) as mock_evaluate:
+        mock_result = MagicMock(spec=EvaluationResult)
+        mock_result.is_safe = True
+        mock_result.matches = []
+        mock_result.errors = []
+        mock_evaluate.return_value = mock_result
+
+        await agent_control_hook._evaluate_and_enforce(
+            step_name="web_search",
+            input={},
+            step_type="tool",
+            stage="pre",
+        )
+
+        assert mock_evaluate.await_args.kwargs["canonical_step_name"] == "web_search"
 
 
 @pytest.mark.asyncio

sdks/typescript/src/generated/models/control-selector.ts

@@ -18,7 +18,7 @@ import { SDKValidationError } from "./errors/sdk-validation-error.js";
  */
 export type ControlSelector = {
   /**
-   * Path to data using dot notation. Examples: 'input', 'output', 'context.user_id', 'name', 'type', '*'
+   * Path to data using dot notation. Examples: 'input', 'output', 'context.user_id', 'name', 'canonical_name', 'type', '*'
    */
   path?: string | null | undefined;
 };