feat(sdk): add step_context to StepRecorder.call()/acall()

call()/acall() run func(*args, **kwargs) and forward every unrecognized
keyword straight to func, so a plain `context=` parameter on call()
would collide with (and silently steal) a same-named kwarg on func -
exactly the case for functions like run_banking_model(..., context=...).
step_context is merged into the recorded child's Step.context instead,
without touching what gets forwarded to func. On failure, the error
key added to context takes precedence over the same key in step_context.

Co-Authored-By: Claude Code <noreply@anthropic.com>

Author: jasmine-ab-tea

sdks/python/src/agent_control/step_recorder.py

@@ -132,13 +132,17 @@ def _record_call_result(
         step_name: str | None,
         step_type: str | None,
         tools: ToolsConfiguration,
+        step_context: dict[str, Any] | None,
         error: BaseException | None,
     ) -> None:
         payload = _create_evaluation_payload(
             func, args, kwargs, output, step_name, step_type, tools
         )
+        context = {**(payload.get("context") or {}), **(step_context or {})}
         if error is not None:
-            payload["context"] = {**(payload.get("context") or {}), "error": repr(error)}
+            context["error"] = repr(error)
+        if context:
+            payload["context"] = context
         self._children.append(Step.model_validate(payload))
 
     def call(
@@ -148,15 +152,27 @@ def call(
         step_type: str | None = None,
         step_name: str | None = None,
         tools: ToolsConfiguration = None,
+        step_context: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> T:
-        """Run ``func``, record it via ``@control()``'s capture logic, return its output."""
+        """Run ``func``, record it via ``@control()``'s capture logic, return its output.
+
+        ``step_context`` is merged into the recorded child's ``context`` (an
+        ``error`` key added on failure takes precedence over the same key in
+        ``step_context``). It is named ``step_context`` rather than ``context``
+        so it can't collide with a ``context`` keyword argument on ``func``
+        itself, which ``**kwargs`` would otherwise forward unchanged.
+        """
         try:
             output = func(*args, **kwargs)
         except Exception as exc:
-            self._record_call_result(func, args, kwargs, None, step_name, step_type, tools, exc)
+            self._record_call_result(
+                func, args, kwargs, None, step_name, step_type, tools, step_context, exc
+            )
             raise
-        self._record_call_result(func, args, kwargs, output, step_name, step_type, tools, None)
+        self._record_call_result(
+            func, args, kwargs, output, step_name, step_type, tools, step_context, None
+        )
         return output
 
     async def acall(
@@ -166,15 +182,20 @@ async def acall(
         step_type: str | None = None,
         step_name: str | None = None,
         tools: ToolsConfiguration = None,
+        step_context: dict[str, Any] | None = None,
         **kwargs: Any,
     ) -> T:
         """Async counterpart to :meth:`call`."""
         try:
             output = await func(*args, **kwargs)
         except Exception as exc:
-            self._record_call_result(func, args, kwargs, None, step_name, step_type, tools, exc)
+            self._record_call_result(
+                func, args, kwargs, None, step_name, step_type, tools, step_context, exc
+            )
             raise
-        self._record_call_result(func, args, kwargs, output, step_name, step_type, tools, None)
+        self._record_call_result(
+            func, args, kwargs, output, step_name, step_type, tools, step_context, None
+        )
         return output
 
     def build(self) -> Step:

sdks/python/tests/test_step_recorder.py

@@ -1,5 +1,6 @@
 """Tests for StepRecorder, the incremental Step tree builder."""
 
+from typing import Any
 from unittest.mock import AsyncMock, patch
 
 import pytest
@@ -89,6 +90,72 @@ async def run_banking_model(prompt: str) -> str:
     assert child.output == "answer: hello"
 
 
+def test_call_step_context_is_recorded():
+    """step_context is merged into the recorded child's context."""
+
+    def lookup_account(account_id: str) -> dict:
+        return {"balance": 100}
+
+    with record_step("trace", "t") as trace:
+        trace.call(
+            lookup_account,
+            account_id="acct-1",
+            step_type="tool",
+            step_name="lookup_account",
+            step_context={"executed_function": "lookup_account"},
+        )
+
+    [child] = trace.build().children
+    assert child.context == {"executed_function": "lookup_account"}
+
+
+def test_call_step_context_does_not_collide_with_func_context_kwarg():
+    """step_context is merged into Step.context, not forwarded to func - a plain
+    `context` kwarg on call() would instead be swallowed by **kwargs and sent
+    to func, silently dropping the caller's intended Step metadata."""
+
+    received: dict[str, Any] = {}
+
+    def run_banking_model(prompt: str, *, context: dict[str, Any] | None = None) -> str:
+        received["context"] = context
+        return f"answer: {prompt}"
+
+    with record_step("trace", "t") as trace:
+        result = trace.call(
+            run_banking_model,
+            "hello",
+            context={"account": "acct-1"},
+            step_type="llm",
+            step_name="respond",
+            step_context={"executed_function": "run_banking_model"},
+        )
+
+    assert result == "answer: hello"
+    assert received["context"] == {"account": "acct-1"}
+    [child] = trace.build().children
+    assert child.context == {"executed_function": "run_banking_model"}
+
+
+def test_call_step_context_and_error_merge_on_failure():
+    """On failure, the error key is added to step_context's dict and wins on collision."""
+
+    def flaky(x: int) -> int:
+        raise ValueError("boom")
+
+    with record_step("trace", "t") as trace:
+        with pytest.raises(ValueError, match="boom"):
+            trace.call(
+                flaky,
+                1,
+                step_type="tool",
+                step_name="flaky",
+                step_context={"attempt": 1, "error": "overwritten"},
+            )
+
+    [child] = trace.build().children
+    assert child.context == {"attempt": 1, "error": "ValueError('boom')"}
+
+
 def test_call_records_exception_and_reraises():
     """A failing call is still recorded as a child (with the error in context), then re-raised."""