"""Tests for the agent factory.""" from typing import Any import pytest from pydantic_ai.models.test import TestModel from pydantic_ai.workspaces import Workspace from pydantic_ai_backends import StateWorkspace from pydantic_deep import ( DeepAgentDeps, UploadedFile, create_deep_agent, create_default_deps, run_with_files, ) from pydantic_deep.agent import _DepsTodoProxy, _TodoProxyBinder from pydantic_deep.deps import _format_size from pydantic_deep.types import SubAgentConfig, Todo from tests.workspaces import state_workspace # Use TestModel to avoid requiring API keys TEST_MODEL = TestModel() class TestCreateDeepAgent: """Tests for create_deep_agent factory.""" def test_create_default_agent(self): """Test creating an agent with default settings.""" agent = create_deep_agent(model=TEST_MODEL) assert agent is not None def test_create_with_custom_model(self): """Test creating an agent with a custom model.""" agent = create_deep_agent(model=TEST_MODEL) assert agent is not None def test_create_with_instructions(self): """Test creating an agent with custom instructions.""" agent = create_deep_agent(model=TEST_MODEL, instructions="You are a test agent") assert agent is not None def test_create_without_todo(self): """Test creating an agent without todo toolset.""" agent = create_deep_agent(model=TEST_MODEL, include_todo=False) assert agent is not None def test_create_without_filesystem(self): """Test creating an agent without filesystem toolset.""" agent = create_deep_agent(model=TEST_MODEL, include_filesystem=False) assert agent is not None def test_create_without_subagents(self): """Test creating an agent without subagent toolset.""" agent = create_deep_agent(model=TEST_MODEL, include_subagents=False) assert agent is not None def test_create_with_subagent_configs(self): """Test creating an agent with custom subagent configs.""" subagents = [ SubAgentConfig( name="researcher", description="A research agent", instructions="You research topics", ), ] agent = create_deep_agent(model=TEST_MODEL, subagents=subagents) assert agent is not None @staticmethod def _default_factory(**parent_kwargs: object) -> Any: """Build the default subagent factory the way create_deep_agent does. Tests the factory in isolation rather than reaching into the caller's subagent-config list - which create_deep_agent no longer mutates. """ from pydantic_deep.agent import _make_default_deep_agent_factory defaults = dict( model=TEST_MODEL, edit_format=None, context_files=None, context_discovery=False, memory_store=None, memory_namespace="", web_search=False, web_fetch=False, eviction_token_limit=20_000, max_binary_content=3, on_eviction=None, include_execute=True, ) defaults.update(parent_kwargs) return _make_default_deep_agent_factory(**defaults) def test_default_subagent_factory_propagates_web_flags(self): """Regression for #77: default subagent factory must inherit parent's `web_search` and `web_fetch` flags. Otherwise Bedrock/Vertex Anthropic models error out on unsupported beta tools. """ from pydantic_ai.capabilities import WebFetch, WebSearch def _sub_caps(parent_web_search: bool, parent_web_fetch: bool) -> list[type]: factory = self._default_factory( web_search=parent_web_search, web_fetch=parent_web_fetch ) sub_agent = factory({"instructions": "sub instructions", "model": TEST_MODEL}) return [type(c) for c in sub_agent._root_capability.capabilities] off_types = _sub_caps(parent_web_search=False, parent_web_fetch=False) assert WebSearch not in off_types assert WebFetch not in off_types on_types = _sub_caps(parent_web_search=True, parent_web_fetch=True) assert WebSearch in on_types assert WebFetch in on_types def test_default_subagent_factory_propagates_eviction_config(self): """Regression for #206: default subagent factory must inherit parent's eviction settings. Otherwise every subagent gets a default EvictionCapability at /large_tool_results, which fails on sandboxed backends and silently degrades large tool results. """ from pydantic_deep.features.eviction import EvictionCapability marker = object() def _sub_eviction( limit: int | None, binary: int | None, on_eviction: Any ) -> list[EvictionCapability]: factory = self._default_factory( eviction_token_limit=limit, max_binary_content=binary, on_eviction=on_eviction, ) sub_agent = factory({"instructions": "sub instructions", "model": TEST_MODEL}) return [ c for c in sub_agent._root_capability.capabilities if isinstance(c, EvictionCapability) ] caps = _sub_eviction(5_000, 1, marker) assert len(caps) == 1 assert caps[0].token_limit == 5_000 assert caps[0].max_binary_content == 1 assert caps[0].on_eviction is marker # eviction_token_limit=None suppresses the default capability, so a # consumer who replaced the parent's capability gets no surprise # defaults in subagents either. assert _sub_eviction(None, 3, None) == [] async def test_default_subagent_factory_inherits_include_execute(self): """A subagent works in its parent's workspace, so it runs commands exactly when the parent does - not in a workspace that has none.""" from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart from pydantic_ai.models.function import AgentInfo, FunctionModel async def _tool_names(include_execute: bool) -> set[str]: seen: set[str] = set() def _model(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse: seen.update(t.name for t in info.function_tools) return ModelResponse(parts=[TextPart("done")]) factory = self._default_factory(include_execute=include_execute) sub_agent = factory({"instructions": "", "model": FunctionModel(_model)}) await sub_agent.run( "go", deps=DeepAgentDeps(), workspace=Workspace(StateWorkspace().backend()) ) return seen assert "execute" not in await _tool_names(include_execute=False) assert "execute" in await _tool_names(include_execute=True) def test_default_subagent_factory_prepends_base_prompt(self): """Subagent factory always prepends BASE_PROMPT before task instructions.""" from pydantic_deep.prompts import BASE_PROMPT factory = self._default_factory() sub_agent = factory({"instructions": "Research topics carefully.", "model": TEST_MODEL}) assert any(BASE_PROMPT in str(i.instruction) for i in sub_agent._instructions) assert any( "Research topics carefully." in str(i.instruction) for i in sub_agent._instructions ) def test_default_subagent_factory_no_task_instructions(self): """Subagent factory with empty instructions uses only BASE_PROMPT.""" from pydantic_deep.prompts import BASE_PROMPT factory = self._default_factory() sub_agent = factory({"instructions": "", "model": TEST_MODEL}) assert any(BASE_PROMPT in str(i.instruction) for i in sub_agent._instructions) def test_subagent_configs_not_mutated_and_no_toolset_doubling(self): """create_deep_agent must not mutate caller subagent dicts, so reusing the same list (or calling twice) never injects agent_factory into the caller's dict nor doubles the per-subagent context/memory toolsets.""" cfg: SubAgentConfig = SubAgentConfig( name="helper", description="A helper agent", instructions="help", context_files=["NOTES.md"], toolsets=[], ) cfgs = [cfg] for _ in range(2): create_deep_agent(model=TEST_MODEL, subagents=cfgs, include_memory=True) # Caller's dict stays pristine across calls. assert "agent_factory" not in cfg assert cfg["toolsets"] == [] def test_subagent_factory_gives_one_memory_named_after_the_subagent(self): """Regression for #155: a subagent has one memory, its own, never the parent's "main" notebook beside it.""" from pydantic_ai_harness import Memory cfg: SubAgentConfig = SubAgentConfig( name="researcher", description="explores", instructions="explore" ) from pydantic_ai_harness.memory import FileStore sub_agent = self._default_factory(memory_store=FileStore(".deep/memory"))(cfg) memories = [c for c in sub_agent._root_capability.capabilities if isinstance(c, Memory)] assert [m.agent_name for m in memories] == ["researcher"] def test_create_with_interrupt_on(self): """Test creating an agent with interrupt_on config.""" agent = create_deep_agent( model=TEST_MODEL, interrupt_on={ "execute": True, "write_file": True, }, ) assert agent is not None def test_create_with_interrupt_on_all_false(self): """Test creating an agent with interrupt_on all False (no DeferredToolRequests).""" agent = create_deep_agent( model=TEST_MODEL, interrupt_on={ "execute": False, "write_file": False, }, ) assert agent is not None class TestBasePrompt: """Tests for BASE_PROMPT and default instructions.""" def test_base_prompt_is_default(self): """Default agent uses BASE_PROMPT as instructions.""" from pydantic_deep.prompts import BASE_PROMPT agent = create_deep_agent(model=TEST_MODEL, cost_tracking=False) # pydantic-ai stores instructions as a normalized list assert any(BASE_PROMPT in str(i.instruction) for i in agent._instructions) def test_custom_instructions_replace_base_prompt(self): """Custom instructions replace BASE_PROMPT entirely.""" from pydantic_deep.prompts import BASE_PROMPT custom = "You are a custom agent." agent = create_deep_agent(model=TEST_MODEL, instructions=custom, cost_tracking=False) assert any(custom in str(i.instruction) for i in agent._instructions) assert not any(BASE_PROMPT in str(i.instruction) for i in agent._instructions) def test_custom_instructions_with_base_prompt_fstring(self): """User can combine BASE_PROMPT with their own instructions via f-string.""" from pydantic_deep import BASE_PROMPT custom = f"{BASE_PROMPT}\n\nYou are a coding assistant." agent = create_deep_agent(model=TEST_MODEL, instructions=custom, cost_tracking=False) assert any(BASE_PROMPT in str(i.instruction) for i in agent._instructions) assert any("coding assistant" in str(i.instruction) for i in agent._instructions) def test_empty_instructions_uses_base_prompt(self): """instructions=None (default) uses BASE_PROMPT.""" from pydantic_deep.prompts import BASE_PROMPT agent = create_deep_agent(model=TEST_MODEL, instructions=None, cost_tracking=False) assert any(BASE_PROMPT in str(i.instruction) for i in agent._instructions) def test_create_with_all_capabilities_disabled(self): """Agent can be created with all built-in capabilities disabled (all_capabilities empty).""" agent = create_deep_agent( model=TEST_MODEL, context_manager=False, cost_tracking=False, stuck_loop_detection=False, patch_tool_calls=False, eviction_token_limit=None, web_search=False, web_fetch=False, thinking=False, capabilities=[], ) assert agent is not None def test_base_prompt_importable(self): """BASE_PROMPT is importable from pydantic_deep.""" from pydantic_deep import BASE_PROMPT assert isinstance(BASE_PROMPT, str) assert len(BASE_PROMPT) > 100 assert "You are a Deep Agent" in BASE_PROMPT def test_base_prompt_content(self): """BASE_PROMPT contains the core sections.""" from pydantic_deep.prompts import BASE_PROMPT assert "# Tool usage" in BASE_PROMPT assert "# Doing tasks" in BASE_PROMPT assert "# Acting with care" in BASE_PROMPT assert "# Communicating" in BASE_PROMPT class TestSubmodelInheritance: """Submodels inherit the primary model when not explicitly configured.""" def _capture_summarization(self, monkeypatch: pytest.MonkeyPatch, **kwargs: Any) -> Any: from pydantic_ai_summarization import ContextManagerCapability seen: dict[str, Any] = {} orig = ContextManagerCapability.__init__ def spy(self: Any, *a: Any, **k: Any) -> Any: seen["sm"] = k.get("summarization_model") return orig(self, *a, **k) monkeypatch.setattr(ContextManagerCapability, "__init__", spy) create_deep_agent(cost_tracking=False, include_subagents=False, **kwargs) return seen.get("sm") def test_summarization_inherits_primary_model(self, monkeypatch: pytest.MonkeyPatch) -> None: # anthropic + a dummy key: provider construction needs a key but no network # or ADC, so this runs in CI. The test only checks the inherited string. monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key") sm = self._capture_summarization(monkeypatch, model="anthropic:claude-sonnet-4-6") assert sm == "anthropic:claude-sonnet-4-6" def test_summarization_inherits_a_primary_model_instance( self, monkeypatch: pytest.MonkeyPatch ) -> None: """An OpenAI-compatible gateway is a `Model`, not a string: compaction fell back to a default on another provider the deployment cannot reach.""" from pydantic_ai.models.openai import OpenAIChatModel from pydantic_ai.providers.openai import OpenAIProvider gateway = OpenAIChatModel( "qwen2.5", provider=OpenAIProvider(base_url="http://localhost:8080/v1", api_key="local"), ) sm = self._capture_summarization(monkeypatch, model=gateway) assert sm is gateway def test_explicit_summarization_model_wins(self, monkeypatch: pytest.MonkeyPatch) -> None: monkeypatch.setenv("ANTHROPIC_API_KEY", "test-key") sm = self._capture_summarization( monkeypatch, model="anthropic:claude-opus-4-6", summarization_model="anthropic:claude-haiku-4-5", ) assert sm == "anthropic:claude-haiku-4-5" class TestImageSupport: """Tests that image support is always enabled.""" def test_agent_has_image_support(self): """Test that agents are created with image support enabled.""" agent = create_deep_agent(model=TEST_MODEL) assert agent is not None class TestCreateDefaultDeps: """Tests for create_default_deps.""" def test_create_with_defaults(self): """Test creating deps with default settings.""" deps = create_default_deps() assert deps is not None assert deps.todos == [] assert deps.subagents == {} class TestDeepAgentDeps: """Tests for DeepAgentDeps.""" def test_get_todo_prompt_empty(self): """Test todo prompt with no todos.""" deps = DeepAgentDeps() prompt = deps.get_todo_prompt() assert prompt == "" def test_get_todo_prompt_with_todos(self): """Test todo prompt with todos.""" from pydantic_deep.types import Todo deps = DeepAgentDeps( todos=[ Todo(content="Test task", status="pending", active_form="Testing"), ], ) prompt = deps.get_todo_prompt() assert "Test task" in prompt assert "[ ]" in prompt def test_get_todo_prompt_blocked_status(self): """Test todo prompt renders the blocked status distinctly.""" from pydantic_deep.types import Todo deps = DeepAgentDeps( todos=[ Todo(content="Blocked task", status="blocked", active_form="Blocking"), ], ) prompt = deps.get_todo_prompt() assert "Blocked task" in prompt assert "[!]" in prompt def test_clone_for_subagent(self): """Test cloning deps for a subagent.""" from pydantic_deep.types import Todo original = DeepAgentDeps( todos=[Todo(content="Task", status="pending", active_form="Working")], ) cloned = original.clone_for_subagent() # Should have empty todos and subagents assert cloned.todos == [] assert cloned.subagents == {} def test_checkpoint_store_default(self): """checkpoint_store defaults to None.""" deps = DeepAgentDeps() assert deps.checkpoint_store is None def test_checkpoint_store_set(self): """checkpoint_store can be passed to constructor.""" from pydantic_deep.features.checkpointing import InMemoryCheckpointStore store = InMemoryCheckpointStore() deps = DeepAgentDeps(checkpoint_store=store) assert deps.checkpoint_store is store def test_clone_for_subagent_propagates_checkpoint_store(self): """checkpoint_store is shared with subagent deps.""" from pydantic_deep.features.checkpointing import InMemoryCheckpointStore store = InMemoryCheckpointStore() original = DeepAgentDeps(checkpoint_store=store) cloned = original.clone_for_subagent() assert cloned.checkpoint_store is store async def test_upload_file(self): """Test uploading a file.""" deps = DeepAgentDeps() content = b"id,name,value\n1,foo,100\n2,bar,200\n" path = await deps.upload_file("data.csv", content) # Check path is correct assert path == "uploads/data.csv" # The file reaches a workspace when a run starts workspace = state_workspace() await deps.write_uploads(workspace) assert await workspace.read_bytes(path) == content # Check upload metadata is tracked assert path in deps.uploads assert deps.uploads[path]["name"] == "data.csv" assert deps.uploads[path]["size"] == len(content) assert deps.uploads[path]["line_count"] == 3 async def test_every_workspace_gets_the_uploads_once(self): """A run in a new in-memory document must find what the prompt lists.""" deps = DeepAgentDeps() path = await deps.upload_file("a.csv", b"v1") documents = StateWorkspace() first, second = Workspace(documents.backend()), Workspace(documents.backend()) await deps.write_uploads(first) await deps.write_uploads(second) assert await second.read_bytes(path) == b"v1" # The agent's own change to an upload survives its next run there... await first.write_bytes(path, b"edited") await deps.write_uploads(first) assert await first.read_bytes(path) == b"edited" # ...until the file is uploaded again, which replaces it. await deps.upload_file("a.csv", b"v2") await deps.write_uploads(first) assert await first.read_bytes(path) == b"v2" async def test_a_workspace_without_a_ref_gets_only_missing_uploads(self): """A fork branch's view: the parent's file shows through, so nothing is written.""" from pydantic_deep import BranchOverlay deps = DeepAgentDeps() path = await deps.upload_file("a.csv", b"v1") await deps.upload_file("b.csv", b"b") parent = state_workspace({path: "parent's"}) overlay = BranchOverlay(parent) await deps.write_uploads(Workspace(overlay)) assert [change.path for change in overlay.changes()] == ["/uploads/b.csv"] async def test_a_second_run_in_a_new_document_finds_the_upload(self): agent = create_deep_agent( model=TestModel(call_tools=[]), include_subagents=False, web_search=False, web_fetch=False, ) deps = DeepAgentDeps() await deps.upload_file("a.csv", b"x,y") await agent.run("first", deps=deps) second = await agent.run("second", deps=deps) assert await second.workspace.read_bytes("uploads/a.csv") == b"x,y" async def test_upload_file_custom_dir(self): """Test uploading a file to a custom directory.""" deps = DeepAgentDeps() content = b"test content" path = await deps.upload_file("test.txt", content, upload_dir="/custom/dir") assert path == "/custom/dir/test.txt" assert path in deps.uploads async def test_upload_file_binary(self): """Test uploading a binary file (non-UTF-8).""" from unittest.mock import patch deps = DeepAgentDeps() # Binary content - mock chardet to return no encoding (platform-dependent) content = bytes([0x80, 0x81, 0x82, 0xFF]) with patch("pydantic_deep.deps.chardet.detect", return_value={"encoding": None}): path = await deps.upload_file("binary.dat", content) assert path == "uploads/binary.dat" # Binary files should have line_count = None assert deps.uploads[path]["line_count"] is None async def test_upload_files_skips_failures(self, caplog): """A failing file should not abort the rest of the batch (and is logged, B9).""" from unittest.mock import patch deps = DeepAgentDeps() real_upload_file = deps.upload_file async def flaky_upload_file(name, content, *, upload_dir="uploads"): # Simulate a non-RuntimeError failure mid-batch (e.g. an # encoding/metadata error or a backend that raises directly). if name == "bad.bin": raise ValueError("boom") return await real_upload_file(name, content, upload_dir=upload_dir) with ( patch.object(deps, "upload_file", side_effect=flaky_upload_file), caplog.at_level("WARNING"), ): paths = await deps.upload_files( [ ("good1.csv", b"a,b\n1,2\n"), ("bad.bin", b"\x00\x01"), ("good2.txt", b"hello"), ] ) # The bad file is skipped, the good ones still upload. assert paths == ["uploads/good1.csv", "uploads/good2.txt"] # The skip is surfaced, not silent (B9). assert any("bad.bin" in r.getMessage() for r in caplog.records) def test_get_uploads_summary_empty(self): """Test uploads summary with no uploads.""" deps = DeepAgentDeps() summary = deps.get_uploads_summary() assert summary == "" async def test_get_uploads_summary_with_files(self): """Test uploads summary with uploaded files.""" deps = DeepAgentDeps() await deps.upload_file("data.csv", b"a,b,c\n1,2,3\n") await deps.upload_file("config.json", b'{"key": "value"}') summary = deps.get_uploads_summary() assert "## Uploaded Files" in summary assert "`uploads/data.csv`" in summary assert "`uploads/config.json`" in summary assert "2 lines" in summary # data.csv has 2 lines assert "read_file" in summary # instruction about how to use files async def test_get_uploads_summary_with_binary_files(self): """Test uploads summary with binary files (no line count).""" from unittest.mock import patch deps = DeepAgentDeps() # Binary file - mock chardet to return no encoding (platform-dependent) with patch("pydantic_deep.deps.chardet.detect", return_value={"encoding": None}): await deps.upload_file("binary.dat", bytes([0x80, 0x81, 0x82])) summary = deps.get_uploads_summary() assert "## Uploaded Files" in summary assert "`uploads/binary.dat`" in summary # Binary files should not show line count assert "lines" not in summary.split("`uploads/binary.dat`")[1].split("\n")[0] def test_clone_for_subagent_shares_uploads(self): """Test that cloned deps share uploads with original.""" original = DeepAgentDeps() original.uploads["uploads/test.txt"] = UploadedFile( name="test.txt", path="uploads/test.txt", size=100, line_count=5, mime_type="text/plain", encoding="utf-8", ) cloned = original.clone_for_subagent() # Should share uploads assert cloned.uploads is original.uploads assert "uploads/test.txt" in cloned.uploads class TestFormatSize: """Tests for _format_size helper.""" def test_bytes(self): """Test formatting bytes.""" assert _format_size(500) == "500 B" assert _format_size(0) == "0 B" assert _format_size(1023) == "1023 B" def test_kilobytes(self): """Test formatting kilobytes.""" assert _format_size(1024) == "1.0 KB" assert _format_size(2048) == "2.0 KB" assert _format_size(1536) == "1.5 KB" def test_megabytes(self): """Test formatting megabytes.""" assert _format_size(1024 * 1024) == "1.0 MB" assert _format_size(2 * 1024 * 1024) == "2.0 MB" assert _format_size(int(2.5 * 1024 * 1024)) == "2.5 MB" class TestRunWithFiles: """Tests for run_with_files helper.""" @pytest.mark.anyio async def test_run_with_files_uploads_files(self): """Test that run_with_files uploads files before running agent.""" agent = create_deep_agent( model=TEST_MODEL, web_search=False, web_fetch=False, include_memory=False ) deps = DeepAgentDeps() files = [ ("data.csv", b"a,b\n1,2\n"), ("config.json", b"{}"), ] await run_with_files( agent, "Test query", deps, files=files, ) # Files should be uploaded assert "uploads/data.csv" in deps.uploads assert "uploads/config.json" in deps.uploads @pytest.mark.anyio async def test_run_with_files_custom_upload_dir(self): """Test run_with_files with custom upload directory.""" agent = create_deep_agent( model=TEST_MODEL, web_search=False, web_fetch=False, include_memory=False ) deps = DeepAgentDeps() files = [("test.txt", b"content")] await run_with_files( agent, "Test query", deps, files=files, upload_dir="/custom", ) assert "/custom/test.txt" in deps.uploads @pytest.mark.anyio async def test_run_with_files_no_files(self): """Test run_with_files with no files.""" agent = create_deep_agent( model=TEST_MODEL, web_search=False, web_fetch=False, include_memory=False ) deps1 = DeepAgentDeps() deps2 = DeepAgentDeps() # Should not raise error await run_with_files(agent, "Test query", deps1, files=None) await run_with_files(agent, "Test query", deps2, files=[]) assert deps1.uploads == {} assert deps2.uploads == {} class TestDepsTodoProxy: """Tests for _DepsTodoProxy.""" def test_returns_empty_list_when_deps_is_none(self): """Proxy returns [] before being bound to any deps.""" proxy = _DepsTodoProxy() assert proxy.todos == [] def test_delegates_read_to_deps(self): """Proxy reads from deps.todos.""" proxy = _DepsTodoProxy() deps = DeepAgentDeps() todo = Todo(content="Test task", status="pending", active_form="Testing") deps.todos = [todo] proxy._deps = deps assert proxy._deps is deps assert proxy.todos == [todo] def test_delegates_write_to_deps(self): """Proxy writes to deps.todos.""" proxy = _DepsTodoProxy() deps = DeepAgentDeps() proxy._deps = deps todo = Todo(content="Test task", status="pending", active_form="Testing") proxy.todos = [todo] assert deps.todos == [todo] def test_setter_copies_list(self): """Setter creates a copy, not assigns reference.""" proxy = _DepsTodoProxy() deps = DeepAgentDeps() proxy._deps = deps original = [Todo(content="Task", status="pending", active_form="Working")] proxy.todos = original assert deps.todos == original assert deps.todos is not original def test_setter_noop_when_deps_is_none(self): """Setter silently no-ops without deps.""" proxy = _DepsTodoProxy() proxy.todos = [Todo(content="Task", status="pending", active_form="Working")] assert proxy.todos == [] def test_agent_created_with_proxy_storage(self): """Integration: agent is created with proxy-backed todo toolset.""" agent = create_deep_agent(model=TEST_MODEL, include_todo=True) # The agent should have a toolset with id "deep-todo" assert agent is not None async def test_concurrent_runs_isolated_via_contextvar(self): """Concurrent runs of one proxy see their own deps (no cross-run race). The proxy stores its bound deps in a ContextVar, and asyncio copies the context per task, so binding in one task must not leak into another. """ import asyncio proxy = _DepsTodoProxy() deps_a = DeepAgentDeps() deps_b = DeepAgentDeps() todo_a = Todo(content="A task", status="pending", active_form="Doing A") todo_b = Todo(content="B task", status="pending", active_form="Doing B") async def run(deps: DeepAgentDeps, todo: Todo, other_starts: asyncio.Event) -> list[Any]: proxy._deps = deps # Yield so the sibling task can bind its own deps in between, which # would clobber a shared attribute but not an isolated ContextVar. other_starts.set() await asyncio.sleep(0) proxy.todos = [todo] await asyncio.sleep(0) return proxy.todos ev_a = asyncio.Event() ev_b = asyncio.Event() result_a, result_b = await asyncio.gather( run(deps_a, todo_a, ev_a), run(deps_b, todo_b, ev_b), ) assert result_a == [todo_a] assert result_b == [todo_b] assert deps_a.todos == [todo_a] assert deps_b.todos == [todo_b] class TestTodoProxyBinder: """Tests for _TodoProxyBinder (issue #148).""" async def test_binds_proxy_to_ctx_deps(self): """before_tool_execute binds the proxy to the run's deps and passes args through.""" proxy = _DepsTodoProxy() deps = DeepAgentDeps() binder = _TodoProxyBinder(proxy) class _Ctx: pass ctx = _Ctx() ctx.deps = deps # type: ignore[attr-defined] sentinel: dict[str, Any] = {"todos": []} result = await binder.before_tool_execute( ctx, call=None, tool_def=None, args=sentinel, ) assert proxy._deps is deps assert result is sentinel async def test_write_todos_persists_to_deps_end_to_end(self): """write_todos must land on deps.todos through a real run. The todo tools are ``tool_plain`` and pydantic-ai runs each tool in its own ``contextvars`` context, so binding the proxy only in instructions never reached the tools — writes were silently dropped. The binder re-binds the proxy in each tool's own context. """ from pydantic_ai.messages import ModelResponse, TextPart, ToolCallPart from pydantic_ai.models.function import FunctionModel def model(messages: list[Any], info: Any) -> ModelResponse: returns = [ p for m in messages for p in getattr(m, "parts", []) if hasattr(p, "tool_name") ] if not returns: return ModelResponse( parts=[ ToolCallPart( tool_name="write_todos", args={ "todos": [ { "content": "Task A", "status": "pending", "active_form": "Doing A", }, { "content": "Task B", "status": "in_progress", "active_form": "Doing B", }, ] }, ) ] ) return ModelResponse(parts=[TextPart("done")]) agent = create_deep_agent( model=FunctionModel(model), include_todo=True, include_filesystem=False, include_execute=False, include_subagents=False, include_skills=False, include_plan=False, include_builtin_subagents=False, include_memory=False, context_manager=False, cost_tracking=False, ) deps = DeepAgentDeps() await agent.run("make todos", deps=deps) assert [t.content for t in deps.todos] == ["Task A", "Task B"] def test_no_capabilities_when_all_disabled(self): """Agent builds with an empty capability list when every feature is off.""" agent = create_deep_agent( model=TestModel(), include_todo=False, patch_tool_calls=False, stuck_loop_detection=False, web_search=False, web_fetch=False, thinking=False, cost_tracking=False, context_manager=False, eviction_token_limit=None, ) assert agent is not None class TestInjectSubagentExtraToolsets: """Tests for _inject_subagent_extra_toolsets helper.""" def test_noop_when_empty(self): """Empty extra_toolsets should be a no-op.""" from pydantic_deep.agent import _inject_subagent_extra_toolsets sa_config: dict[str, Any] = {"name": "test", "instructions": "test"} _inject_subagent_extra_toolsets(sa_config, ()) assert sa_config.get("toolsets") is None def test_injects_into_config_without_toolsets(self): """Extra toolsets should be added even when config has no toolsets key.""" from pydantic_deep.agent import _inject_subagent_extra_toolsets class _FakeToolset: pass sa_config: dict[str, Any] = {"name": "test", "instructions": "test"} toolset = _FakeToolset() _inject_subagent_extra_toolsets(sa_config, (toolset,)) assert sa_config["toolsets"] == [toolset] def test_extends_existing_toolsets(self): """Extra toolsets should be appended to existing ones.""" from pydantic_deep.agent import _inject_subagent_extra_toolsets class _FakeToolset: pass sa_config: dict[str, Any] = { "name": "test", "instructions": "test", "toolsets": ["existing"], } toolset = _FakeToolset() _inject_subagent_extra_toolsets(sa_config, (toolset,)) assert sa_config["toolsets"] == ["existing", toolset]