furyhawk/pydantic-ai-harness
1
0
Fork 0
mirror of https://github.com/pydantic/pydantic-ai-harness.git synced 2026-07-21 02:45:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Bump pydantic-ai floor to 2.1 (#303)

Browse Source
This commit is contained in:
Aditya Vardhan
2026-07-01 00:02:14 +05:30
committed by GitHub
parent 51ba5693bf
commit 6eb91dfe86
12 changed files with 306 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pyproject.toml
+3 -3
View File
@@ -29,7 +29,7 @@ classifiers = [
]
dependencies = [
"httpx>=0.28.1",
"pydantic-ai-slim>=1.105.0",
"pydantic-ai-slim>=2.1.0",
]
[project.optional-dependencies]
@@ -48,7 +48,7 @@ dbos = [
]
logfire = [
'logfire>=4.31.0',
"pydantic-ai-slim[spec]>=1.105.0",
"pydantic-ai-slim[spec]>=2.1.0",
]
[project.urls]
@@ -70,7 +70,7 @@ dev = [
'logfire[httpx]>=4.31.0',
'dirty-equals>=0.9.0',
'inline-snapshot>=0.32.5',
'pydantic-ai-slim[spec]>=1.95.1',
'pydantic-ai-slim[spec]>=2.1.0',
"pytest-examples>=0.0.18",
"pytest-recording>=0.13.4",
]
tests/code_mode/test_code_mode.py
+128 -128
View File
@@ -156,7 +156,7 @@ def _make_address_tool_def(name: str, description: str, addr_field: str) -> Tool
)
class _StaticToolset(AbstractToolset[None]):
class _StaticToolset(AbstractToolset[object]):
"""A minimal `AbstractToolset` that returns a fixed set of `ToolDefinition`s.
Mirrors the `MockToolsetWithInstructions` pattern from `pydantic_ai/tests/test_toolsets.py`.
@@ -172,7 +172,7 @@ class _StaticToolset(AbstractToolset[None]):
def id(self) -> str | None:
return None # pragma: no cover - required by AbstractToolset, never read in tests
async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
async def get_tools(self, ctx: RunContext[object]) -> dict[str, ToolsetTool[object]]:
return {
td.name: ToolsetTool(
toolset=self,
@@ -187,8 +187,8 @@ class _StaticToolset(AbstractToolset[None]):
self,
name: str,
tool_args: dict[str, Any],
ctx: RunContext[None],
tool: ToolsetTool[None],
ctx: RunContext[object],
tool: ToolsetTool[object],
) -> Any:
# Tests always set up `_results` for every tool name they invoke; the
# fallback exists only to keep the abstract contract satisfied.
@@ -198,8 +198,8 @@ class _StaticToolset(AbstractToolset[None]):
_ANY_VALIDATOR = SchemaValidator(schema=core_schema.any_schema())
def _build_function_toolset(*tools: Any) -> FunctionToolset[None]:
return FunctionToolset[None](tools=[Tool(t) for t in tools])
def _build_function_toolset(*tools: Any) -> FunctionToolset[object]:
return FunctionToolset[object](tools=[Tool(t) for t in tools])
# ---------------------------------------------------------------------------
@@ -222,7 +222,7 @@ class TestCodeMode:
async def test_default_wraps_all_tools_behind_run_code(self) -> None:
"""`CodeMode()` exposes only `run_code` and renders every tool as an `async def`."""
toolset = _build_function_toolset(add, greet)
wrapper = CodeMode[None]().get_wrapper_toolset(toolset)
wrapper = CodeMode[object]().get_wrapper_toolset(toolset)
assert isinstance(wrapper, CodeModeToolset)
tools = await wrapper.get_tools(build_run_context(None))
@@ -239,7 +239,7 @@ class TestCodeMode:
async def test_run_code_executes_call_through_monty(self) -> None:
"""End-to-end: `run_code` runs Python in Monty and dispatches to a sync wrapped tool."""
toolset = _build_function_toolset(add)
wrapper = CodeMode[None]().get_wrapper_toolset(toolset)
wrapper = CodeMode[object]().get_wrapper_toolset(toolset)
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -268,7 +268,7 @@ class TestCodeMode:
Exercises (a) string return values flowing back through the await/dispatch loop,
(b) default-argument handling -- the LLM-side code only passes `name`, not `greeting`.
"""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(greet))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(greet))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -282,7 +282,7 @@ class TestCodeMode:
async def test_run_code_can_chain_multiple_tool_calls_in_one_snippet(self) -> None:
"""A realistic LLM snippet that calls two tools in one `run_code` invocation."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add, greet))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add, greet))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -292,7 +292,7 @@ class TestCodeMode:
async def test_run_code_parallel_tool_calls_via_gather(self) -> None:
"""Concurrent tool calls via asyncio.gather work and record all nested metadata."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -313,7 +313,7 @@ class TestCodeMode:
"""Always fails."""
raise ModelRetry('not allowed')
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add, flaky))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add, flaky))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -332,7 +332,7 @@ class TestCodeMode:
"""Return a fake fixed timestamp."""
return '2026-04-08T12:00:00Z'
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(now_iso))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(now_iso))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -353,7 +353,7 @@ class TestCodeMode:
async def test_run_code_state_persists_between_calls(self) -> None:
"""REPL state must survive across consecutive `run_code` calls within a run."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -367,7 +367,7 @@ class TestCodeMode:
async def test_run_code_restart_resets_repl_state(self) -> None:
"""Passing `restart=True` clears any previously-set names in the sandbox."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -382,7 +382,7 @@ class TestCodeMode:
async def test_run_code_returns_last_expression_value(self) -> None:
"""When the last statement is an expression, its value is returned in `result`."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -392,7 +392,7 @@ class TestCodeMode:
async def test_run_code_syntax_error_becomes_model_retry(self) -> None:
"""A Python syntax error is surfaced as `ModelRetry` so the model can fix it."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -416,7 +416,7 @@ class TestCodeMode:
On a fresh REPL (first call or after restart), the code is type-checked
before execution using Monty's stateless type checker.
"""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -435,7 +435,7 @@ class TestCodeMode:
async def test_for_run_returns_fresh_instance_with_cleared_repl(self) -> None:
"""`for_run` must hand back a new toolset instance -- concurrent runs cannot share REPL state."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -451,7 +451,7 @@ class TestCodeMode:
async def test_for_run_step_short_circuits_when_wrapped_unchanged(self) -> None:
"""If the inner toolset doesn't change between steps, `for_run_step` returns `self` unchanged."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
same = await wrapper.for_run_step(ctx)
@@ -460,7 +460,7 @@ class TestCodeMode:
async def test_for_run_step_preserves_repl_when_wrapped_changes(self) -> None:
"""When the wrapped toolset changes between steps, REPL state must carry over to the new instance."""
class _SwappingToolset(AbstractToolset[None]):
class _SwappingToolset(AbstractToolset[object]):
"""Returns a *different* underlying toolset on each `for_run_step` call."""
def __init__(self) -> None:
@@ -471,19 +471,19 @@ class TestCodeMode:
def id(self) -> str | None:
return None # pragma: no cover - required by AbstractToolset, never read
async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
async def get_tools(self, ctx: RunContext[object]) -> dict[str, ToolsetTool[object]]:
return await self._inner.get_tools(ctx)
async def call_tool( # pragma: no cover - test only exercises lifecycle methods, not call_tool
self,
name: str,
tool_args: dict[str, Any],
ctx: RunContext[None],
tool: ToolsetTool[None],
ctx: RunContext[object],
tool: ToolsetTool[object],
) -> Any:
return await self._inner.call_tool(name, tool_args, ctx, tool)
async def for_run_step(self, ctx: RunContext[None]) -> AbstractToolset[None]:
async def for_run_step(self, ctx: RunContext[object]) -> AbstractToolset[object]:
# Return a brand-new toolset on every step so `is` comparison fails in
# `CodeModeToolset.for_run_step`, forcing the rebuild branch.
self._step += 1
@@ -491,7 +491,7 @@ class TestCodeMode:
new_self._step = self._step
return new_self
wrapper = CodeMode[None]().get_wrapper_toolset(_SwappingToolset())
wrapper = CodeMode[object]().get_wrapper_toolset(_SwappingToolset())
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -513,7 +513,7 @@ class TestCodeMode:
async def test_filter_keeps_rejected_tools_native(self) -> None:
"""A callable filter sandboxes accepted tools and leaves the rest visible to the model."""
capability = CodeMode[None](tools=lambda ctx, td: td.name == 'add')
capability = CodeMode[object](tools=lambda ctx, td: td.name == 'add')
wrapper = capability.get_wrapper_toolset(_build_function_toolset(add, greet))
assert isinstance(wrapper, CodeModeToolset)
@@ -528,7 +528,7 @@ class TestCodeMode:
async def test_native_tool_call_passes_through(self) -> None:
"""Calling a native (non-sandboxed) tool passes through to the wrapped toolset."""
capability = CodeMode[None](tools=lambda ctx, td: td.name == 'add')
capability = CodeMode[object](tools=lambda ctx, td: td.name == 'add')
wrapper = capability.get_wrapper_toolset(_build_function_toolset(add, greet))
assert isinstance(wrapper, CodeModeToolset)
@@ -545,7 +545,7 @@ class TestCodeMode:
"""A tool that collides with the reserved name."""
return 'oops' # pragma: no cover
capability = CodeMode[None](tools=lambda ctx, td: td.name != 'run_code')
capability = CodeMode[object](tools=lambda ctx, td: td.name != 'run_code')
wrapper = capability.get_wrapper_toolset(_build_function_toolset(run_code, add))
assert isinstance(wrapper, CodeModeToolset)
@@ -560,7 +560,7 @@ class TestCodeMode:
"""A tool that collides with the meta-tool name."""
return 'oops' # pragma: no cover
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(run_code, add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(run_code, add))
assert isinstance(wrapper, CodeModeToolset)
with pytest.raises(UserError, match='conflicts with the code mode'):
@@ -568,7 +568,7 @@ class TestCodeMode:
async def test_filter_excluding_everything_yields_run_code_with_no_functions(self) -> None:
"""A filter that rejects every tool produces a `run_code` with no functions block."""
capability = CodeMode[None](tools=lambda ctx, td: False)
capability = CodeMode[object](tools=lambda ctx, td: False)
wrapper = capability.get_wrapper_toolset(_build_function_toolset(add, greet))
assert isinstance(wrapper, CodeModeToolset)
@@ -583,11 +583,11 @@ class TestCodeMode:
"""The filter receives the live `RunContext` so it can vary per run/step."""
seen_steps: list[int] = []
def filter_func(ctx: RunContext[None], td: Any) -> bool:
def filter_func(ctx: RunContext[object], td: Any) -> bool:
seen_steps.append(ctx.run_step)
return td.name == 'add'
wrapper = CodeMode[None](tools=filter_func).get_wrapper_toolset(_build_function_toolset(add, greet))
wrapper = CodeMode[object](tools=filter_func).get_wrapper_toolset(_build_function_toolset(add, greet))
assert isinstance(wrapper, CodeModeToolset)
await wrapper.get_tools(build_run_context(None, run_step=7))
assert 7 in seen_steps
@@ -598,7 +598,7 @@ class TestCodeMode:
async def test_typed_dict_arguments_render_as_prelude(self) -> None:
"""Tools with structured (TypedDict) parameters render their types in the prelude."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(lookup_person))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(lookup_person))
assert isinstance(wrapper, CodeModeToolset)
description = (await wrapper.get_tools(build_run_context(None)))['run_code'].tool_def.description
@@ -618,7 +618,7 @@ class TestCodeMode:
that static type checking (which only runs on the first snippet) doesn't
reject the dict-to-TypedDict coercion that Monty handles at runtime.
"""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(lookup_person))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(lookup_person))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -641,7 +641,7 @@ class TestCodeMode:
results={'get_user': 'user-result', 'get_company': 'company-result'},
)
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -680,8 +680,8 @@ class TestCodeMode:
"""A deferred-loading tool."""
return str(x) # pragma: no cover - tool body is not invoked in this test
toolset = FunctionToolset[None](tools=[Tool(add), Tool(later, defer_loading=True)])
wrapper = CodeMode[None]().get_wrapper_toolset(toolset)
toolset = FunctionToolset[object](tools=[Tool(add), Tool(later, defer_loading=True)])
wrapper = CodeMode[object]().get_wrapper_toolset(toolset)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -706,8 +706,8 @@ class TestCodeMode:
return str(x) # pragma: no cover - tool body is not invoked in this test
# `defer_loading=False` mimics the post-discovery state ToolSearchToolset hands back.
toolset = FunctionToolset[None](tools=[Tool(add), Tool(later, defer_loading=False)])
wrapper = CodeMode[None]().get_wrapper_toolset(toolset)
toolset = FunctionToolset[object](tools=[Tool(add), Tool(later, defer_loading=False)])
wrapper = CodeMode[object]().get_wrapper_toolset(toolset)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -733,7 +733,7 @@ class TestCodeMode:
tool_kind='capability-load',
)
static = _StaticToolset([_make_address_tool_def('get_user', 'Get a user.', 'street'), td_loader])
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
tools = await wrapper.get_tools(build_run_context(None))
@@ -755,7 +755,7 @@ class TestCodeMode:
unless_native='web_search',
)
static = _StaticToolset([_make_address_tool_def('get_user', 'Get a user.', 'street'), td_fallback])
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -778,7 +778,7 @@ class TestCodeMode:
unless_native='web_search',
)
static = _StaticToolset([td_fallback])
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -798,7 +798,7 @@ class TestCodeMode:
return_schema={'type': 'string'},
)
static = _StaticToolset([td_plain])
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -820,7 +820,7 @@ class TestCodeMode:
kind='external',
)
static = _StaticToolset([_make_address_tool_def('get_user', 'Get a user.', 'street'), td_external])
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -842,7 +842,7 @@ class TestCodeMode:
# No return_schema -- simulates an MCP tool without outputSchema.
)
static = _StaticToolset([td], results={'search': 'found it'})
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -872,7 +872,7 @@ class TestCodeMode:
return_schema={'type': 'object', 'properties': {'name': {'type': 'string'}}},
)
static = _StaticToolset([td], results={'get_user': {'name': 'Alice'}})
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
with _warnings.catch_warnings():
@@ -924,7 +924,7 @@ class TestCodeMode:
observed_tool_returns.append(run_code_return.content)
return ModelResponse(parts=[TextPart(f'sum is {observed_tool_returns[-1]["result"]}')])
agent: Agent[None, str] = Agent(FunctionModel(model_fn), capabilities=[CodeMode[None]()])
agent: Agent[object, str] = Agent(FunctionModel(model_fn), capabilities=[CodeMode[object]()])
@agent.tool_plain
def add(a: int, b: int) -> int: # pyright: ignore[reportUnusedFunction]
@@ -962,7 +962,7 @@ class TestCodeMode:
"""
from pydantic_ai.capabilities import Capability
capability = Capability[None](
capability = Capability[object](
id='demo',
description='Demo deferred capability.',
instructions='Use demo_tool.',
@@ -974,9 +974,9 @@ class TestCodeMode:
return 'ok' # pragma: no cover - deferred tool stays hidden, body is not invoked
model = TestModel(call_tools=[])
agent: Agent[None, str] = Agent(
agent: Agent[object, str] = Agent(
model,
capabilities=[capability, CodeMode[None](tools='all')],
capabilities=[capability, CodeMode[object](tools='all')],
)
await agent.run('inspect tools')
@@ -1000,7 +1000,7 @@ class TestCodeMode:
async def test_code_mode_can_be_registered_as_agent_capability(self) -> None:
"""`CodeMode` can be passed via `Agent(capabilities=[...])` without raising."""
Agent(TestModel(), capabilities=[CodeMode[None]()])
Agent(TestModel(), capabilities=[CodeMode[object]()])
# ---------------------------------------------------------------------------
# Tool name sanitization
@@ -1036,7 +1036,7 @@ class TestCodeMode:
return_schema={'type': 'string'},
)
static = _StaticToolset([td], results={'get-weather': 'sunny'})
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -1070,7 +1070,7 @@ class TestCodeMode:
return_schema={'type': 'string'},
)
static = _StaticToolset([td], results={'api.lookup': 'found'})
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -1098,7 +1098,7 @@ class TestCodeMode:
return_schema={'type': 'string'},
)
static = _StaticToolset([td1, td2], results={'get-weather': 'rain'})
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -1126,7 +1126,7 @@ class TestCodeMode:
return_schema={'type': 'string'},
)
static = _StaticToolset([td_native, td_hyphen], results={'get_weather': 'ok'})
wrapper = CodeMode[None]().get_wrapper_toolset(static)
wrapper = CodeMode[object]().get_wrapper_toolset(static)
assert isinstance(wrapper, CodeModeToolset)
ctx = build_run_context(None)
@@ -1144,7 +1144,7 @@ class TestCodeMode:
async def test_run_code_tool_has_code_metadata(self) -> None:
"""The `run_code` ToolDefinition carries metadata for Logfire code rendering."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
tools = await wrapper.get_tools(build_run_context(None))
@@ -1161,7 +1161,7 @@ class TestCodeMode:
"""Return a ToolReturn with metadata."""
return ToolReturnMsg(return_value=42, metadata={'source': 'test'})
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(fancy))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(fancy))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1183,7 +1183,7 @@ class TestCodeMode:
"""A tool that requires approval."""
raise _ApprovalRequired()
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(needs_approval))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(needs_approval))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1211,12 +1211,12 @@ class TestCodeMode:
"""A tool that requires approval."""
raise _ApprovalRequired()
async def handler(ctx: RunContext[None], requests: DeferredToolRequests) -> DeferredToolResults:
async def handler(ctx: RunContext[object], requests: DeferredToolRequests) -> DeferredToolResults:
return DeferredToolResults(
approvals={call.tool_call_id: ToolDenied(message='nope') for call in requests.approvals}
)
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(needs_approval))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(needs_approval))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper, root_capability=HandleDeferredToolCalls(handler=handler))
tools = await wrapper.get_tools(ctx)
@@ -1239,14 +1239,14 @@ class TestCodeMode:
from pydantic_ai.exceptions import ApprovalRequired as _ApprovalRequired
from pydantic_ai.tools import DeferredToolRequests, DeferredToolResults, ToolApproved
def always_needs_approval(ctx: RunContext[None]) -> str:
def always_needs_approval(ctx: RunContext[object]) -> str:
"""Raises `ApprovalRequired` every time, even after being approved."""
raise _ApprovalRequired()
async def handler(ctx: RunContext[None], requests: DeferredToolRequests) -> DeferredToolResults:
async def handler(ctx: RunContext[object], requests: DeferredToolRequests) -> DeferredToolResults:
return DeferredToolResults(approvals={call.tool_call_id: ToolApproved() for call in requests.approvals})
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(always_needs_approval))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(always_needs_approval))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper, root_capability=HandleDeferredToolCalls(handler=handler))
tools = await wrapper.get_tools(ctx)
@@ -1265,7 +1265,7 @@ class TestCodeMode:
"""A tool that always retries."""
raise ModelRetry('try again please')
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(flaky))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(flaky))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1278,7 +1278,7 @@ class TestCodeMode:
On a fresh REPL, the static type checker catches this before execution.
"""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1307,7 +1307,7 @@ class TestCodeMode:
"""Generate an image."""
return BinaryContent(data=image_bytes, media_type='image/png')
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(gen_image))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(gen_image))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1330,7 +1330,7 @@ class TestCodeMode:
"""Generate an image."""
return BinaryContent(data=image_bytes, media_type='image/png')
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(gen_image))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(gen_image))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1359,7 +1359,7 @@ class TestCodeMode:
"""Generate a list with an image."""
return [BinaryContent(data=image_bytes, media_type='image/png'), 'caption']
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(gen_images))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(gen_images))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1392,7 +1392,7 @@ class TestCodeMode:
return_value=BinaryContent(data=image_bytes, media_type='image/png'), metadata={'src': 'test'}
)
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(gen_image))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(gen_image))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1431,9 +1431,9 @@ class TestCodeMode:
return ModelResponse(parts=[ToolCallPart(tool_name='run_code', args={'code': 'await add(a=1, b=2)'})])
return ModelResponse(parts=[TextPart('done')])
agent: Agent[None, str] = Agent(
agent: Agent[object, str] = Agent(
FunctionModel(model_fn),
capabilities=[CodeMode[None](), Instrumentation(settings=InstrumentationSettings(include_content=True))],
capabilities=[CodeMode[object](), Instrumentation(settings=InstrumentationSettings(include_content=True))],
)
@agent.tool_plain
@@ -1469,7 +1469,7 @@ class TestCodeMode:
On a fresh REPL, the type checker catches this; on subsequent calls,
it becomes a runtime NameError.
"""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1490,7 +1490,7 @@ class TestCodeMode:
On a fresh REPL the type checker catches it; on subsequent calls
the runtime positional-args guard catches it.
"""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1517,7 +1517,7 @@ class TestCodeMode:
async def test_print_output_preserved_in_runtime_error(self) -> None:
"""When sandbox code prints before crashing, the print output is included
in the ModelRetry error message so the model can use it for debugging."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1542,7 +1542,7 @@ class TestCodeMode:
resolved inline at FunctionSnapshot via `resume({'return_value': ...})`."""
from dataclasses import replace as dc_replace
class _SeqToolset(AbstractToolset[None]):
class _SeqToolset(AbstractToolset[object]):
"""Marks add as sequential; greet stays parallel."""
def __init__(self) -> None:
@@ -1552,7 +1552,7 @@ class TestCodeMode:
def id(self) -> str | None:
return None # pragma: no cover
async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
async def get_tools(self, ctx: RunContext[object]) -> dict[str, ToolsetTool[object]]:
tools = await self._inner.get_tools(ctx)
return {
n: dc_replace(t, tool_def=dc_replace(t.tool_def, sequential=True)) if n == 'add' else t
@@ -1560,11 +1560,11 @@ class TestCodeMode:
}
async def call_tool(
self, name: str, tool_args: dict[str, Any], ctx: RunContext[None], tool: ToolsetTool[None]
self, name: str, tool_args: dict[str, Any], ctx: RunContext[object], tool: ToolsetTool[object]
) -> Any:
return await self._inner.call_tool(name, tool_args, ctx, tool)
seq_wrapper = CodeModeToolset[None](wrapped=_SeqToolset(), tool_selector='all')
seq_wrapper = CodeModeToolset[object](wrapped=_SeqToolset(), tool_selector='all')
ctx = await build_ctx(None, seq_wrapper)
tools = await seq_wrapper.get_tools(ctx)
run_code = tools['run_code']
@@ -1603,7 +1603,7 @@ class TestCodeMode:
the pending tasks are awaited first (barrier) before dispatching."""
from dataclasses import replace as dc_replace
class _SeqToolset(AbstractToolset[None]):
class _SeqToolset(AbstractToolset[object]):
def __init__(self) -> None:
self._inner = _build_function_toolset(add, greet)
@@ -1611,7 +1611,7 @@ class TestCodeMode:
def id(self) -> str | None:
return None # pragma: no cover
async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
async def get_tools(self, ctx: RunContext[object]) -> dict[str, ToolsetTool[object]]:
tools = await self._inner.get_tools(ctx)
return {
n: dc_replace(t, tool_def=dc_replace(t.tool_def, sequential=True)) if n == 'add' else t
@@ -1619,11 +1619,11 @@ class TestCodeMode:
}
async def call_tool(
self, name: str, tool_args: dict[str, Any], ctx: RunContext[None], tool: ToolsetTool[None]
self, name: str, tool_args: dict[str, Any], ctx: RunContext[object], tool: ToolsetTool[object]
) -> Any:
return await self._inner.call_tool(name, tool_args, ctx, tool)
seq_wrapper = CodeModeToolset[None](wrapped=_SeqToolset(), tool_selector='all')
seq_wrapper = CodeModeToolset[object](wrapped=_SeqToolset(), tool_selector='all')
ctx = await build_ctx(None, seq_wrapper)
tools = await seq_wrapper.get_tools(ctx)
@@ -1661,7 +1661,7 @@ class TestCodeMode:
"""An error from a sequential tool (resolved inline) surfaces as ModelRetry."""
from dataclasses import replace as dc_replace
class _SeqToolset(AbstractToolset[None]):
class _SeqToolset(AbstractToolset[object]):
def __init__(self) -> None:
self._inner = _build_function_toolset(add)
@@ -1669,16 +1669,16 @@ class TestCodeMode:
def id(self) -> str | None:
return None # pragma: no cover
async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
async def get_tools(self, ctx: RunContext[object]) -> dict[str, ToolsetTool[object]]:
tools = await self._inner.get_tools(ctx)
return {n: dc_replace(t, tool_def=dc_replace(t.tool_def, sequential=True)) for n, t in tools.items()}
async def call_tool(
self, name: str, tool_args: dict[str, Any], ctx: RunContext[None], tool: ToolsetTool[None]
self, name: str, tool_args: dict[str, Any], ctx: RunContext[object], tool: ToolsetTool[object]
) -> Any:
return await self._inner.call_tool(name, tool_args, ctx, tool)
seq_wrapper = CodeModeToolset[None](wrapped=_SeqToolset(), tool_selector='all')
seq_wrapper = CodeModeToolset[object](wrapped=_SeqToolset(), tool_selector='all')
ctx = await build_ctx(None, seq_wrapper)
tools = await seq_wrapper.get_tools(ctx)
run_code = tools['run_code']
@@ -1693,7 +1693,7 @@ class TestCodeMode:
sandbox are resolved sequentially via FutureSnapshot. Signatures stay `async def`."""
from pydantic_ai.tool_manager import ToolManager
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
@@ -1721,7 +1721,7 @@ class TestCodeMode:
from pydantic_ai.tool_manager import ToolManager
class _SeqToolset(AbstractToolset[None]):
class _SeqToolset(AbstractToolset[object]):
def __init__(self) -> None:
self._inner = _build_function_toolset(add)
@@ -1729,16 +1729,16 @@ class TestCodeMode:
def id(self) -> str | None:
return None # pragma: no cover
async def get_tools(self, ctx: RunContext[None]) -> dict[str, ToolsetTool[None]]:
async def get_tools(self, ctx: RunContext[object]) -> dict[str, ToolsetTool[object]]:
tools = await self._inner.get_tools(ctx)
return {n: dc_replace(t, tool_def=dc_replace(t.tool_def, sequential=True)) for n, t in tools.items()}
async def call_tool(
self, name: str, tool_args: dict[str, Any], ctx: RunContext[None], tool: ToolsetTool[None]
self, name: str, tool_args: dict[str, Any], ctx: RunContext[object], tool: ToolsetTool[object]
) -> Any:
return await self._inner.call_tool(name, tool_args, ctx, tool)
seq_wrapper = CodeModeToolset[None](wrapped=_SeqToolset(), tool_selector='all')
seq_wrapper = CodeModeToolset[object](wrapped=_SeqToolset(), tool_selector='all')
ctx = await build_ctx(None, seq_wrapper)
with ToolManager.parallel_execution_mode('sequential'):
@@ -1757,7 +1757,7 @@ class TestCodeMode:
async def test_restart_with_invalid_code_clears_repl_for_retry(self) -> None:
"""When `restart=True` and type checking fails, the REPL is cleared so
the next retry still gets type-checked on a fresh REPL."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -1868,7 +1868,7 @@ class TestToolSearchIntegration:
"""A deferred-loading tool."""
return str(x) # pragma: no cover - tool body is not invoked in this test
base = FunctionToolset[None](tools=[Tool(add), Tool(later, defer_loading=True)])
base = FunctionToolset[object](tools=[Tool(add), Tool(later, defer_loading=True)])
code_mode = CodeModeToolset(wrapped=ToolSearchToolset(wrapped=base), tool_selector='all')
tools = await code_mode.get_tools(build_run_context(None))
@@ -1893,7 +1893,7 @@ class TestToolSearchIntegration:
"""A deferred-loading tool."""
return str(x) # pragma: no cover - tool body is not invoked in this test
base = FunctionToolset[None](tools=[Tool(add), Tool(later, defer_loading=True)])
base = FunctionToolset[object](tools=[Tool(add), Tool(later, defer_loading=True)])
code_mode = CodeModeToolset(wrapped=ToolSearchToolset(wrapped=base), tool_selector='all')
messages: list[ModelMessage] = [
@@ -1906,7 +1906,7 @@ class TestToolSearchIntegration:
]
)
]
ctx = RunContext[None](
ctx = RunContext[object](
deps=None,
model=TestModel(),
usage=RunUsage(),
@@ -1979,8 +1979,8 @@ class TestDynamicCatalog:
async def test_get_instructions_appends_to_upstream_string(self) -> None:
from pydantic_ai.messages import InstructionPart
class _UpstreamToolset(FunctionToolset[None]):
async def get_instructions(self, ctx: RunContext[None]) -> str: # pyright: ignore[reportIncompatibleMethodOverride]
class _UpstreamToolset(FunctionToolset[object]):
async def get_instructions(self, ctx: RunContext[object]) -> str: # pyright: ignore[reportIncompatibleMethodOverride]
return 'wrapped instructions'
toolset = CodeModeToolset(
@@ -1998,9 +1998,9 @@ class TestDynamicCatalog:
async def test_get_instructions_appends_to_upstream_sequence(self) -> None:
from pydantic_ai.messages import InstructionPart
class _UpstreamToolset(FunctionToolset[None]):
class _UpstreamToolset(FunctionToolset[object]):
async def get_instructions( # pyright: ignore[reportIncompatibleMethodOverride]
self, ctx: RunContext[None]
self, ctx: RunContext[object]
) -> list[str | InstructionPart]:
return ['a', InstructionPart(content='b')]
@@ -2055,8 +2055,8 @@ class TestDynamicCatalog:
async def test_for_run_step_preserves_catalog_stash(self) -> None:
"""A per-step rebuild must carry `_last_catalog` so instructions stay populated."""
class _ChangingToolset(FunctionToolset[None]):
async def for_run_step(self, ctx: RunContext[None]) -> AbstractToolset[None]:
class _ChangingToolset(FunctionToolset[object]):
async def for_run_step(self, ctx: RunContext[object]) -> AbstractToolset[object]:
# Force `CodeModeToolset.for_run_step` down the `new_wrapped is not self.wrapped`
# branch by returning a distinct (but equivalent) wrapped instance.
return type(self)(tools=list(self.tools.values()))
@@ -2077,14 +2077,14 @@ class TestDynamicCatalog:
# -- capability per-run state -----------------------------------------
async def test_for_run_returns_fresh_state_when_enabled(self) -> None:
cap = CodeMode[None](dynamic_catalog=True)
cap = CodeMode[object](dynamic_catalog=True)
cap._announced_tools.add('foo') # pyright: ignore[reportPrivateUsage]
fresh = await cap.for_run(build_run_context(None))
assert fresh is not cap
assert fresh._announced_tools == set() # pyright: ignore[reportPrivateUsage]
async def test_for_run_returns_self_when_disabled(self) -> None:
cap = CodeMode[None]()
cap = CodeMode[object]()
assert await cap.for_run(build_run_context(None)) is cap
# -- discovery announcement: local search path ------------------------
@@ -2092,7 +2092,7 @@ class TestDynamicCatalog:
async def test_announce_on_local_search_return(self) -> None:
from pydantic_ai.messages import ModelRequest, SystemPromptPart, ToolCallPart
cap = CodeMode[None](dynamic_catalog=True)
cap = CodeMode[object](dynamic_catalog=True)
ctx = build_run_context(None)
await cap.after_tool_execute(
ctx,
@@ -2114,7 +2114,7 @@ class TestDynamicCatalog:
"""With `dynamic_catalog=False`, the hooks are inert even on a real search return."""
from pydantic_ai.messages import ToolCallPart
cap = CodeMode[None]()
cap = CodeMode[object]()
ctx = build_run_context(None)
await cap.after_tool_execute(
ctx,
@@ -2128,7 +2128,7 @@ class TestDynamicCatalog:
async def test_announce_skipped_when_no_discoveries(self) -> None:
from pydantic_ai.messages import ToolCallPart
cap = CodeMode[None](dynamic_catalog=True)
cap = CodeMode[object](dynamic_catalog=True)
ctx = build_run_context(None)
await cap.after_tool_execute(
ctx,
@@ -2143,7 +2143,7 @@ class TestDynamicCatalog:
"""`tool_kind != 'tool-search'` short-circuits before reading the result."""
from pydantic_ai.messages import ToolCallPart
cap = CodeMode[None](dynamic_catalog=True)
cap = CodeMode[object](dynamic_catalog=True)
ctx = build_run_context(None)
await cap.after_tool_execute(
ctx,
@@ -2159,7 +2159,7 @@ class TestDynamicCatalog:
async def test_no_duplicate_announcement_for_same_tool(self) -> None:
from pydantic_ai.messages import ToolCallPart
cap = CodeMode[None](dynamic_catalog=True)
cap = CodeMode[object](dynamic_catalog=True)
ctx = build_run_context(None)
result = {'discovered_tools': [{'name': 'weather'}]}
for cid in ('c1', 'c2'):
@@ -2180,7 +2180,7 @@ class TestDynamicCatalog:
from pydantic_ai.messages import ModelRequest, ModelResponse, NativeToolSearchReturnPart, SystemPromptPart
from pydantic_ai.usage import RequestUsage
cap = CodeMode[None](dynamic_catalog=True)
cap = CodeMode[object](dynamic_catalog=True)
ctx = build_run_context(None)
response = ModelResponse(
parts=[
@@ -2205,7 +2205,7 @@ class TestDynamicCatalog:
from pydantic_ai.messages import ModelResponse, NativeToolReturnPart, TextPart
from pydantic_ai.usage import RequestUsage
cap = CodeMode[None](dynamic_catalog=True)
cap = CodeMode[object](dynamic_catalog=True)
ctx = build_run_context(None)
response = ModelResponse(
parts=[
@@ -2294,10 +2294,10 @@ class TestDynamicCatalog:
"""Get the weather."""
return f'sunny in {city}' # pragma: no cover — only the signature matters.
agent: Agent[None, str] = Agent(
agent: Agent[object, str] = Agent(
FunctionModel(model_fn),
tools=[Tool(weather, defer_loading=True)],
capabilities=[ToolSearch[None](), CodeMode[None](dynamic_catalog=True)],
capabilities=[ToolSearch[object](), CodeMode[object](dynamic_catalog=True)],
)
result = await agent.run('please find a weather tool')
@@ -2356,10 +2356,10 @@ class TestDynamicCatalog:
usage=RequestUsage(input_tokens=1, output_tokens=1),
)
agent: Agent[None, str] = Agent(
agent: Agent[object, str] = Agent(
FunctionModel(model_fn),
tools=[Tool(add)],
capabilities=[CodeMode[None](dynamic_catalog=True)],
capabilities=[CodeMode[object](dynamic_catalog=True)],
)
result = await agent.run('add 3 and 4 via run_code')
assert result.output == 'got 7'
@@ -2376,7 +2376,7 @@ class TestCodeModeOSAccess:
async def test_description_default_notes_no_fs_env_or_clock(self) -> None:
"""Without `os`/`mount`, the description states filesystem, env, and clock calls are
unavailable, so the model does not waste retries calling `pathlib`/`os` I/O."""
wrapper = CodeMode[None]().get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object]().get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
description = (await wrapper.get_tools(build_run_context(None)))['run_code'].tool_def.description
assert description is not None
@@ -2385,7 +2385,7 @@ class TestCodeModeOSAccess:
async def test_description_with_os_callback_notes_host_access(self) -> None:
"""An `os` callback swaps the restriction line for the host-access note."""
wrapper = CodeMode[None](os_access=_unused_os_callback).get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object](os_access=_unused_os_callback).get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
description = (await wrapper.get_tools(build_run_context(None)))['run_code'].tool_def.description
assert description is not None
@@ -2394,7 +2394,7 @@ class TestCodeModeOSAccess:
async def test_description_mount_only_advertises_filesystem_not_env_or_clock(self, tmp_path: Path) -> None:
"""A `mount` without `os` advertises filesystem access only -- it must not tell the model
that env/clock are host-backed, since a mount cannot route `os.getenv`/`datetime.now()`."""
wrapper = CodeMode[None](mount=MountDir('/work', str(tmp_path))).get_wrapper_toolset(
wrapper = CodeMode[object](mount=MountDir('/work', str(tmp_path))).get_wrapper_toolset(
_build_function_toolset(add)
)
assert isinstance(wrapper, CodeModeToolset)
@@ -2407,7 +2407,7 @@ class TestCodeModeOSAccess:
async def test_description_host_access_note_shows_with_no_sandboxed_tools(self) -> None:
"""The host-access note appears even when no tools are sandboxed (base description)."""
# `tools=[]` sandboxes nothing, so `run_code` renders the base description path.
wrapper = CodeMode[None](os_access=_unused_os_callback, tools=[]).get_wrapper_toolset(
wrapper = CodeMode[object](os_access=_unused_os_callback, tools=[]).get_wrapper_toolset(
_build_function_toolset(add)
)
assert isinstance(wrapper, CodeModeToolset)
@@ -2424,7 +2424,7 @@ class TestCodeModeOSAccess:
return 'envval'
return NOT_HANDLED # pragma: no cover - sandbox only calls os.getenv here
wrapper = CodeMode[None](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -2443,7 +2443,7 @@ class TestCodeModeOSAccess:
return 'persisted'
return NOT_HANDLED # pragma: no cover - sandbox only calls os.getenv here
wrapper = CodeMode[None](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -2455,7 +2455,7 @@ class TestCodeModeOSAccess:
async def test_abstract_os_instance_dispatches_inside_run_code(self) -> None:
"""An `AbstractOS` instance is accepted as the `os` value and dispatches OS calls."""
wrapper = CodeMode[None](os_access=OSAccess(environ={'THING': 'fromabs'})).get_wrapper_toolset(
wrapper = CodeMode[object](os_access=OSAccess(environ={'THING': 'fromabs'})).get_wrapper_toolset(
_build_function_toolset(add)
)
assert isinstance(wrapper, CodeModeToolset)
@@ -2471,7 +2471,7 @@ class TestCodeModeOSAccess:
def os_cb(fn: OsFunction, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any:
raise ValueError('boom from os')
wrapper = CodeMode[None](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -2492,7 +2492,7 @@ class TestCodeModeOSAccess:
return allowed.get(args[0])
return NOT_HANDLED # pragma: no cover - sandbox only calls os.getenv here
wrapper = CodeMode[None](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -2511,7 +2511,7 @@ class TestCodeModeOSAccess:
def os_cb(fn: OsFunction, args: tuple[Any, ...], kwargs: dict[str, Any]) -> Any:
return NOT_HANDLED
wrapper = CodeMode[None](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object](os_access=os_cb).get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -2521,7 +2521,7 @@ class TestCodeModeOSAccess:
async def test_mount_exposes_host_directory(self, tmp_path: Path) -> None:
"""A `mount` exposes a host directory inside the sandbox, threaded through resumes."""
(tmp_path / 'data.txt').write_text('hello-from-host')
wrapper = CodeMode[None](mount=MountDir('/work', str(tmp_path))).get_wrapper_toolset(
wrapper = CodeMode[object](mount=MountDir('/work', str(tmp_path))).get_wrapper_toolset(
_build_function_toolset(add)
)
assert isinstance(wrapper, CodeModeToolset)
@@ -2538,7 +2538,7 @@ class TestCodeModeOSAccess:
(tmp_path / 'a' / 'f.txt').write_text('AA')
(tmp_path / 'b' / 'f.txt').write_text('BB')
mounts = [MountDir('/a', str(tmp_path / 'a')), MountDir('/b', str(tmp_path / 'b'))]
wrapper = CodeMode[None](mount=mounts).get_wrapper_toolset(_build_function_toolset(add))
wrapper = CodeMode[object](mount=mounts).get_wrapper_toolset(_build_function_toolset(add))
assert isinstance(wrapper, CodeModeToolset)
ctx = await build_ctx(None, wrapper)
tools = await wrapper.get_tools(ctx)
@@ -2549,7 +2549,7 @@ class TestCodeModeOSAccess:
def test_capability_forwards_os_and_mount_to_toolset(self, tmp_path: Path) -> None:
"""`CodeMode` forwards `os_access`/`mount` onto the `CodeModeToolset` it builds."""
mount = MountDir('/work', str(tmp_path))
wrapper = CodeMode[None](os_access=_unused_os_callback, mount=mount).get_wrapper_toolset(
wrapper = CodeMode[object](os_access=_unused_os_callback, mount=mount).get_wrapper_toolset(
_build_function_toolset(add)
)
assert isinstance(wrapper, CodeModeToolset)
tests/experimental/authoring/test_authoring.py
+5 -5
View File
@@ -439,24 +439,24 @@ class TestAuthoringToolset:
class TestRuntimeAuthoringCapability:
def test_get_instructions_default(self, tmp_path: Path) -> None:
instructions = RuntimeAuthoring[None](directory=tmp_path).get_instructions()
instructions = RuntimeAuthoring[object](directory=tmp_path).get_instructions()
assert isinstance(instructions, str)
assert 'author_capability' in instructions
def test_get_instructions_custom(self, tmp_path: Path) -> None:
assert RuntimeAuthoring[None](directory=tmp_path, guidance='X').get_instructions() == 'X'
assert RuntimeAuthoring[object](directory=tmp_path, guidance='X').get_instructions() == 'X'
def test_get_instructions_empty_omitted(self, tmp_path: Path) -> None:
assert RuntimeAuthoring[None](directory=tmp_path, guidance='').get_instructions() is None
assert RuntimeAuthoring[object](directory=tmp_path, guidance='').get_instructions() is None
def test_get_toolset_type(self, tmp_path: Path) -> None:
assert isinstance(RuntimeAuthoring[None](directory=tmp_path).get_toolset(), AuthoringToolset)
assert isinstance(RuntimeAuthoring[object](directory=tmp_path).get_toolset(), AuthoringToolset)
def test_serialization_name_none(self) -> None:
assert RuntimeAuthoring.get_serialization_name() is None
def test_store_property(self, tmp_path: Path) -> None:
store = RuntimeAuthoring[None](directory=tmp_path).store
store = RuntimeAuthoring[object](directory=tmp_path).store
assert isinstance(store, CapabilityStore)
assert store.directory == tmp_path
tests/experimental/context/test_context.py
+23 -22
View File
@@ -33,8 +33,8 @@ def anyio_backend() -> str:
return 'asyncio'
def _run_context() -> RunContext[None]:
return RunContext[None](
def _run_context() -> RunContext[object]:
return RunContext[object](
deps=None,
model=TestModel(),
usage=RunUsage(),
@@ -141,31 +141,31 @@ class TestRender:
class TestInstructions:
def test_includes_files_and_inventory_hint(self, tmp_path: Path) -> None:
_write(tmp_path / 'CLAUDE.md', 'be nice')
cap = RepoContext[None](workspace_dir=tmp_path)
cap = RepoContext[object](workspace_dir=tmp_path)
instructions = cap.get_instructions()
assert isinstance(instructions, str)
assert 'be nice' in instructions
assert 'inventory_agent_context' in instructions
def test_none_when_all_disabled(self, tmp_path: Path) -> None:
cap = RepoContext[None](workspace_dir=tmp_path, autoload_instructions=False, expose_inventory_tool=False)
cap = RepoContext[object](workspace_dir=tmp_path, autoload_instructions=False, expose_inventory_tool=False)
assert cap.get_instructions() is None
def test_autoload_off_keeps_inventory_hint(self, tmp_path: Path) -> None:
_write(tmp_path / 'CLAUDE.md', 'ignored')
cap = RepoContext[None](workspace_dir=tmp_path, autoload_instructions=False)
cap = RepoContext[object](workspace_dir=tmp_path, autoload_instructions=False)
instructions = cap.get_instructions()
assert isinstance(instructions, str)
assert 'ignored' not in instructions
assert 'inventory_agent_context' in instructions
def test_no_files_no_inventory_is_none(self, tmp_path: Path) -> None:
cap = RepoContext[None](workspace_dir=tmp_path, expose_inventory_tool=False)
cap = RepoContext[object](workspace_dir=tmp_path, expose_inventory_tool=False)
assert cap.get_instructions() is None
def test_files_cached_across_calls(self, tmp_path: Path) -> None:
_write(tmp_path / 'CLAUDE.md', 'first')
cap = RepoContext[None](workspace_dir=tmp_path)
cap = RepoContext[object](workspace_dir=tmp_path)
assert cap.get_instructions() is not None and 'first' in cap.get_instructions() # type: ignore[operator]
_write(tmp_path / 'CLAUDE.md', 'second')
# Read-once: the cached result is reused, so the edit is not picked up.
@@ -174,15 +174,16 @@ class TestInstructions:
class TestToolset:
def test_get_toolset_none_when_disabled(self, tmp_path: Path) -> None:
assert RepoContext[None](workspace_dir=tmp_path, expose_inventory_tool=False).get_toolset() is None
assert RepoContext[object](workspace_dir=tmp_path, expose_inventory_tool=False).get_toolset() is None
def test_get_toolset_present(self, tmp_path: Path) -> None:
assert isinstance(RepoContext[None](workspace_dir=tmp_path).get_toolset(), RepoContextToolset)
assert isinstance(RepoContext[object](workspace_dir=tmp_path).get_toolset(), RepoContextToolset)
async def test_inventory_tool_runs_through_agent(self, tmp_path: Path) -> None:
_write(tmp_path / '.claude' / 'skills' / 'foo' / 'SKILL.md', 'skill')
agent = Agent(
TestModel(call_tools=['inventory_agent_context']), capabilities=[RepoContext[None](workspace_dir=tmp_path)]
TestModel(call_tools=['inventory_agent_context']),
capabilities=[RepoContext[object](workspace_dir=tmp_path)],
)
result = await agent.run('go')
assert 'inventory_agent_context' in result.output
@@ -230,14 +231,14 @@ class TestScanAssets:
class TestNestedTraversal:
async def test_off_by_default_returns_result(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
cap = RepoContext[None](workspace_dir=tmp_path)
cap = RepoContext[object](workspace_dir=tmp_path)
call, tool_def, args = _call('list_directory', path='sub')
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='listing')
assert out == 'listing'
async def test_pointer_appended_on_first_traversal(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
call, tool_def, args = _call('list_directory', path='sub')
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='listing')
assert out.startswith('listing')
@@ -246,7 +247,7 @@ class TestNestedTraversal:
async def test_second_traversal_no_reappend(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
call, tool_def, args = _call('list_directory', path='sub')
ctx = _run_context()
first = await cap.after_tool_execute(ctx, call=call, tool_def=tool_def, args=args, result='one')
@@ -256,13 +257,13 @@ class TestNestedTraversal:
async def test_tool_name_not_matched(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
call, tool_def, args = _call('write_file', path='sub')
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='r')
assert out == 'r'
async def test_non_str_path_arg_ignored(self, tmp_path: Path) -> None:
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
call = ToolCallPart(tool_name='list_directory', args={'path': 123})
out = await cap.after_tool_execute(
_run_context(), call=call, tool_def=ToolDefinition(name='list_directory'), args={'path': 123}, result='r'
@@ -271,7 +272,7 @@ class TestNestedTraversal:
async def test_dir_without_context_file_untouched(self, tmp_path: Path) -> None:
(tmp_path / 'sub').mkdir()
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
call, tool_def, args = _call('list_directory', path='sub')
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='r')
assert out == 'r'
@@ -279,14 +280,14 @@ class TestNestedTraversal:
async def test_read_file_uses_parent_dir(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
target = _write(tmp_path / 'sub' / 'code.py', 'x = 1')
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
call, tool_def, args = _call('read_file', path=str(target))
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='file body')
assert 'CLAUDE.md' in out
async def test_contents_mode_inlines_body(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'NESTED BODY')
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True, nested_inject='contents')
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True, nested_inject='contents')
call, tool_def, args = _call('list_directory', path='sub')
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='r')
assert 'NESTED BODY' in out
@@ -295,14 +296,14 @@ class TestNestedTraversal:
workspace = tmp_path / 'ws'
workspace.mkdir()
outside = _write(tmp_path / 'outside' / 'CLAUDE.md', 'nested').parent
cap = RepoContext[None](workspace_dir=workspace, nested_traversal=True)
cap = RepoContext[object](workspace_dir=workspace, nested_traversal=True)
call, tool_def, args = _call('list_directory', path=str(outside))
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='r')
assert outside.resolve().as_posix() in out
async def test_non_str_result_returned_unchanged(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
cap = RepoContext[None](
cap = RepoContext[object](
workspace_dir=tmp_path, nested_traversal=True, traversal_tool_names=frozenset({'list_dir', 'read_file'})
)
call, tool_def, args = _call('list_dir', path='sub')
@@ -312,7 +313,7 @@ class TestNestedTraversal:
async def test_string_result_still_gets_note(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
cap = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
cap = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
call, tool_def, args = _call('list_directory', path='sub')
out = await cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='listing')
assert out.startswith('listing')
@@ -322,7 +323,7 @@ class TestNestedTraversal:
class TestForRunAndMisc:
async def test_for_run_isolates_state(self, tmp_path: Path) -> None:
_write(tmp_path / 'sub' / 'CLAUDE.md', 'nested')
base = RepoContext[None](workspace_dir=tmp_path, nested_traversal=True)
base = RepoContext[object](workspace_dir=tmp_path, nested_traversal=True)
run_cap = await base.for_run(_run_context())
call, tool_def, args = _call('list_directory', path='sub')
await run_cap.after_tool_execute(_run_context(), call=call, tool_def=tool_def, args=args, result='r')
tests/experimental/docs/test_docs.py
+13 -13
View File
@@ -63,7 +63,7 @@ class TestPyaiDocsToolset:
async def test_local_hit_is_cached(self, tmp_path: Path) -> None:
(tmp_path / 'hooks.md').write_text('# Hooks local', encoding='utf-8')
cache: dict[PyaiDocsTopic, str] = {}
toolset = PyaiDocsToolset[None](local_docs_path=tmp_path, cache=cache)
toolset = PyaiDocsToolset[object](local_docs_path=tmp_path, cache=cache)
assert await toolset.read_pyai_docs(PyaiDocsTopic.hooks) == '# Hooks local'
assert cache[PyaiDocsTopic.hooks] == '# Hooks local'
@@ -74,7 +74,7 @@ class TestPyaiDocsToolset:
async def test_remote_fallback_without_local_and_caching_disabled(self, monkeypatch: pytest.MonkeyPatch) -> None:
_install_fake_httpx(monkeypatch, text='# Capabilities remote')
toolset = PyaiDocsToolset[None](local_docs_path=None, cache=None)
toolset = PyaiDocsToolset[object](local_docs_path=None, cache=None)
assert await toolset.read_pyai_docs(PyaiDocsTopic.capabilities) == '# Capabilities remote'
@@ -83,21 +83,21 @@ class TestPyaiDocsToolset:
) -> None:
_install_fake_httpx(monkeypatch, text='# Agent remote')
cache: dict[PyaiDocsTopic, str] = {}
toolset = PyaiDocsToolset[None](local_docs_path=tmp_path, cache=cache)
toolset = PyaiDocsToolset[object](local_docs_path=tmp_path, cache=cache)
assert await toolset.read_pyai_docs(PyaiDocsTopic.agent) == '# Agent remote'
assert cache[PyaiDocsTopic.agent] == '# Agent remote'
async def test_remote_error_without_local_checkout(self, monkeypatch: pytest.MonkeyPatch) -> None:
_install_fake_httpx(monkeypatch, error=httpx.ConnectError('boom'))
toolset = PyaiDocsToolset[None](local_docs_path=None, cache=None)
toolset = PyaiDocsToolset[object](local_docs_path=None, cache=None)
with pytest.raises(RuntimeError, match='no local checkout configured'):
await toolset.read_pyai_docs(PyaiDocsTopic.tools)
async def test_remote_error_reports_local_path(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
_install_fake_httpx(monkeypatch, status=404)
toolset = PyaiDocsToolset[None](local_docs_path=tmp_path, cache=None)
toolset = PyaiDocsToolset[object](local_docs_path=tmp_path, cache=None)
with pytest.raises(RuntimeError, match=str(tmp_path)):
await toolset.read_pyai_docs(PyaiDocsTopic.toolsets)
@@ -110,14 +110,14 @@ class TestPyaiDocsToolset:
async def test_tools_advanced_reads_hyphenated_file(self, tmp_path: Path) -> None:
(tmp_path / 'tools-advanced.md').write_text('# Tools advanced local', encoding='utf-8')
toolset = PyaiDocsToolset[None](local_docs_path=tmp_path, cache=None)
toolset = PyaiDocsToolset[object](local_docs_path=tmp_path, cache=None)
assert await toolset.read_pyai_docs(PyaiDocsTopic.tools_advanced) == '# Tools advanced local'
async def test_local_path_expands_user(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv('HOME', str(tmp_path))
(tmp_path / 'hooks.md').write_text('# Hooks home', encoding='utf-8')
toolset = PyaiDocsToolset[None](local_docs_path=Path('~'), cache=None)
toolset = PyaiDocsToolset[object](local_docs_path=Path('~'), cache=None)
assert await toolset.read_pyai_docs(PyaiDocsTopic.hooks) == '# Hooks home'
@@ -125,29 +125,29 @@ class TestPyaiDocsToolset:
class TestPyaiDocsCapability:
def test_resolved_path_prefers_constructor_arg(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv('PYDANTIC_AI_HARNESS_DOCS_PATH', '/env/ignored')
assert PyaiDocs[None](local_docs_path=tmp_path)._resolved_local_path() == tmp_path
assert PyaiDocs[object](local_docs_path=tmp_path)._resolved_local_path() == tmp_path
def test_resolved_path_falls_back_to_env(self, monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
monkeypatch.setenv('PYDANTIC_AI_HARNESS_DOCS_PATH', str(tmp_path))
assert PyaiDocs[None]()._resolved_local_path() == tmp_path
assert PyaiDocs[object]()._resolved_local_path() == tmp_path
def test_resolved_path_none_when_unset(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv('PYDANTIC_AI_HARNESS_DOCS_PATH', raising=False)
assert PyaiDocs[None]()._resolved_local_path() is None
assert PyaiDocs[object]()._resolved_local_path() is None
def test_get_toolset_shares_cache_when_enabled(self) -> None:
capability = PyaiDocs[None]()
capability = PyaiDocs[object]()
toolset = capability.get_toolset()
assert isinstance(toolset, PyaiDocsToolset)
assert toolset._cache is capability._cache
def test_get_toolset_disables_cache(self) -> None:
toolset = PyaiDocs[None](cache=False).get_toolset()
toolset = PyaiDocs[object](cache=False).get_toolset()
assert isinstance(toolset, PyaiDocsToolset)
assert toolset._cache is None
def test_instructions_mention_the_tool(self) -> None:
instructions = PyaiDocs[None]().get_instructions()
instructions = PyaiDocs[object]().get_instructions()
assert isinstance(instructions, str)
assert 'read_pyai_docs' in instructions
tests/experimental/overflow/test_overflow.py
+40 -38
View File
@@ -101,7 +101,7 @@ def _tool_def(name: str = 'big_tool') -> ToolDefinition:
return ToolDefinition(name=name)
async def _run(cap: OverflowingToolOutput[None], result: Any, *, ctx: Any = None, tool_name: str = 'big_tool') -> Any:
async def _run(cap: OverflowingToolOutput[object], result: Any, *, ctx: Any = None, tool_name: str = 'big_tool') -> Any:
return await cap.after_tool_execute(
ctx if ctx is not None else _make_ctx(),
call=_call(tool_name),
@@ -296,14 +296,14 @@ class TestCleanup:
class TestConstruction:
def test_default_band_is_spill_then_truncate(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput()
cap: OverflowingToolOutput[object] = OverflowingToolOutput()
assert len(cap._bands) == 1
action = cap._bands[0].action
assert isinstance(action, Spill)
assert isinstance(action.then, Truncate)
def test_bands_sorted_descending(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=10, action=Truncate()), Band(over=100, action=Spill())]
)
assert [b.over for b in cap._bands] == [100, 10]
@@ -314,11 +314,11 @@ class TestConstruction:
def test_provided_store_used(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(store=store)
cap: OverflowingToolOutput[object] = OverflowingToolOutput(store=store)
assert cap._store is store
def test_per_tool_prepared(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
per_tool={'read_file': [Band(over=5, action=Truncate())]}
)
assert 'read_file' in cap._per_tool
@@ -331,19 +331,19 @@ class TestConstruction:
class TestPassthrough:
async def test_read_tool_exempt(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=1, action=Truncate(max_chars=2))])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=1, action=Truncate(max_chars=2))])
out = await _run(cap, 'x' * 100, tool_name=READ_TOOL_NAME)
assert out == 'x' * 100
async def test_tool_filter_skips_unmatched(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=1, action=Truncate(max_chars=2))], tool_filter=['other']
)
out = await _run(cap, 'x' * 100)
assert out == 'x' * 100
async def test_callable_filter(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=1, action=Truncate(max_chars=2))],
tool_filter=lambda ctx, td: td.name == 'big_tool',
)
@@ -351,12 +351,12 @@ class TestPassthrough:
assert isinstance(out, str) and 'truncated' in out
async def test_below_threshold_passthrough(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=1000, action=Truncate())])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=1000, action=Truncate())])
out = await _run(cap, 'small')
assert out == 'small'
async def test_exception_result_passthrough(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=1, action=Truncate(max_chars=2))])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=1, action=Truncate(max_chars=2))])
err = ValueError('boom')
assert await _run(cap, err) is err
@@ -368,26 +368,26 @@ class TestPassthrough:
class TestTruncate:
async def test_truncates_text(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=10, action=Truncate(max_chars=20, strategy=TruncationStrategy.head))]
)
out = await _run(cap, 'a' * 100)
assert isinstance(out, str) and out.startswith('a' * 20)
async def test_strip_ansi_applied(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=5, action=Truncate(max_chars=1000))], strip_ansi=True
)
out = await _run(cap, '\x1b[31m' + 'red text ' * 10 + '\x1b[0m')
assert isinstance(out, str) and '\x1b[' not in out
async def test_binary_truncate_falls_back_to_passthrough(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=1, action=Truncate())])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=1, action=Truncate())])
data = b'\x00\x01' * 100
assert await _run(cap, data) == data
async def test_tool_return_envelope_preserved(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=10, action=Truncate(max_chars=20))])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=10, action=Truncate(max_chars=20))])
out = await _run(cap, ToolReturn(return_value='a' * 100, content='note', metadata={'k': 1}))
assert isinstance(out, ToolReturn)
assert out.content == 'note'
@@ -402,7 +402,7 @@ class TestTruncate:
class TestSpill:
async def test_spill_roundtrip(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=10, action=Spill(preview_chars=20))], store=store
)
text = 'line\n' * 1000
@@ -415,7 +415,7 @@ class TestSpill:
async def test_spill_binary_verbatim(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=1, action=Spill())], store=store)
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=1, action=Spill())], store=store)
data = b'\x00\xff' * 100
out = await _run(cap, data)
assert isinstance(out, ToolReturn)
@@ -424,20 +424,20 @@ class TestSpill:
async def test_spill_structured_includes_sketch(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=5, action=Spill())], store=store)
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=5, action=Spill())], store=store)
out = await _run(cap, {'rows': list(range(1000)), 'ok': True})
assert isinstance(out, ToolReturn)
assert 'shape:' in out.return_value # type: ignore[operator]
async def test_spill_failure_falls_back_to_truncate(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=10, action=Spill(then=Truncate(max_chars=15)))], store=_BrokenStore()
)
out = await _run(cap, 'a' * 100)
assert isinstance(out, str) and 'truncated' in out
async def test_spill_failure_no_fallback_returns_original(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=10, action=Spill())], store=_BrokenStore()
)
out = await _run(cap, 'a' * 100)
@@ -445,14 +445,14 @@ class TestSpill:
async def test_handle_distinct_per_retry(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=5, action=Spill())], store=store)
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=5, action=Spill())], store=store)
out0 = await _run(cap, 'a' * 100, ctx=_make_ctx(retry=0))
out1 = await _run(cap, 'b' * 100, ctx=_make_ctx(retry=1))
assert out0.metadata['overflow_handle'] != out1.metadata['overflow_handle'] # type: ignore[union-attr]
async def test_spill_merges_existing_metadata(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=5, action=Spill())], store=store)
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=5, action=Spill())], store=store)
out = await _run(cap, ToolReturn(return_value='a' * 100, metadata={'orig': True}))
assert isinstance(out, ToolReturn)
assert out.metadata['orig'] is True
@@ -477,7 +477,7 @@ class _BrokenStore:
class TestContentReduction:
async def test_large_content_spilled(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=100, action=Spill(preview_chars=20))], store=store
)
out = await _run(cap, ToolReturn(return_value='small', content='C' * 5000))
@@ -488,20 +488,20 @@ class TestContentReduction:
assert await store.read(handle) == ('C' * 5000).encode('utf-8')
async def test_large_content_truncated(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=10, action=Truncate(max_chars=20))])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=10, action=Truncate(max_chars=20))])
out = await _run(cap, ToolReturn(return_value='small', content='C' * 200))
assert isinstance(out, ToolReturn)
assert isinstance(out.content, str) and 'truncated' in out.content
async def test_both_value_and_content_reduced(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=50, action=Spill())], store=store)
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=50, action=Spill())], store=store)
out = await _run(cap, ToolReturn(return_value='v' * 500, content='c' * 500))
assert isinstance(out, ToolReturn)
assert out.metadata['overflow_handle'] != out.metadata['overflow_content_handle']
async def test_nontext_content_warns_and_passes_through(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=10, action=Truncate())])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=10, action=Truncate())])
content = ['x' * 5000, BinaryContent(data=b'\x00', media_type='application/octet-stream')]
original = ToolReturn(return_value='small', content=content)
with pytest.warns(UserWarning, match='non-text content'):
@@ -509,14 +509,14 @@ class TestContentReduction:
assert out is original
async def test_nontext_content_passthrough_action_no_warn(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=1, action=Passthrough())])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=1, action=Passthrough())])
content = ['x' * 5000, BinaryContent(data=b'\x00', media_type='application/octet-stream')]
original = ToolReturn(return_value='small', content=content)
out = await _run(cap, original) # Passthrough action -> no warning, returned unchanged
assert out is original
async def test_small_nontext_content_no_warn(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=10_000, action=Truncate())])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=10_000, action=Truncate())])
content = ['tiny', BinaryContent(data=b'\x00', media_type='application/octet-stream')]
original = ToolReturn(return_value='small', content=content)
out = await _run(cap, original)
@@ -530,7 +530,7 @@ class TestContentReduction:
class TestSummarize:
async def test_custom_sync_summarizer(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=5, action=Summarize(summarize=lambda name, text: f'{name}:{len(text)}'))]
)
out = await _run(cap, 'x' * 100)
@@ -540,21 +540,23 @@ class TestSummarize:
async def summ(name: str, text: str) -> str:
return f'async:{len(text)}'
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=5, action=Summarize(summarize=summ))])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=5, action=Summarize(summarize=summ))]
)
out = await _run(cap, 'x' * 100)
assert out == 'async:100'
async def test_inherited_model_and_usage(self):
# model=None resolves to ctx.model, and the call threads usage=ctx.usage.
ctx = _make_ctx(model=_fixed_model('THE SUMMARY'))
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=5, action=Summarize())])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=5, action=Summarize())])
out = await _run(cap, 'x' * 100, ctx=ctx)
assert out == 'THE SUMMARY'
assert ctx.usage.requests == 1
async def test_explicit_model_overrides_ctx(self):
ctx = _make_ctx(model=_fixed_model('FROM CTX MODEL'))
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=5, action=Summarize(model=_fixed_model('FROM EXPLICIT MODEL')))]
)
out = await _run(cap, 'x' * 100, ctx=ctx)
@@ -562,7 +564,7 @@ class TestSummarize:
assert ctx.usage.requests == 1
async def test_binary_summarize_falls_back(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=1, action=Summarize(then=Passthrough()))]
)
data = b'\x00' * 100
@@ -572,7 +574,7 @@ class TestSummarize:
def boom(name: str, text: str) -> str:
raise RuntimeError('model down')
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=5, action=Summarize(summarize=boom, then=Truncate(max_chars=10)))]
)
out = await _run(cap, 'a' * 100)
@@ -586,11 +588,11 @@ class TestSummarize:
class TestActionsAndSelection:
async def test_passthrough_action(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(bands=[Band(over=1, action=Passthrough())])
cap: OverflowingToolOutput[object] = OverflowingToolOutput(bands=[Band(over=1, action=Passthrough())])
assert await _run(cap, 'x' * 100) == 'x' * 100
async def test_per_tool_replaces_bands(self):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=1, action=Truncate(max_chars=5))],
per_tool={'big_tool': [Band(over=100_000, action=Truncate())]},
)
@@ -709,7 +711,7 @@ class TestReadBack:
async def test_get_toolset_registers_read_tool(self, tmp_path: Path):
store = LocalFileStore(base_dir=tmp_path)
await store.write('h/1.0', b'hello\nworld')
cap: OverflowingToolOutput[None] = OverflowingToolOutput(store=store)
cap: OverflowingToolOutput[object] = OverflowingToolOutput(store=store)
toolset = cap.get_toolset()
assert toolset is not None
tool = toolset.tools[READ_TOOL_NAME] # type: ignore[union-attr]
@@ -725,7 +727,7 @@ class TestReadBack:
class TestAgentIntegration:
async def test_spill_persists_in_history(self, tmp_path: Path, anyio_backend: str):
store = LocalFileStore(base_dir=tmp_path)
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=100, action=Spill(preview_chars=50))], store=store
)
agent = Agent(TestModel(call_tools=['big_tool']), capabilities=[cap])
@@ -744,7 +746,7 @@ class TestAgentIntegration:
assert await store.read(part.metadata['overflow_handle']) == ('data line\n' * 500).encode('utf-8')
async def test_small_output_untouched(self, tmp_path: Path, anyio_backend: str):
cap: OverflowingToolOutput[None] = OverflowingToolOutput(
cap: OverflowingToolOutput[object] = OverflowingToolOutput(
bands=[Band(over=10_000, action=Spill())], store=LocalFileStore(base_dir=tmp_path)
)
agent = Agent(TestModel(call_tools=['small_tool']), capabilities=[cap])
tests/experimental/step_persistence/test_step_persistence.py
+20 -20
View File
@@ -75,8 +75,8 @@ def build_run_context(
)
def make_simple_agent(capabilities: list[Any]) -> Agent[None, str]:
agent: Agent[None, str] = Agent(TestModel(), capabilities=capabilities)
def make_simple_agent(capabilities: list[Any]) -> Agent[object, str]:
agent: Agent[object, str] = Agent(TestModel(), capabilities=capabilities)
@agent.tool_plain
def add(a: int, b: int) -> int: # pyright: ignore[reportUnusedFunction]
@@ -723,9 +723,9 @@ class TestStepPersistenceCapability:
async def test_single_capability_instance_reused_gets_fresh_ids(self) -> None:
"""One `StepPersistence(agent_name=...)` reused for two runs -> two distinct ids."""
store = InMemoryStepStore()
cap: StepPersistence[None] = StepPersistence(store=store, agent_name='librarian')
cap: StepPersistence[object] = StepPersistence(store=store, agent_name='librarian')
agent1: Agent[None, str] = Agent(TestModel(), capabilities=[cap])
agent1: Agent[object, str] = Agent(TestModel(), capabilities=[cap])
@agent1.tool_plain
def add(a: int, b: int) -> int: # pyright: ignore[reportUnusedFunction]
@@ -746,7 +746,7 @@ class TestStepPersistenceCapability:
is auto-set to the orchestrator's `run_id` without manual threading."""
store = InMemoryStepStore()
delegate: Agent[None, str] = Agent(
delegate: Agent[object, str] = Agent(
TestModel(),
capabilities=[StepPersistence(store=store, agent_name='delegate')],
)
@@ -755,7 +755,7 @@ class TestStepPersistenceCapability:
def add(a: int, b: int) -> int: # pyright: ignore[reportUnusedFunction]
return a + b
orchestrator: Agent[None, str] = Agent(
orchestrator: Agent[object, str] = Agent(
TestModel(),
capabilities=[StepPersistence(store=store, agent_name='orchestrator')],
)
@@ -792,7 +792,7 @@ class TestStepPersistenceCapability:
async def test_list_runs_parent_and_conversation_filters_combine(self) -> None:
store = InMemoryStepStore()
delegate: Agent[None, str] = Agent(
delegate: Agent[object, str] = Agent(
TestModel(),
capabilities=[StepPersistence(store=store, agent_name='delegate')],
)
@@ -801,7 +801,7 @@ class TestStepPersistenceCapability:
def add(a: int, b: int) -> int: # pyright: ignore[reportUnusedFunction]
return a + b
orchestrator: Agent[None, str] = Agent(
orchestrator: Agent[object, str] = Agent(
TestModel(),
capabilities=[StepPersistence(store=store, agent_name='orchestrator')],
)
@@ -880,7 +880,7 @@ class TestStepPersistenceCapability:
async def test_tool_failure_records_failed_status_and_event(self) -> None:
store = InMemoryStepStore()
agent: Agent[None, str] = Agent(TestModel(), capabilities=[StepPersistence(store=store)])
agent: Agent[object, str] = Agent(TestModel(), capabilities=[StepPersistence(store=store)])
@agent.tool_plain
def boom() -> int: # pyright: ignore[reportUnusedFunction]
@@ -966,8 +966,8 @@ class TestCrashMidToolCallContract:
async def test_visible_trail_no_false_continuation_point(self) -> None:
store = InMemoryStepStore()
cap: StepPersistence[None] = StepPersistence(store=store, agent_name='delegate')
agent: Agent[None, str] = Agent(TestModel(), capabilities=[cap])
cap: StepPersistence[object] = StepPersistence(store=store, agent_name='delegate')
agent: Agent[object, str] = Agent(TestModel(), capabilities=[cap])
@agent.tool_plain
def add(a: int, b: int) -> int: # pyright: ignore[reportUnusedFunction]
@@ -1024,7 +1024,7 @@ class TestCapabilityHookBranches:
async def test_effective_run_id_falls_back_to_capability_field(self) -> None:
"""When `ctx.run_id` is missing, the capability uses its own `run_id`."""
store = InMemoryStepStore()
cap: StepPersistence[None] = StepPersistence(store=store, run_id='configured', agent_name='a')
cap: StepPersistence[object] = StepPersistence(store=store, run_id='configured', agent_name='a')
ctx_no_run_id = build_run_context(deps=None, run_id=None)
await cap.before_run(ctx_no_run_id)
@@ -1036,7 +1036,7 @@ class TestCapabilityHookBranches:
async def test_after_run_skips_snapshot_when_history_not_provider_valid(self) -> None:
"""`after_run` only persists a snapshot when the history is provider-valid."""
store = InMemoryStepStore()
cap: StepPersistence[None] = StepPersistence(store=store)
cap: StepPersistence[object] = StepPersistence(store=store)
ctx = build_run_context(deps=None, run_id='r1')
unmatched: list[ModelMessage] = [
@@ -1057,7 +1057,7 @@ class TestCapabilityHookBranches:
async def test_after_run_saves_fallback_snapshot_when_no_node_snapshot(self) -> None:
"""With no `CallToolsNode` snapshot taken, `after_run` saves the final valid history."""
store = InMemoryStepStore()
cap: StepPersistence[None] = StepPersistence(store=store)
cap: StepPersistence[object] = StepPersistence(store=store)
ctx = build_run_context(deps=None, run_id='r1', run_step=3)
valid: list[ModelMessage] = [
@@ -1078,7 +1078,7 @@ class TestCapabilityHookBranches:
async def test_on_model_request_error_records_event_and_reraises(self) -> None:
store = InMemoryStepStore()
cap: StepPersistence[None] = StepPersistence(store=store)
cap: StepPersistence[object] = StepPersistence(store=store)
ctx = build_run_context(deps=None, run_id='r1')
request_context = ModelRequestContext(
model=ctx.model,
@@ -1098,7 +1098,7 @@ class TestCapabilityHookBranches:
async def test_for_run_returns_self_when_resolution_is_no_op(self) -> None:
"""When `run_id` is explicit and no contextvar is set, `for_run` returns `self`."""
store = InMemoryStepStore()
cap: StepPersistence[None] = StepPersistence(store=store, run_id='fixed')
cap: StepPersistence[object] = StepPersistence(store=store, run_id='fixed')
ctx = build_run_context(deps=None, run_id='ignored')
result = await cap.for_run(ctx)
@@ -1172,10 +1172,10 @@ class TestToolEffectMetadataPreservation:
from pydantic_ai_harness.experimental.step_persistence import annotate_tool_effect
store = InMemoryStepStore()
agent: Agent[None, str] = Agent(TestModel(), capabilities=[StepPersistence(store=store, run_id='r1')])
agent: Agent[object, str] = Agent(TestModel(), capabilities=[StepPersistence(store=store, run_id='r1')])
@agent.tool
async def write_label(ctx: RunContext[None], label: str) -> str: # pyright: ignore[reportUnusedFunction]
async def write_label(ctx: RunContext[object], label: str) -> str: # pyright: ignore[reportUnusedFunction]
await annotate_tool_effect(
store,
ctx,
@@ -1197,10 +1197,10 @@ class TestToolEffectMetadataPreservation:
from pydantic_ai_harness.experimental.step_persistence import annotate_tool_effect
store = InMemoryStepStore()
agent: Agent[None, str] = Agent(TestModel(), capabilities=[StepPersistence(store=store, run_id='r1')])
agent: Agent[object, str] = Agent(TestModel(), capabilities=[StepPersistence(store=store, run_id='r1')])
@agent.tool
async def boom(ctx: RunContext[None]) -> int: # pyright: ignore[reportUnusedFunction]
async def boom(ctx: RunContext[object]) -> int: # pyright: ignore[reportUnusedFunction]
await annotate_tool_effect(store, ctx, idempotency_key='boom-key')
raise ValueError('kaboom')
tests/experimental/subagents/test_subagents.py
+28 -28
View File
@@ -112,10 +112,10 @@ class TestConstruction:
assert SubAgents.get_serialization_name() is None
def test_empty_agents_no_instructions(self) -> None:
assert SubAgents[None]().get_instructions() is None
assert SubAgents[object]().get_instructions() is None
def test_empty_agents_no_toolset(self) -> None:
assert SubAgents[None]().get_toolset() is None
assert SubAgents[object]().get_toolset() is None
class TestInstructions:
@@ -171,7 +171,7 @@ class TestToolset:
class TestDelegation:
async def test_delegates_and_returns_output(self) -> None:
worker = Agent(TestModel(custom_output_text='WORKER RESULT'), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'), capabilities=[SubAgents(agents=[SubAgent(worker)])]
)
result = await parent.run('go')
@@ -186,7 +186,7 @@ class TestDelegation:
async def test_delegates_via_name_override(self) -> None:
worker = Agent(TestModel(custom_output_text='WORKER RESULT'), name='internal')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('public'),
capabilities=[SubAgents(agents=[SubAgent(worker, name='public')])],
)
@@ -197,7 +197,7 @@ class TestDelegation:
async def test_unknown_agent_triggers_retry_then_succeeds(self) -> None:
worker = Agent(TestModel(custom_output_text='OK'), name='worker')
helper = Agent(TestModel(custom_output_text='OK'), name='helper')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker', retries_before=1),
capabilities=[SubAgents(agents=[SubAgent(worker), SubAgent(helper)])],
)
@@ -254,7 +254,7 @@ class TestDelegation:
return ModelResponse(parts=[TextPart('sub done')])
worker = Agent(FunctionModel(worker_fn), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker)], inherit_tools=True)],
)
@@ -282,7 +282,7 @@ class TestDelegation:
return ModelResponse(parts=[TextPart('sub done')])
worker = Agent(FunctionModel(worker_fn), name='worker')
toolset: SubAgentToolset[None] = SubAgentToolset(
toolset: SubAgentToolset[object] = SubAgentToolset(
agents={'worker': SubAgent(worker)},
forward_usage=True,
inherit_tools=True,
@@ -291,7 +291,7 @@ class TestDelegation:
tool_name='delegate_task',
call_counts={},
)
parent: Agent[None, str] = Agent(_delegate_then_finish('worker'), toolsets=[toolset])
parent: Agent[object, str] = Agent(_delegate_then_finish('worker'), toolsets=[toolset])
@parent.tool_plain
def parent_tool() -> str: # pyright: ignore[reportUnusedFunction]
@@ -311,12 +311,12 @@ class TestDelegation:
from pydantic_ai.toolsets import FunctionToolset
@dataclass
class _ToolCapability(AbstractCapability[None]):
class _ToolCapability(AbstractCapability[object]):
def get_toolset(self) -> Any:
def cap_tool() -> str:
return 'CT' # pragma: no cover - never offered to the sub-agent
return FunctionToolset[None](tools=[cap_tool])
return FunctionToolset[object](tools=[cap_tool])
offered: list[str] = []
@@ -325,7 +325,7 @@ class TestDelegation:
return ModelResponse(parts=[TextPart('sub done')])
worker = Agent(FunctionModel(worker_fn), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker)], inherit_tools=True), _ToolCapability()],
)
@@ -340,9 +340,9 @@ class TestDelegation:
assert 'cap_tool' not in offered
async def test_shared_capabilities_applied_to_subagent(self) -> None:
cap: _RecordingCapability[None] = _RecordingCapability()
cap: _RecordingCapability[object] = _RecordingCapability()
worker = Agent(TestModel(custom_output_text='W'), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker)], shared_capabilities=[cap])],
)
@@ -353,12 +353,12 @@ class TestDelegation:
async def test_event_stream_handler_forwarded_to_subagent(self) -> None:
events: list[str] = []
async def handler(ctx: RunContext[None], stream: AsyncIterable[AgentStreamEvent]) -> None:
async def handler(ctx: RunContext[object], stream: AsyncIterable[AgentStreamEvent]) -> None:
async for event in stream:
events.append(type(event).__name__)
worker = Agent(TestModel(custom_output_text='W'), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker)], event_stream_handler=handler)],
)
@@ -371,7 +371,7 @@ class TestDelegation:
raise UsageLimitExceeded('limit hit')
limited = Agent(FunctionModel(boom), name='limited')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('limited'),
capabilities=[SubAgents(agents=[SubAgent(limited)])],
)
@@ -400,7 +400,7 @@ class TestDelegation:
)
return ModelResponse(parts=[TextPart('all done')])
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
FunctionModel(parent_fn),
capabilities=[SubAgents(agents=[SubAgent(boomer), SubAgent(worker)])],
)
@@ -451,17 +451,17 @@ class TestRunControls:
worker = Agent(TestModel(custom_output_text='W'), name='worker')
@worker.instructions
def _capture(ctx: RunContext[None]) -> str: # pyright: ignore[reportUnusedFunction]
def _capture(ctx: RunContext[object]) -> str: # pyright: ignore[reportUnusedFunction]
captured['usage_is_parent'] = ctx.usage is parent_usage.get('usage')
return ''
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker, usage_limits=UsageLimits(request_limit=5))])],
)
@parent.instructions
def _remember_usage(ctx: RunContext[None]) -> str: # pyright: ignore[reportUnusedFunction]
def _remember_usage(ctx: RunContext[object]) -> str: # pyright: ignore[reportUnusedFunction]
parent_usage['usage'] = ctx.usage
return ''
@@ -485,7 +485,7 @@ class TestRunControls:
def noop() -> str: # pyright: ignore[reportUnusedFunction]
return 'x'
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker, usage_limits=UsageLimits(request_limit=1))])],
)
@@ -500,7 +500,7 @@ class TestRunControls:
# No per-child limit -> the child shares accounting and a parent-level usage
# limit remains a hard stop for the whole tree.
worker = Agent(TestModel(custom_output_text='W'), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker)])],
)
@@ -513,7 +513,7 @@ class TestRunControls:
return ModelResponse(parts=[TextPart('late')]) # pragma: no cover - cancelled by the timeout
worker = Agent(FunctionModel(slow_fn), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker, timeout_seconds=0.01)])],
)
@@ -531,7 +531,7 @@ class TestRunControls:
return ModelResponse(parts=[TextPart('W')])
worker = Agent(FunctionModel(worker_fn), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_n_then_finish('worker', 2),
capabilities=[SubAgents(agents=[SubAgent(worker, max_calls=1)])],
)
@@ -561,7 +561,7 @@ class TestRunControls:
raise UnexpectedModelBehavior('kaboom')
boomer = Agent(FunctionModel(boom), name='boomer')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('boomer'),
capabilities=[SubAgents(agents=[SubAgent(boomer, on_failure='steer: use existing evidence')])],
)
@@ -579,7 +579,7 @@ class TestRunControls:
async def test_on_failure_overrides_default_steering(self) -> None:
worker = Agent(TestModel(custom_output_text='W'), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_n_then_finish('worker', 2),
capabilities=[SubAgents(agents=[SubAgent(worker, max_calls=1, on_failure='custom budget note')])],
)
@@ -591,7 +591,7 @@ class TestRunControls:
async def test_limits_without_budget_run_normally(self) -> None:
# A SubAgent with only an unrelated control set must not alter the happy path.
worker = Agent(TestModel(custom_output_text='W'), name='worker')
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
_delegate_then_finish('worker'),
capabilities=[SubAgents(agents=[SubAgent(worker, timeout_seconds=30)])],
)
@@ -622,7 +622,7 @@ class TestRunControls:
)
return ModelResponse(parts=[TextPart('all done')])
parent: Agent[None, str] = Agent(
parent: Agent[object, str] = Agent(
FunctionModel(parent_fn),
capabilities=[SubAgents(agents=[SubAgent(worker, max_calls=1)])],
)
tests/experimental/subagents/test_subagents_disk.py
+16 -16
View File
@@ -175,7 +175,7 @@ class TestDiskLoading:
# The isolation fixture points the home root at an empty dir; populate its
# conventional folder and the default `SubAgents()` picks it up with no config.
_write_agent(Path.home() / '.agents' / 'agents', 'planner.md', '---\nname: planner\n---\nPlan.')
cap: SubAgents[None] = SubAgents()
cap: SubAgents[object] = SubAgents()
assert 'planner' in cap._by_name
def test_cwd_equal_home_loads_once_without_shadow_warning(self, monkeypatch: pytest.MonkeyPatch) -> None:
@@ -191,17 +191,17 @@ class TestDiskLoading:
_write_agent(root / '.agents' / 'agents', 'planner.md', '---\nname: planner\n---\nPlan.')
with warnings.catch_warnings():
warnings.simplefilter('error')
cap: SubAgents[None] = SubAgents()
cap: SubAgents[object] = SubAgents()
assert 'planner' in cap._by_name
def test_none_disables_loading(self) -> None:
cap: SubAgents[None] = SubAgents(agent_folders=None)
cap: SubAgents[object] = SubAgents(agent_folders=None)
assert cap._by_name == {}
assert cap.get_toolset() is None
def test_loads_agents_from_folder(self, tmp_path: Path) -> None:
_write_agent(tmp_path, 'researcher.md', '---\nname: researcher\ndescription: Researches\n---\nResearch well.')
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path])
assert 'researcher' in cap._by_name
agent = cap._by_name['researcher'].agent
assert agent.name == 'researcher'
@@ -209,11 +209,11 @@ class TestDiskLoading:
def test_name_falls_back_to_filename_stem(self, tmp_path: Path) -> None:
_write_agent(tmp_path, 'planner.md', 'No frontmatter, just a body.')
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path])
assert 'planner' in cap._by_name
def test_missing_folder_is_skipped(self, tmp_path: Path) -> None:
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path / 'does-not-exist'])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path / 'does-not-exist'])
assert cap._by_name == {}
def test_undecodable_file_is_skipped_with_warning(self, tmp_path: Path) -> None:
@@ -223,13 +223,13 @@ class TestDiskLoading:
(tmp_path / 'broken.md').write_bytes(b'---\nname: broken\n---\n\xff\xfe not utf-8')
_write_agent(tmp_path, 'valid.md', '---\nname: valid\n---\nWork.')
with pytest.warns(UserWarning, match='Skipping unreadable disk sub-agent file'):
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path])
assert 'valid' in cap._by_name
assert 'broken' not in cap._by_name
def test_listing_uses_description(self, tmp_path: Path) -> None:
_write_agent(tmp_path, 'r.md', '---\nname: r\ndescription: Researches\n---\nB')
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path])
instructions = cap.get_instructions()
assert isinstance(instructions, str)
assert '- r: Researches' in instructions
@@ -240,7 +240,7 @@ class TestPrecedence:
_write_agent(tmp_path, 'worker.md', '---\nname: worker\ndescription: from disk\n---\nB')
explicit = Agent(TestModel(), name='worker', description='from code')
with pytest.warns(UserWarning, match="Disk sub-agent 'worker' is shadowed"):
cap: SubAgents[None] = SubAgents(agents=[SubAgent(explicit)], agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agents=[SubAgent(explicit)], agent_folders=[tmp_path])
assert cap._by_name['worker'].agent is explicit
def test_earlier_folder_shadows_later(self, tmp_path: Path) -> None:
@@ -249,7 +249,7 @@ class TestPrecedence:
_write_agent(project, 'worker.md', '---\nname: worker\ndescription: project\n---\nB')
_write_agent(home, 'worker.md', '---\nname: worker\ndescription: home\n---\nB')
with pytest.warns(UserWarning, match="Disk sub-agent 'worker' is shadowed"):
cap: SubAgents[None] = SubAgents(agent_folders=[project, home])
cap: SubAgents[object] = SubAgents(agent_folders=[project, home])
listing = cap.get_instructions()
assert isinstance(listing, str)
assert 'project' in listing
@@ -260,7 +260,7 @@ class TestOverrides:
def test_model_and_effort_override(self, tmp_path: Path) -> None:
_write_agent(tmp_path, 'w.md', '---\nname: w\n---\nB')
model = TestModel()
cap: SubAgents[None] = SubAgents(
cap: SubAgents[object] = SubAgents(
agent_folders=[tmp_path],
agent_overrides={'w': AgentOverride(model=model, effort='high')},
)
@@ -269,7 +269,7 @@ class TestOverrides:
def test_effort_floored_without_override(self, tmp_path: Path) -> None:
_write_agent(tmp_path, 'w.md', '---\nname: w\n---\nB')
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path])
# No override -> effort defaults to the floor on the built agent's settings.
agent = cap._by_name['w'].agent
assert isinstance(agent, Agent)
@@ -287,7 +287,7 @@ class TestToolResolver:
def resolver(name: str) -> Sequence[AgentToolset[object]] | None:
return [toolset] if name == 'search' else None
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path], tool_resolver=resolver)
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path], tool_resolver=resolver)
# The resolved toolset is attached to the built agent.
assert toolset in cap._by_name['w'].agent.toolsets
@@ -303,14 +303,14 @@ class TestToolResolver:
def test_no_resolver_ignores_frontmatter_tools(self, tmp_path: Path) -> None:
_write_agent(tmp_path, 'w.md', '---\nname: w\ntools: Read, Edit\n---\nB')
# Without a resolver, no warning and no own tools -- inheritance is the path.
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path])
assert 'w' in cap._by_name
class TestModelInheritance:
async def test_disk_agent_inherits_parent_model(self, tmp_path: Path) -> None:
_write_agent(tmp_path, 'worker.md', '---\nname: worker\n---\nDo the work.')
cap: SubAgents[None] = SubAgents(agent_folders=[tmp_path])
cap: SubAgents[object] = SubAgents(agent_folders=[tmp_path])
disk_agent = cap._by_name['worker'].agent
assert isinstance(disk_agent, Agent)
@@ -322,7 +322,7 @@ class TestModelInheritance:
return ''
parent_model = _delegate_then_finish('worker')
parent: Agent[None, str] = Agent(parent_model, capabilities=[cap])
parent: Agent[object, str] = Agent(parent_model, capabilities=[cap])
result = await parent.run('go')
assert result.output == 'all done'
# The model-less disk agent ran on the parent's resolved model.
tests/logfire_variables/test_managed_prompt.py
+17 -30
View File
@@ -15,7 +15,6 @@ default Logfire instance keeps its variable registry across `configure()` calls.
from __future__ import annotations
import importlib.metadata
from collections.abc import Generator
from contextlib import contextmanager
from dataclasses import dataclass
@@ -41,13 +40,6 @@ pytestmark = pytest.mark.anyio
DEFAULT = 'You are a helpful assistant.'
# pydantic-ai 2.0.0 reworked Instrumentation: the agent run span was renamed from
# `agent run` to `invoke_agent agent`, the tool span from `running tool` to
# `execute_tool noop`, and several span attribute keys were renamed. Harness still
# supports the 1.x floor (`pydantic-ai-slim>=1.105.0`), so version-tolerant tests
# keep both the locked 1.x jobs and the `test on latest` (2.0.0) job green.
_PYDANTIC_AI_GE_2 = int(importlib.metadata.version('pydantic-ai-slim').split('.')[0]) >= 2
@pytest.fixture(autouse=True, scope='module')
def _configure_logfire() -> None:
@@ -216,10 +208,6 @@ async def test_records_variable_resolution_span(capfire: CaptureLogfire) -> None
)
@pytest.mark.skipif(
_PYDANTIC_AI_GE_2,
reason='pydantic-ai 2.0.0 reworked instrumentation span/attribute names; logfire snapshot needs a 2.0.0 refresh -- tracked',
)
async def test_baggage_propagates_to_run_and_child_spans(capfire: CaptureLogfire) -> None:
# `Instrumentation` produces the agent run / model request / tool spans; `ManagedPrompt`
# runs outermost so its `logfire.variables.prompt__baggage_slug` baggage lands on all of them.
@@ -267,28 +255,28 @@ async def test_baggage_propagates_to_run_and_child_spans(capfire: CaptureLogfire
'logfire.span_type': 'span',
'logfire.msg': 'chat test',
'logfire.variables.prompt__baggage_slug': '<code_default>',
'gen_ai.input.messages': '[{"role": "user", "parts": [{"type": "text", "content": "hello"}]}]',
'gen_ai.output.messages': '[{"role": "assistant", "parts": [{"type": "tool_call", "id": "pyd_ai_tool_call_id__noop", "name": "noop", "arguments": {}}]}]',
'gen_ai.system_instructions': '[{"type": "text", "content": "You are a helpful assistant."}]',
'logfire.json_schema': '{"type": "object", "properties": {"gen_ai.input.messages": {"type": "array"}, "gen_ai.output.messages": {"type": "array"}, "gen_ai.system_instructions": {"type": "array"}, "model_request_parameters": {"type": "object"}}}',
'gen_ai.input.messages': '[{"role":"user","parts":[{"type":"text","content":"hello"}]}]',
'gen_ai.output.messages': '[{"role":"assistant","parts":[{"type":"tool_call","id":"pyd_ai_tool_call_id__noop","name":"noop","arguments":{}}]}]',
'gen_ai.system_instructions': '[{"type":"text","content":"You are a helpful assistant."}]',
'logfire.json_schema': '{"type":"object","properties":{"gen_ai.input.messages":{"type":"array"},"gen_ai.output.messages":{"type":"array"},"gen_ai.system_instructions":{"type":"array"},"model_request_parameters":{"type":"object"}}}',
'gen_ai.usage.input_tokens': 51,
'gen_ai.usage.output_tokens': 2,
'gen_ai.response.model': 'test',
},
},
{
'name': 'running tool',
'name': 'execute_tool noop',
'attributes': {
'gen_ai.operation.name': 'execute_tool',
'gen_ai.tool.name': 'noop',
'gen_ai.tool.call.id': 'pyd_ai_tool_call_id__noop',
'tool_arguments': '{}',
'gen_ai.tool.call.arguments': '{}',
'gen_ai.agent.name': 'agent',
'logfire.msg': 'running tool: noop',
'logfire.json_schema': '{"type":"object","properties":{"tool_arguments":{"type":"object"},"tool_response":{"type":"object"},"gen_ai.tool.name":{},"gen_ai.tool.call.id":{}}}',
'logfire.json_schema': '{"type":"object","properties":{"gen_ai.tool.call.arguments":{"type":"object"},"gen_ai.tool.call.result":{"type":"object"},"gen_ai.tool.name":{},"gen_ai.tool.call.id":{}}}',
'logfire.span_type': 'span',
'logfire.variables.prompt__baggage_slug': '<code_default>',
'tool_response': 'ok',
'gen_ai.tool.call.result': 'ok',
},
},
{
@@ -304,17 +292,17 @@ async def test_baggage_propagates_to_run_and_child_spans(capfire: CaptureLogfire
'logfire.span_type': 'span',
'logfire.msg': 'chat test',
'logfire.variables.prompt__baggage_slug': '<code_default>',
'gen_ai.input.messages': '[{"role": "user", "parts": [{"type": "text", "content": "hello"}]}, {"role": "assistant", "parts": [{"type": "tool_call", "id": "pyd_ai_tool_call_id__noop", "name": "noop", "arguments": {}}]}, {"role": "user", "parts": [{"type": "tool_call_response", "id": "pyd_ai_tool_call_id__noop", "name": "noop", "result": "ok"}]}]',
'gen_ai.output.messages': '[{"role": "assistant", "parts": [{"type": "text", "content": "{\\"noop\\":\\"ok\\"}"}]}]',
'gen_ai.system_instructions': '[{"type": "text", "content": "You are a helpful assistant."}]',
'logfire.json_schema': '{"type": "object", "properties": {"gen_ai.input.messages": {"type": "array"}, "gen_ai.output.messages": {"type": "array"}, "gen_ai.system_instructions": {"type": "array"}, "model_request_parameters": {"type": "object"}}}',
'gen_ai.input.messages': '[{"role":"user","parts":[{"type":"text","content":"hello"}]},{"role":"assistant","parts":[{"type":"tool_call","id":"pyd_ai_tool_call_id__noop","name":"noop","arguments":{}}]},{"role":"user","parts":[{"type":"tool_call_response","id":"pyd_ai_tool_call_id__noop","name":"noop","result":"ok"}]}]',
'gen_ai.output.messages': '[{"role":"assistant","parts":[{"type":"text","content":"{\\"noop\\":\\"ok\\"}"}]}]',
'gen_ai.system_instructions': '[{"type":"text","content":"You are a helpful assistant."}]',
'logfire.json_schema': '{"type":"object","properties":{"gen_ai.input.messages":{"type":"array"},"gen_ai.output.messages":{"type":"array"},"gen_ai.system_instructions":{"type":"array"},"model_request_parameters":{"type":"object"}}}',
'gen_ai.usage.input_tokens': 52,
'gen_ai.usage.output_tokens': 6,
'gen_ai.response.model': 'test',
},
},
{
'name': 'agent run',
'name': 'invoke_agent agent',
'attributes': {
'model_name': 'test',
'agent_name': 'agent',
@@ -324,10 +312,10 @@ async def test_baggage_propagates_to_run_and_child_spans(capfire: CaptureLogfire
'logfire.span_type': 'span',
'logfire.variables.prompt__baggage_slug': '<code_default>',
'final_result': '{"noop":"ok"}',
'gen_ai.usage.input_tokens': 103,
'gen_ai.usage.output_tokens': 8,
'gen_ai.aggregated_usage.input_tokens': 103,
'gen_ai.aggregated_usage.output_tokens': 8,
'pydantic_ai.all_messages': '[{"role":"user","parts":[{"type":"text","content":"hello"}]},{"role":"assistant","parts":[{"type":"tool_call","id":"pyd_ai_tool_call_id__noop","name":"noop","arguments":{}}]},{"role":"user","parts":[{"type":"tool_call_response","id":"pyd_ai_tool_call_id__noop","name":"noop","result":"ok"}]},{"role":"assistant","parts":[{"type":"text","content":"{\\"noop\\":\\"ok\\"}"}]}]',
'gen_ai.system_instructions': '[{"type": "text", "content": "You are a helpful assistant."}]',
'gen_ai.system_instructions': '[{"type":"text","content":"You are a helpful assistant."}]',
'logfire.json_schema': '{"type":"object","properties":{"pydantic_ai.all_messages":{"type":"array"},"gen_ai.system_instructions":{"type":"array"},"final_result":{"type":"object"}}}',
},
},
@@ -481,8 +469,7 @@ async def test_provider_backed_resolution_tags_v1_instrumentation_spans(capfire:
spans = capfire.exporter.exported_spans_as_dict()
# Child spans are tagged with the resolved label via baggage.
tagged = {s['name'] for s in spans if s['attributes'].get('logfire.variables.prompt__remote_slug') == 'production'}
agent_span = 'invoke_agent agent' if _PYDANTIC_AI_GE_2 else 'agent run'
assert {agent_span, 'chat test'} <= tagged
assert {'invoke_agent agent', 'chat test'} <= tagged
def test_logfire_instance_with_prebuilt_variable_warns() -> None:
tests/test_readme_quick_start.py
+4 -4
View File
@@ -292,7 +292,7 @@ class TestReadmeQuickStart:
"""End-to-end check that the README's Quick start example still works."""
async def test_quick_start_runs_through_codemode_with_faked_io(self) -> None:
agent: Agent[None, str] = Agent(
agent: Agent[object, str] = Agent(
FunctionModel(_model_fn),
capabilities=[
# Wire the fake HN tools through the `MCP` capability the same way
@@ -301,7 +301,7 @@ class TestReadmeQuickStart:
# capability composition path as production without any network.
# MCP's `__init__` narrows `local` to MCP-specific types, but the
# parent `NativeOrLocalTool` accepts any `AbstractToolset` at runtime.
MCP[None](
MCP[object](
'https://hn.caseyjhand.com/mcp',
native=False,
local=_make_fake_hn_toolset(), # pyright: ignore[reportArgumentType]
@@ -309,8 +309,8 @@ class TestReadmeQuickStart:
# The auto-wrapped Tool would take its name from the function
# (`_fake_web_search`); pass `name='web_search'` so the sandbox
# exposes it under the same name the model uses.
WebSearch[None](native=False, local=Tool(_fake_web_search, name='web_search')),
CodeMode[None](),
WebSearch[object](native=False, local=Tool(_fake_web_search, name='web_search')),
CodeMode[object](),
],
)
uv.lock
Generated
+9 -9
View File
@@ -1059,9 +1059,9 @@ lint = [
requires-dist = [
{ name = "httpx", specifier = ">=0.28.1" },
{ name = "logfire", marker = "extra == 'logfire'", specifier = ">=4.31.0" },
{ name = "pydantic-ai-slim", specifier = ">=1.105.0" },
{ name = "pydantic-ai-slim", specifier = ">=2.1.0" },
{ name = "pydantic-ai-slim", extras = ["dbos"], marker = "extra == 'dbos'" },
{ name = "pydantic-ai-slim", extras = ["spec"], marker = "extra == 'logfire'", specifier = ">=1.105.0" },
{ name = "pydantic-ai-slim", extras = ["spec"], marker = "extra == 'logfire'", specifier = ">=2.1.0" },
{ name = "pydantic-ai-slim", extras = ["temporal"], marker = "extra == 'temporal'" },
{ name = "pydantic-monty", marker = "extra == 'code-mode'", specifier = ">=0.0.16" },
{ name = "pydantic-monty", marker = "extra == 'codemode'", specifier = ">=0.0.16" },
@@ -1076,7 +1076,7 @@ dev = [
{ name = "inline-snapshot", specifier = ">=0.32.5" },
{ name = "logfire", extras = ["httpx"], specifier = ">=4.31.0" },
{ name = "pydantic-ai-harness", extras = ["code-mode"] },
{ name = "pydantic-ai-slim", extras = ["spec"], specifier = ">=1.95.1" },
{ name = "pydantic-ai-slim", extras = ["spec"], specifier = ">=2.1.0" },
{ name = "pytest", specifier = ">=9.0.0" },
{ name = "pytest-anyio" },
{ name = "pytest-examples", specifier = ">=0.0.18" },
@@ -1089,7 +1089,7 @@ lint = [
[[package]]
name = "pydantic-ai-slim"
version = "1.107.0"
version = "2.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "exceptiongroup", marker = "python_full_version < '3.11'" },
@@ -1101,9 +1101,9 @@ dependencies = [
{ name = "pydantic-graph" },
{ name = "typing-inspection" },
]
sdist = { url = "https://files.pythonhosted.org/packages/4c/26/ced63dfaabbc77f3beb86d59689cdea748e7ccffb6b419dbaf4780f211e8/pydantic_ai_slim-1.107.0.tar.gz", hash = "sha256:4616f689a92fcfecfecf2a7af27aca22f139a873cf6d7a8929eaeee9c0eedbb4", size = 779902, upload-time = "2026-06-10T14:53:10.574Z" }
sdist = { url = "https://files.pythonhosted.org/packages/ca/d1/78fd15c9c68b95ac0bec2d2afe22feb3b2f46e4b8e3a6dd1cead61cde434/pydantic_ai_slim-2.1.0.tar.gz", hash = "sha256:f79dca2429dbb9d2e32a0e2c613cfb9b9d6f0dc81d42cff8a5c81ede8ca0a6c7", size = 738698, upload-time = "2026-06-29T09:51:18.634Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/15/57/71044e17f931b08cc3930bc0fe5a1e1fd37fa474ae826be004729ef1cb4a/pydantic_ai_slim-1.107.0-py3-none-any.whl", hash = "sha256:1af49bbae06a6c598f72c54d4734ba377100cac493c9a05fa8e089bebeae0da6", size = 964046, upload-time = "2026-06-10T14:53:03.333Z" },
{ url = "https://files.pythonhosted.org/packages/5e/c7/b531cf65a1f8b221ab39739a16fe08f7b6a56fd24db875b9c33010320c9c/pydantic_ai_slim-2.1.0-py3-none-any.whl", hash = "sha256:2afda56459606226113ab433ee659aac7ff29bb66feefa716c67144bd82e5ce2", size = 910109, upload-time = "2026-06-29T09:51:11.016Z" },
]
[package.optional-dependencies]
@@ -1236,7 +1236,7 @@ wheels = [
[[package]]
name = "pydantic-graph"
version = "1.107.0"
version = "2.1.0"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "httpx" },
@@ -1244,9 +1244,9 @@ dependencies = [
{ name = "pydantic" },
{ name = "typing-inspection" },
]
sdist = { url = "https://files.pythonhosted.org/packages/dd/c3/6e8c2d13b8701041f1b3eac5deb41f25d4dbfa479a190d5c6becc23f2a49/pydantic_graph-1.107.0.tar.gz", hash = "sha256:278dd89b3e33f3a2963ac949f27a53aef705c5d883a8ce5d06d23e6e3cfbd972", size = 62564, upload-time = "2026-06-10T14:53:13.366Z" }
sdist = { url = "https://files.pythonhosted.org/packages/39/20/018532c826aba3c29ffc37bda46c20f3db4584bb555a8c7dc2769866267d/pydantic_graph-2.1.0.tar.gz", hash = "sha256:36ed6af24543421fb628fee593ccf5553a286c1dd1677018f5064a4e44c4daba", size = 43052, upload-time = "2026-06-29T09:51:20.832Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fc/72/621556e3f5068400d43a0375d38e5963de30256eaa5a702aba12e82ed0ff/pydantic_graph-1.107.0-py3-none-any.whl", hash = "sha256:71add94fe7e14c703977a895117c475aae6c0b02a774a036c4d00d9a63c78b00", size = 80106, upload-time = "2026-06-10T14:53:06.543Z" },
{ url = "https://files.pythonhosted.org/packages/42/a8/7f32bdeda6cff28652bba1c21ab2155c81d1733abd4b8bfef46a26e8d3fe/pydantic_graph-2.1.0-py3-none-any.whl", hash = "sha256:bae1e99829abf590a8693442ba5693ace38d2d93a80ae48be55c610a503e3709", size = 50772, upload-time = "2026-06-29T09:51:14.161Z" },
]
[[package]]