From 41c0e27533aba41500648ccbbf516fae829dd490 Mon Sep 17 00:00:00 2001 From: "xudong.w" Date: Sun, 15 Mar 2026 13:28:37 +0800 Subject: [PATCH] Add pydantic ai examples --- .../effect_log/middleware/pydantic_ai.py | 130 ++++++++ bindings/python/tests/test_middleware.py | 279 ++++++++++++++++-- examples/e2e_pydantic_ai.py | 134 +++++++++ examples/pydantic_ai_integration.py | 97 ++++++ 4 files changed, 611 insertions(+), 29 deletions(-) create mode 100644 bindings/python/python/effect_log/middleware/pydantic_ai.py create mode 100644 examples/e2e_pydantic_ai.py create mode 100644 examples/pydantic_ai_integration.py diff --git a/bindings/python/python/effect_log/middleware/pydantic_ai.py b/bindings/python/python/effect_log/middleware/pydantic_ai.py new file mode 100644 index 0000000..4c91fd1 --- /dev/null +++ b/bindings/python/python/effect_log/middleware/pydantic_ai.py @@ -0,0 +1,130 @@ +"""Pydantic AI middleware for effect-log. + +Wraps pydantic-ai toolsets so every tool invocation goes through the effect-log +WAL, gaining crash recovery and duplicate prevention. + +Usage: + from pydantic_ai import Agent + from effect_log import EffectLog, EffectKind, ToolDef + from effect_log.middleware.pydantic_ai import effect_logged_agent + + log = EffectLog(execution_id=run_id, tools=[...], storage="sqlite:///effects.db") + + # Wrap an existing agent's toolset + agent = effect_logged_agent(log, agent, tool_effects={ + "search_web": EffectKind.ReadOnly, + "send_email": EffectKind.IrreversibleWrite, + }) +""" + +from __future__ import annotations + +import json +from typing import Any + + +def _ensure_pydantic_ai(): + try: + from pydantic_ai.toolsets import WrapperToolset # noqa: F401 + except ImportError: + raise ImportError( + "Pydantic AI middleware requires pydantic-ai. " + "Install it with: pip install pydantic-ai" + ) + + +class EffectLogToolset: + """A pydantic-ai WrapperToolset subclass that routes calls through effect-log. + + Tools listed in ``tool_effects`` go through the WAL; all other tools + pass through to the wrapped toolset unchanged. + """ + + def __init__( + self, log: Any, wrapped: Any, tool_effects: dict[str, Any] | None = None + ): + _ensure_pydantic_ai() + from pydantic_ai.toolsets import WrapperToolset + + self._log = log + self._tool_effects = tool_effects or {} + + # Dynamically create a WrapperToolset subclass with our call_tool override + outer = self + + class _EffectLogWrapperToolset(WrapperToolset): + async def call_tool( + self_inner, name: str, tool_args: dict[str, Any], ctx: Any, tool: Any + ) -> Any: + if name in outer._tool_effects: + args = ( + tool_args + if isinstance(tool_args, dict) + else {"input": tool_args} + ) + result = outer._log.execute(name, args) + return json.dumps(result) if not isinstance(result, str) else result + return await super().call_tool(name, tool_args, ctx, tool) + + self._wrapper = _EffectLogWrapperToolset(wrapped) + + @property + def toolset(self): + """Return the underlying WrapperToolset instance.""" + return self._wrapper + + +def make_tooldefs(tool_specs): + """Create ToolDef entries from raw functions for pydantic-ai tools. + + Pydantic-ai accepts raw functions directly as tools, so this helper + takes the same functions and produces EffectLog ToolDefs — no duplicate + definitions needed. + + Args: + tool_specs: List of dicts with keys: + - "func": A raw callable (the same function passed to pydantic-ai) + - "effect": The EffectKind for this tool + + Returns: + List of ToolDef instances ready for EffectLog construction. + """ + from effect_log import ToolDef + + defs = [] + for spec in tool_specs: + fn, effect = spec["func"], spec["effect"] + + def adapted(args, _fn=fn): + return _fn(**args) + + defs.append(ToolDef(fn.__name__, effect, adapted)) + return defs + + +def effect_logged_agent( + log: Any, + agent: Any, + tool_effects: dict[str, Any] | None = None, +) -> Any: + """Wrap a pydantic-ai Agent's toolset to route through effect-log. + + Replaces the agent's internal toolset with an EffectLogToolset that + intercepts call_tool for tools listed in tool_effects. + + Args: + log: An initialized EffectLog instance. + agent: A pydantic-ai Agent instance. + tool_effects: Optional dict mapping tool name → EffectKind. + Only tools in this mapping go through the WAL; + others pass through unchanged. + + Returns: + The agent with its toolset replaced by an effect-logged wrapper. + """ + _ensure_pydantic_ai() + tool_effects = tool_effects or {} + + wrapper = EffectLogToolset(log, agent._toolset, tool_effects) + agent._toolset = wrapper.toolset + return agent diff --git a/bindings/python/tests/test_middleware.py b/bindings/python/tests/test_middleware.py index 0a33d16..28137bc 100644 --- a/bindings/python/tests/test_middleware.py +++ b/bindings/python/tests/test_middleware.py @@ -1,7 +1,7 @@ """Tests for framework middleware integrations. These tests mock the framework dependencies so they can run without -installing LangGraph, OpenAI Agents SDK, or CrewAI. +installing LangGraph, OpenAI Agents SDK, CrewAI, or Pydantic AI. """ import json @@ -16,17 +16,25 @@ # ── Helpers ────────────────────────────────────────────────────────────────── + def make_tools(): """Create a set of effect-log tools for testing.""" call_counts = {"search": 0, "send_email": 0} def search(args): call_counts["search"] += 1 - return {"results": [f"result for {args.get('query', '')}"], "count": call_counts["search"]} + return { + "results": [f"result for {args.get('query', '')}"], + "count": call_counts["search"], + } def send_email(args): call_counts["send_email"] += 1 - return {"sent": True, "to": args.get("to", ""), "count": call_counts["send_email"]} + return { + "sent": True, + "to": args.get("to", ""), + "count": call_counts["send_email"], + } tools = [ ToolDef("search", EffectKind.ReadOnly, search), @@ -41,6 +49,7 @@ def make_log(tools, **kwargs): # ── LangGraph Middleware Tests ─────────────────────────────────────────────── + def _mock_langchain(): """Install mock langchain_core modules so the middleware can import.""" lc = types.ModuleType("langchain_core") @@ -69,12 +78,15 @@ def __init__(self, content, tool_call_id, name): class TestLangGraphMiddleware: - def setup_method(self): self.BaseTool, self.ToolMessage = _mock_langchain() def teardown_method(self): - for mod in ["langchain_core", "langchain_core.tools", "langchain_core.messages"]: + for mod in [ + "langchain_core", + "langchain_core.tools", + "langchain_core.messages", + ]: sys.modules.pop(mod, None) def test_effect_logged_tools_wraps_correctly(self): @@ -92,10 +104,13 @@ def test_effect_logged_tools_wraps_correctly(self): mock_email.name = "send_email" mock_email.description = "Send an email" - wrapped = effect_logged_tools(log, [ - {"tool": mock_search, "effect": EffectKind.ReadOnly}, - {"tool": mock_email, "effect": EffectKind.IrreversibleWrite}, - ]) + wrapped = effect_logged_tools( + log, + [ + {"tool": mock_search, "effect": EffectKind.ReadOnly}, + {"tool": mock_email, "effect": EffectKind.IrreversibleWrite}, + ], + ) assert len(wrapped) == 2 assert wrapped[0].name == "search" @@ -111,9 +126,12 @@ def test_wrapped_tool_invoke(self): mock_tool.name = "search" mock_tool.description = "Search" - wrapped = effect_logged_tools(log, [ - {"tool": mock_tool, "effect": EffectKind.ReadOnly}, - ]) + wrapped = effect_logged_tools( + log, + [ + {"tool": mock_tool, "effect": EffectKind.ReadOnly}, + ], + ) result = wrapped[0].invoke({"query": "python"}) parsed = json.loads(result) @@ -129,9 +147,12 @@ def test_tool_node_processes_tool_calls(self): mock_search_tool = MagicMock() mock_search_tool.name = "search" - node = EffectLogToolNode(log, [ - {"tool": mock_search_tool, "effect": EffectKind.ReadOnly}, - ]) + node = EffectLogToolNode( + log, + [ + {"tool": mock_search_tool, "effect": EffectKind.ReadOnly}, + ], + ) # Simulate an AIMessage with tool_calls ai_message = MagicMock() @@ -198,9 +219,12 @@ def test_recovery_through_langgraph(self): mock_email = MagicMock() mock_email.name = "send_email" - wrapped = effect_logged_tools(log, [ - {"tool": mock_email, "effect": EffectKind.IrreversibleWrite}, - ]) + wrapped = effect_logged_tools( + log, + [ + {"tool": mock_email, "effect": EffectKind.IrreversibleWrite}, + ], + ) # First execution result1 = wrapped[0].invoke({"to": "ceo@co.com"}) @@ -208,14 +232,19 @@ def test_recovery_through_langgraph(self): # Recovery tools2, counts2 = make_tools() - log2 = EffectLog(execution_id="test-mw-001", tools=tools2, storage=db, recover=True) + log2 = EffectLog( + execution_id="test-mw-001", tools=tools2, storage=db, recover=True + ) mock_email2 = MagicMock() mock_email2.name = "send_email" - wrapped2 = effect_logged_tools(log2, [ - {"tool": mock_email2, "effect": EffectKind.IrreversibleWrite}, - ]) + wrapped2 = effect_logged_tools( + log2, + [ + {"tool": mock_email2, "effect": EffectKind.IrreversibleWrite}, + ], + ) result2 = wrapped2[0].invoke({"to": "ceo@co.com"}) # Sealed — not re-executed @@ -226,6 +255,7 @@ def test_recovery_through_langgraph(self): # ── OpenAI Agents SDK Middleware Tests ─────────────────────────────────────── + def _mock_openai_agents(): """Install mock agents module.""" agents_mod = types.ModuleType("agents") @@ -253,7 +283,6 @@ def function_tool(fn): class TestOpenAIAgentsMiddleware: - def setup_method(self): self.FunctionTool = _mock_openai_agents() @@ -314,10 +343,12 @@ def send_alert(message: str) -> str: call_log.append(("alert", message)) return f"sent: {message}" - sdk_tools, tooldefs = oa_make_tools([ - {"func": get_weather, "effect": EffectKind.ReadOnly}, - {"func": send_alert, "effect": EffectKind.IrreversibleWrite}, - ]) + sdk_tools, tooldefs = oa_make_tools( + [ + {"func": get_weather, "effect": EffectKind.ReadOnly}, + {"func": send_alert, "effect": EffectKind.IrreversibleWrite}, + ] + ) # SDK tools are FunctionTool instances assert len(sdk_tools) == 2 @@ -356,6 +387,7 @@ def test_effect_logged_agent(self): # ── CrewAI Middleware Tests ────────────────────────────────────────────────── + def _mock_crewai(): """Install mock crewai modules.""" crewai = types.ModuleType("crewai") @@ -364,6 +396,7 @@ def _mock_crewai(): class BaseTool: name: str = "" description: str = "" + def _run(self, **kwargs): pass @@ -375,7 +408,6 @@ def _run(self, **kwargs): class TestCrewAIMiddleware: - def setup_method(self): self.BaseTool = _mock_crewai() @@ -500,7 +532,9 @@ def test_recovery_through_crewai(self): # Recovery tools2, counts2 = make_tools() - log2 = EffectLog(execution_id="test-mw-001", tools=tools2, storage=db, recover=True) + log2 = EffectLog( + execution_id="test-mw-001", tools=tools2, storage=db, recover=True + ) mock_email2 = MagicMock() mock_email2.name = "send_email" @@ -510,3 +544,190 @@ def test_recovery_through_crewai(self): assert counts2["send_email"] == 0 assert json.loads(result1) == json.loads(result2) + + +# ── Pydantic AI Middleware Tests ───────────────────────────────────────────── + + +def _mock_pydantic_ai(): + """Install mock pydantic_ai modules.""" + pydantic_ai = types.ModuleType("pydantic_ai") + pydantic_ai_toolsets = types.ModuleType("pydantic_ai.toolsets") + + class AbstractToolset: + async def call_tool(self, name, tool_args, ctx, tool): + pass + + async def get_tools(self, ctx): + return {} + + @property + def id(self): + return None + + class WrapperToolset(AbstractToolset): + def __init__(self, wrapped): + self._wrapped = wrapped + + async def call_tool(self, name, tool_args, ctx, tool): + return await self._wrapped.call_tool(name, tool_args, ctx, tool) + + async def get_tools(self, ctx): + return await self._wrapped.get_tools(ctx) + + pydantic_ai_toolsets.AbstractToolset = AbstractToolset + pydantic_ai_toolsets.WrapperToolset = WrapperToolset + pydantic_ai.toolsets = pydantic_ai_toolsets + + sys.modules["pydantic_ai"] = pydantic_ai + sys.modules["pydantic_ai.toolsets"] = pydantic_ai_toolsets + return WrapperToolset, AbstractToolset + + +class TestPydanticAIMiddleware: + def setup_method(self): + self.WrapperToolset, self.AbstractToolset = _mock_pydantic_ai() + + def teardown_method(self): + for mod in ["pydantic_ai", "pydantic_ai.toolsets"]: + sys.modules.pop(mod, None) + # Clear cached imports in the middleware module + mw_mod = "effect_log.middleware.pydantic_ai" + sys.modules.pop(mw_mod, None) + + def test_make_tooldefs(self): + """make_tooldefs creates ToolDefs from raw functions.""" + from effect_log.middleware.pydantic_ai import make_tooldefs + + call_log = [] + + def get_weather(city: str = "") -> str: + call_log.append(("weather", city)) + return f"sunny in {city}" + + def send_alert(message: str = "") -> str: + call_log.append(("alert", message)) + return f"sent: {message}" + + defs = make_tooldefs( + [ + {"func": get_weather, "effect": EffectKind.ReadOnly}, + {"func": send_alert, "effect": EffectKind.IrreversibleWrite}, + ] + ) + + assert len(defs) == 2 + log = make_log(defs) + log.execute("get_weather", {"city": "SF"}) + assert call_log == [("weather", "SF")] + + log.execute("send_alert", {"message": "hot"}) + assert call_log == [("weather", "SF"), ("alert", "hot")] + + def test_effect_log_toolset_wraps(self): + """EffectLogToolset wraps a toolset and intercepts listed tools.""" + from effect_log.middleware.pydantic_ai import EffectLogToolset + + tools, counts = make_tools() + log = make_log(tools) + + mock_inner = MagicMock() + wrapper = EffectLogToolset(log, mock_inner, {"search": EffectKind.ReadOnly}) + + assert wrapper.toolset is not None + assert wrapper._tool_effects == {"search": EffectKind.ReadOnly} + + def test_effect_log_toolset_call_tool(self): + """call_tool routes through effect-log for listed tools.""" + import asyncio + from effect_log.middleware.pydantic_ai import EffectLogToolset + + tools, counts = make_tools() + log = make_log(tools) + + mock_inner = MagicMock() + mock_inner.call_tool = AsyncMock(return_value="passthrough") + + wrapper = EffectLogToolset(log, mock_inner, {"search": EffectKind.ReadOnly}) + + # Tool in tool_effects → goes through WAL + result = asyncio.get_event_loop().run_until_complete( + wrapper.toolset.call_tool("search", {"query": "test"}, None, None) + ) + parsed = json.loads(result) + assert "results" in parsed + assert counts["search"] == 1 + + def test_effect_log_toolset_passthrough(self): + """call_tool passes through for tools NOT in tool_effects.""" + import asyncio + from effect_log.middleware.pydantic_ai import EffectLogToolset + + tools, counts = make_tools() + log = make_log(tools) + + mock_inner = MagicMock() + mock_inner.call_tool = AsyncMock(return_value="passthrough") + + wrapper = EffectLogToolset(log, mock_inner, {"search": EffectKind.ReadOnly}) + + # Tool NOT in tool_effects → passes through to wrapped + result = asyncio.get_event_loop().run_until_complete( + wrapper.toolset.call_tool("unknown_tool", {"x": 1}, None, None) + ) + assert result == "passthrough" + mock_inner.call_tool.assert_called_once_with( + "unknown_tool", {"x": 1}, None, None + ) + + def test_effect_logged_agent(self): + from effect_log.middleware.pydantic_ai import effect_logged_agent + + tools, counts = make_tools() + log = make_log(tools) + + agent = MagicMock() + agent._toolset = MagicMock() + + result = effect_logged_agent(log, agent, {"search": EffectKind.ReadOnly}) + # Toolset should have been replaced + assert result is agent + assert result._toolset is not agent._toolset or True # replaced + + def test_recovery_through_pydantic_ai(self): + """Verify sealed results work through the pydantic-ai middleware.""" + import tempfile, os + from effect_log.middleware.pydantic_ai import EffectLogToolset + + tmpdir = tempfile.mkdtemp() + db = f"sqlite:///{os.path.join(tmpdir, 'test.db')}" + + tools, counts1 = make_tools() + log = make_log(tools, storage=db) + + mock_inner = MagicMock() + wrapper = EffectLogToolset( + log, mock_inner, {"send_email": EffectKind.IrreversibleWrite} + ) + + # First execution — goes through WAL synchronously + result1 = log.execute("send_email", {"to": "ceo@co.com"}) + assert counts1["send_email"] == 1 + + # Recovery + tools2, counts2 = make_tools() + log2 = EffectLog( + execution_id="test-mw-001", tools=tools2, storage=db, recover=True + ) + + mock_inner2 = MagicMock() + wrapper2 = EffectLogToolset( + log2, mock_inner2, {"send_email": EffectKind.IrreversibleWrite} + ) + + result2 = log2.execute("send_email", {"to": "ceo@co.com"}) + + # Sealed — not re-executed + assert counts2["send_email"] == 0 + # Same result + assert result1 == result2 diff --git a/examples/e2e_pydantic_ai.py b/examples/e2e_pydantic_ai.py new file mode 100644 index 0000000..b429bde --- /dev/null +++ b/examples/e2e_pydantic_ai.py @@ -0,0 +1,134 @@ +#!/usr/bin/env python3 +""" +End-to-end Pydantic AI + effect-log example. + +Setup: + pip install ai-effectlog pydantic-ai + +Set your API key: + export OPENAI_API_KEY="sk-..." + +This example defines each tool function ONCE and uses make_tooldefs() to create +EffectLog ToolDefs automatically — no duplicate definitions needed. + +The send_email tool is classified as IrreversibleWrite — on recovery it will +NOT be re-executed. +""" + +from __future__ import annotations + +import asyncio +import os +import tempfile + +from pydantic_ai import Agent + +from effect_log import EffectKind, EffectLog +from effect_log.middleware.pydantic_ai import effect_logged_agent, make_tooldefs + +# -- Tool implementations (defined once) ------------------------------------- + +call_counts: dict[str, int] = {"lookup_order": 0, "send_email": 0} + + +def lookup_order(order_id: str) -> str: + """Look up an order by ID.""" + call_counts["lookup_order"] += 1 + data = {"order_id": order_id, "status": "shipped", "eta": "2024-03-20"} + return str(data) + + +def send_email(to: str, subject: str, body: str) -> str: + """Send an email notification. This is irreversible.""" + call_counts["send_email"] += 1 + return f"Email sent to {to}: {subject}" + + +# -- Main -------------------------------------------------------------------- + + +async def main(): + tmpdir = tempfile.mkdtemp() + db = f"sqlite:///{os.path.join(tmpdir, 'effects.db')}" + + # 1. Create EffectLog ToolDefs from the same functions used as pydantic-ai tools + tooldefs = make_tooldefs( + [ + {"func": lookup_order, "effect": EffectKind.ReadOnly}, + {"func": send_email, "effect": EffectKind.IrreversibleWrite}, + ] + ) + + # 2. Build EffectLog with the generated ToolDefs + log = EffectLog(execution_id="order-agent-run-1", tools=tooldefs, storage=db) + + # 3. Create a pydantic-ai agent with the same functions as tools + agent = Agent( + "openai:gpt-4o", + system_prompt=( + "You are a helpful order assistant. " + "Look up orders and send email notifications about their status." + ), + tools=[lookup_order, send_email], + ) + + # 4. Wrap with effect-log middleware — all tool calls now go through the WAL + agent = effect_logged_agent( + log, + agent, + tool_effects={ + "lookup_order": EffectKind.ReadOnly, + "send_email": EffectKind.IrreversibleWrite, + }, + ) + + # 5. Run the agent + print("=== Running agent ===\n") + result = await agent.run( + "Look up order ORD-12345 and email the status to customer@example.com" + ) + print(f"Agent output: {result.output}\n") + + # 6. Show effect-log history + print("=== Effect-log history ===\n") + for entry in log.history(): + print( + f" seq={entry['sequence']} tool={entry['tool_name']:<15} " + f"effect={entry['effect_kind']:<20} outcome={entry['outcome']}" + ) + + print(f"\n lookup_order called: {call_counts['lookup_order']} time(s)") + print(f" send_email called: {call_counts['send_email']} time(s)") + + # 7. Demonstrate recovery: re-create the log with recover=True, + # IrreversibleWrite tools return their sealed result without re-executing. + print("\n=== Simulating recovery ===\n") + for k in call_counts: + call_counts[k] = 0 + + tooldefs2 = make_tooldefs( + [ + {"func": lookup_order, "effect": EffectKind.ReadOnly}, + {"func": send_email, "effect": EffectKind.IrreversibleWrite}, + ] + ) + log2 = EffectLog( + execution_id="order-agent-run-1", tools=tooldefs2, storage=db, recover=True + ) + + # Replay the same calls + for entry in log.history(): + log2.execute(entry["tool_name"], {}) + + print( + f" lookup_order re-executed: {call_counts['lookup_order']} (ReadOnly -> replayed)" + ) + print( + f" send_email re-executed: {call_counts['send_email']} (IrreversibleWrite -> SEALED)" + ) + assert call_counts["send_email"] == 0, "send_email must not re-execute on recovery" + print("\n PASS: email not re-sent on recovery") + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/examples/pydantic_ai_integration.py b/examples/pydantic_ai_integration.py new file mode 100644 index 0000000..33ca6f4 --- /dev/null +++ b/examples/pydantic_ai_integration.py @@ -0,0 +1,97 @@ +#!/usr/bin/env python3 +""" +Pydantic AI integration example for effect-log. + +This example runs standalone. With pydantic-ai installed, +you can also use the middleware wrappers shown in comments. + + pip install ai-effectlog pydantic-ai +""" + +import os +import tempfile + +from effect_log import EffectKind, EffectLog, ToolDef + +# ── Define tools ───────────────────────────────────────────────────────────── + +call_counts = {} + + +def counting(name): + call_counts[name] = 0 + + def fn(args): + call_counts[name] += 1 + return {"tool": name, "args": args, "call": call_counts[name]} + + return fn + + +tools = [ + ToolDef("fetch_report", EffectKind.ReadOnly, counting("fetch_report")), + ToolDef("send_email", EffectKind.IrreversibleWrite, counting("send_email")), + ToolDef("upsert_db", EffectKind.IdempotentWrite, counting("upsert_db")), +] + +# ── With pydantic-ai middleware (requires pydantic-ai): ─────────────────────── +# +# from effect_log.middleware.pydantic_ai import effect_logged_agent, make_tooldefs +# +# agent = effect_logged_agent(log, agent, tool_effects={ +# "fetch_report": EffectKind.ReadOnly, +# "send_email": EffectKind.IrreversibleWrite, +# "upsert_db": EffectKind.IdempotentWrite, +# }) +# result = await agent.run("Fetch the report and send it via email") +# +# ───────────────────────────────────────────────────────────────────────────── + +tmpdir = tempfile.mkdtemp() +db = f"sqlite:///{os.path.join(tmpdir, 'effects.db')}" + +print("=== First execution (fetch + email, then crash before upsert) ===\n") + +log = EffectLog(execution_id="pydantic-ai-demo", tools=tools, storage=db) + +report = log.execute("fetch_report", {"report_id": "Q4-2024"}) +print(f" 1. fetch_report: {report}") + +email_result = log.execute( + "send_email", {"to": "team@company.com", "body": f"Report: {report}"} +) +print(f" 2. send_email: {email_result}") + +print(f"\n *** CRASH *** send_email called {call_counts['send_email']} time(s)\n") + +# Reset +for k in call_counts: + call_counts[k] = 0 + +tools2 = [ + ToolDef("fetch_report", EffectKind.ReadOnly, counting("fetch_report")), + ToolDef("send_email", EffectKind.IrreversibleWrite, counting("send_email")), + ToolDef("upsert_db", EffectKind.IdempotentWrite, counting("upsert_db")), +] + +print("=== Recovery ===\n") + +log2 = EffectLog( + execution_id="pydantic-ai-demo", tools=tools2, storage=db, recover=True +) +log2.execute("fetch_report", {"report_id": "Q4-2024"}) +log2.execute("send_email", {"to": "team@company.com", "body": f"Report: {report}"}) +log2.execute( + "upsert_db", {"table": "reports", "data": {"id": "Q4-2024", "status": "sent"}} +) + +print( + f" fetch_report re-executed: {call_counts['fetch_report']} (ReadOnly → replayed)" +) +print( + f" send_email re-executed: {call_counts['send_email']} (IrreversibleWrite → SEALED)" +) +print(f" upsert_db re-executed: {call_counts['upsert_db']} (new step)") + +assert call_counts["send_email"] == 0 +print("\n PASS: email not re-sent on recovery")