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

engine/src/agent_control_engine/selectors.py

@@ -17,6 +17,8 @@ def select_data(step: Step, path: str) -> Any:
     """
     if not path or path == "*":
         return step.model_dump(mode="json")
+    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,17 @@ 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_list_selection():
     """Test that selecting a path pointing to a list returns the whole list."""
     # Given: a payload with a list in the output

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,
@@ -547,6 +548,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

@@ -57,9 +57,9 @@ def json(self) -> dict[str, object]:
         json={
             "agent_name": "agent-example_01",
             "step": {
-                "type": "llm",
-                "name": "chat",
-                "input": "hello",
+                    "type": "llm",
+                    "name": "chat",
+                    "input": "hello",
                 "output": None,
                 "context": None,
             },

sdks/python/tests/test_google_adk_plugin.py

@@ -10,7 +10,6 @@
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
-
 from agent_control import ControlSteerError, ControlViolationError
 from agent_control._state import state
 
@@ -369,6 +368,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/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;
 };

sdks/typescript/src/generated/models/step.ts

@@ -3,11 +3,16 @@
  */
 
 import * as z from "zod/v4-mini";
+import { remap as remap$ } from "../lib/primitives.js";
 
 /**
  * Runtime payload for an agent step invocation.
  */
 export type Step = {
+  /**
+   * Optional integration-independent identity for a qualified step name (for example, 'web_search' for 'writer.web_search').
+   */
+  canonicalName?: string | null | undefined;
   /**
    * Optional context (conversation history, metadata, etc.)
    */
@@ -32,6 +37,7 @@ export type Step = {
 
 /** @internal */
 export type Step$Outbound = {
+  canonical_name?: string | null | undefined;
   context?: { [k: string]: any } | null | undefined;
   input: any;
   name: string;
@@ -40,14 +46,20 @@ export type Step$Outbound = {
 };
 
 /** @internal */
-export const Step$outboundSchema: z.ZodMiniType<Step$Outbound, Step> = z.object(
-  {
+export const Step$outboundSchema: z.ZodMiniType<Step$Outbound, Step> = z.pipe(
+  z.object({
+    canonicalName: z.optional(z.nullable(z.string())),
     context: z.optional(z.nullable(z.record(z.string(), z.any()))),
     input: z.any(),
     name: z.string(),
     output: z.optional(z.nullable(z.any())),
     type: z.string(),
-  },
+  }),
+  z.transform((v) => {
+    return remap$(v, {
+      canonicalName: "canonical_name",
+    });
+  }),
 );
 
 export function stepToJSON(step: Step): string {

server/alembic/versions/f3a1c8d7e2b4_out_of_box_control_seed_identity.py

@@ -17,6 +17,8 @@
 branch_labels = None
 depends_on = None
 
+_CANONICAL_NAME_SEED_SOURCE_ID = "oob-only-approved-tools-may-run"
+
 
 def upgrade() -> None:
     op.add_column("controls", sa.Column("seed_source_id", sa.String(length=255), nullable=True))
@@ -35,6 +37,36 @@ def upgrade() -> None:
 
 
 def downgrade() -> None:
+    # Older servers reject ``canonical_name`` selectors. Retire only the seeded
+    # control that still uses that selector, and make both its current payload
+    # and historical snapshots parseable before rolling the application back.
+    op.execute(
+        f"""
+        UPDATE control_versions AS version
+        SET snapshot = jsonb_set(
+            version.snapshot,
+            '{{data,condition,selector,path}}',
+            '"name"'::jsonb
+        )
+        FROM controls AS control
+        WHERE version.control_id = control.id
+          AND control.seed_source_id = '{_CANONICAL_NAME_SEED_SOURCE_ID}'
+          AND version.snapshot #>> '{{data,condition,selector,path}}' = 'canonical_name'
+        """
+    )
+    op.execute(
+        f"""
+        UPDATE controls
+        SET data = jsonb_set(
+                data,
+                '{{condition,selector,path}}',
+                '"name"'::jsonb
+            ),
+            deleted_at = COALESCE(deleted_at, CURRENT_TIMESTAMP)
+        WHERE seed_source_id = '{_CANONICAL_NAME_SEED_SOURCE_ID}'
+          AND data #>> '{{condition,selector,path}}' = 'canonical_name'
+        """
+    )
     with op.get_context().autocommit_block():
         op.execute("DROP INDEX CONCURRENTLY IF EXISTS idx_controls_namespace_seed_source")
     op.drop_column("controls", "seed_opted_out_at")