From 10dcfaba84119f0b0421118bdb5805f4acd1ab30 Mon Sep 17 00:00:00 2001 From: igalshilman Date: Mon, 9 Nov 2026 11:58:21 +0100 Subject: [PATCH 1/5] Pydantic AI: no first-run banner on an agent's turns Pydantic AI prints a banner on a process's first agent run unless an installed package started it. A turn starts in a Restate handler, behind the event loop, where it cannot tell, so every server printed it. Turn it off with pydantic_ai.BANNER_ENABLED, as Pydantic AI advises a library that runs agents on its users' behalf. --- python/restate/agents/pydantic_ai/_agent.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/python/restate/agents/pydantic_ai/_agent.py b/python/restate/agents/pydantic_ai/_agent.py index 0e1fc615..57f5edbf 100644 --- a/python/restate/agents/pydantic_ai/_agent.py +++ b/python/restate/agents/pydantic_ai/_agent.py @@ -131,6 +131,7 @@ ) from pydantic_core import to_json, to_jsonable_python +import pydantic_ai from pydantic_ai import ( Agent, DeferredToolRequests, @@ -165,6 +166,12 @@ from restate.exceptions import TerminalError from restate.sessions import SessionContext +# Pydantic AI shows a banner on a process's first agent run, unless an installed +# package started the run. Ours start in a Restate handler, behind the event +# loop, where it cannot tell: turn it off, as it advises a library that runs +# agents on its users' behalf. +pydantic_ai.BANNER_ENABLED = False + M = typing.TypeVar("M", bound=BaseModel) AnyModel: typing.TypeAlias = str | Model From 56e65ddb5168c2e988716c90b5b7a8def23ad61f Mon Sep 17 00:00:00 2001 From: igalshilman Date: Mon, 9 Nov 2026 12:03:53 +0100 Subject: [PATCH 2/5] Examples: no two serve one name Restate keeps a key's state, its database with it, when another example is registered: router.py's desk loaded create_agent.py's LangGraph conversation, failed to read it, and was retried for good. The basic agents are openai_assistant, pydantic_ai_assistant, langgraph_assistant and adk_assistant; router.py's desk is front_desk, loop.py's refunds is refund_loop, and create_agent.py's orders service is shipping. A test fails when two examples serve a name; the tutorial's lessons, one application, share theirs. --- examples-agents/README.md | 2 +- examples-agents/basic/adk_agent.py | 6 +-- examples-agents/basic/create_agent.py | 8 ++-- examples-agents/basic/langgraph_agent.py | 6 +-- examples-agents/basic/openai_agent.py | 8 ++-- examples-agents/basic/pydantic_ai_agent.py | 6 +-- examples-agents/intermediate/loop.py | 12 +++--- examples-agents/intermediate/router.py | 10 ++--- tests/examples_agents.py | 46 +++++++++++++++------- 9 files changed, 61 insertions(+), 43 deletions(-) diff --git a/examples-agents/README.md b/examples-agents/README.md index a2654c80..03e20d4c 100644 --- a/examples-agents/README.md +++ b/examples-agents/README.md @@ -47,7 +47,7 @@ To talk with an agent from a terminal, as it works, run the agents' command line (it needs the `client` extra: `pip install "restate-sdk[client]"`): ```sh -python -m restate.agents chat assistant/alice # what you type goes to the agent; /help for the commands +python -m restate.agents chat openai_assistant/alice # what you type goes to the agent; /help for the commands ``` It shows the conversation so far, then each message, tool call and wait as it diff --git a/examples-agents/basic/adk_agent.py b/examples-agents/basic/adk_agent.py index 1c3ef442..a666583c 100644 --- a/examples-agents/basic/adk_agent.py +++ b/examples-agents/basic/adk_agent.py @@ -6,8 +6,8 @@ GOOGLE_API_KEY=... RESTATE_AGENT_MODEL=google/ uv run --extra adk --extra serve examples-agents/basic/adk_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/adk_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/adk_assistant/alice/messages --json '{}' # the conversation The model is RESTATE_AGENT_MODEL, a Gemini model (``google/``). For another provider, pass an ADK model of your own: ``model=LiteLlm(...)``. @@ -25,7 +25,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = adk.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = adk.agent("adk_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/basic/create_agent.py b/examples-agents/basic/create_agent.py index fe1fe413..80189abe 100644 --- a/examples-agents/basic/create_agent.py +++ b/examples-agents/basic/create_agent.py @@ -1,7 +1,7 @@ """ LangChain's ``create_agent`` as a Restate agent: an order desk, each conversation in its key's database. Its tool is Restate's: a call to the -``orders`` service, made once per tool call even if the process crashes +``shipping`` service, made once per tool call even if the process crashes mid-turn:: OPENAI_API_KEY=... uv run --extra langgraph --extra langchain --extra serve --with langchain-openai examples-agents/basic/create_agent.py @@ -21,10 +21,10 @@ import restate from restate.agents import call, langgraph -orders = restate.Service("orders") +shipping = restate.Service("shipping") -@orders.handler() +@shipping.handler() async def status(_ctx: restate.Context, order: str) -> str: return f"Order {order} was delivered yesterday." @@ -38,4 +38,4 @@ async def status(_ctx: restate.Context, order: str) -> str: if __name__ == "__main__": - asyncio.run(restate.serve(restate.app([orders, desk]))) + asyncio.run(restate.serve(restate.app([shipping, desk]))) diff --git a/examples-agents/basic/langgraph_agent.py b/examples-agents/basic/langgraph_agent.py index 337809c5..78573bd5 100644 --- a/examples-agents/basic/langgraph_agent.py +++ b/examples-agents/basic/langgraph_agent.py @@ -6,8 +6,8 @@ OPENAI_API_KEY=... uv run --extra langgraph --extra serve --with langchain-openai examples-agents/basic/langgraph_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/langgraph_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/langgraph_assistant/alice/messages --json '{}' # the conversation The model is ``openai/gpt-6-sol`` unless RESTATE_AGENT_MODEL names another: ``anthropic/`` takes ``langchain-anthropic`` (and @@ -26,7 +26,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = langgraph.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = langgraph.agent("langgraph_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/basic/openai_agent.py b/examples-agents/basic/openai_agent.py index 33b5331f..94a2e9ed 100644 --- a/examples-agents/basic/openai_agent.py +++ b/examples-agents/basic/openai_agent.py @@ -6,12 +6,12 @@ OPENAI_API_KEY=... uv run --extra openai --extra serve examples-agents/basic/openai_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/openai_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/openai_assistant/alice/messages --json '{}' # the conversation The model is ``openai/gpt-6-sol`` unless RESTATE_AGENT_MODEL names another OpenAI model. To bring an ``Agent`` of your own, with handoffs and guardrails, -pass it as ``openai.agent("assistant", agent=Agent(...))``. +pass it as ``openai.agent("openai_assistant", agent=Agent(...))``. """ import asyncio @@ -26,7 +26,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = openai.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = openai.agent("openai_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/basic/pydantic_ai_agent.py b/examples-agents/basic/pydantic_ai_agent.py index ef4c22a0..9413cd68 100644 --- a/examples-agents/basic/pydantic_ai_agent.py +++ b/examples-agents/basic/pydantic_ai_agent.py @@ -6,8 +6,8 @@ OPENAI_API_KEY=... uv run --extra pydantic_ai --extra serve --with 'pydantic-ai-slim[openai]' examples-agents/basic/pydantic_ai_agent.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/assistant/alice/chat -d 'What is the weather in Lisbon?' - curl localhost:8080/assistant/alice/messages --json '{}' # the conversation + curl localhost:8080/pydantic_ai_assistant/alice/chat -d 'What is the weather in Lisbon?' + curl localhost:8080/pydantic_ai_assistant/alice/messages --json '{}' # the conversation The model is ``openai/gpt-6-sol`` unless RESTATE_AGENT_MODEL names another, of any of Pydantic AI's providers: ``anthropic/`` takes @@ -26,7 +26,7 @@ async def get_weather(city: str) -> str: return f"Sunny and 21°C in {city}" -assistant = pydantic_ai.agent("assistant", system="Answer briefly.", tools=[get_weather]) +assistant = pydantic_ai.agent("pydantic_ai_assistant", system="Answer briefly.", tools=[get_weather]) if __name__ == "__main__": diff --git a/examples-agents/intermediate/loop.py b/examples-agents/intermediate/loop.py index dddbb0e8..4239d61a 100644 --- a/examples-agents/intermediate/loop.py +++ b/examples-agents/intermediate/loop.py @@ -7,15 +7,15 @@ OPENAI_API_KEY=... uv run --extra serve --with openai examples-agents/intermediate/loop.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/refunds/alice/chat -d 'Order 103 arrived broken, can I get my $30 back?' + curl localhost:8080/refund_loop/alice/chat -d 'Order 103 arrived broken, can I get my $30 back?' The agent waits for a manager to approve the refund. In another terminal, approve it:: - curl localhost:8080/refunds/alice/signal --json '{"name": "approval", "value": "approved"}' + curl localhost:8080/refund_loop/alice/signal --json '{"name": "approval", "value": "approved"}' Or talk with the agent, and approve, from the agents' command line: -``python -m restate.agents chat refunds/alice``, then ``/signal approval approved``. +``python -m restate.agents chat refund_loop/alice``, then ``/signal approval approved``. The loop is plain Python, and Restate makes it durable: @@ -204,9 +204,9 @@ async def run_call(call: Call) -> None: return await answer_once(ctx, text, MESSAGES, run) -refunds = AgentObject("refunds") -refunds.chat(answer, items=MESSAGES) +refund_loop = AgentObject("refund_loop") +refund_loop.chat(answer, items=MESSAGES) if __name__ == "__main__": - asyncio.run(restate.serve(restate.app([refunds]))) + asyncio.run(restate.serve(restate.app([refund_loop]))) diff --git a/examples-agents/intermediate/router.py b/examples-agents/intermediate/router.py index 98f7f661..42838e53 100644 --- a/examples-agents/intermediate/router.py +++ b/examples-agents/intermediate/router.py @@ -10,8 +10,8 @@ --with langchain-openai --with 'pydantic-ai-slim[openai]' examples-agents/intermediate/router.py npx @restatedev/restate-server # Restate, in another terminal npx @restatedev/restate -y deployments register --force http://localhost:9080 - curl localhost:8080/desk/alice/route --json '"I was charged twice for order 103"' - curl localhost:8080/desk/alice/route --json '"The app crashes when I open my orders"' + curl localhost:8080/front_desk/alice/route --json '"I was charged twice for order 103"' + curl localhost:8080/front_desk/alice/route --json '"The app crashes when I open my orders"' curl localhost:8080/billing/alice/messages --json '{}' # what billing was asked, and answered """ @@ -31,10 +31,10 @@ system="You handle technical problems with the app. Ask for what you need to know, one question at a time.", ) -desk = sessions.SessionObject("desk") +front_desk = sessions.SessionObject("front_desk") -@desk.handler +@front_desk.handler async def route(ctx: sessions.SessionContext, message: str) -> str: # Asked once: a retried attempt gets the same answer. team = await pydantic_ai.prompt( @@ -49,4 +49,4 @@ async def route(ctx: sessions.SessionContext, message: str) -> str: if __name__ == "__main__": - asyncio.run(restate.serve(restate.app([desk, billing, tech]))) + asyncio.run(restate.serve(restate.app([front_desk, billing, tech]))) diff --git a/tests/examples_agents.py b/tests/examples_agents.py index e37a90fb..ae676633 100644 --- a/tests/examples_agents.py +++ b/tests/examples_agents.py @@ -80,6 +80,24 @@ async def serving(path: str, setup: Callable[[ModuleType], None] | None = None) yield Ingress(harness.ingress_url) +async def test_no_two_examples_serve_a_name(monkeypatch: pytest.MonkeyPatch): + """A name is served by one example only. Restate keeps a key's state, its database with it, from one example to + the next: an example of another's name would load that one's conversations. The tutorial's lessons are versions + of one application, and share their names.""" + for framework in ("agents", "langgraph", "langchain.agents", "pydantic_ai", "google.adk"): + pytest.importorskip(framework) + served: dict[str, str] = {} + for path in sorted(EXAMPLES.glob("*/*.py")): + example = str(path.relative_to(EXAMPLES)) + with monkeypatch.context() as env: + if example == "basic/adk_agent.py": + env.setenv("RESTATE_AGENT_MODEL", "google/gemini-x") # an ADK agent needs a model + module = load(example) + owner = "the tutorial" if example.startswith("tutorial/") else example + for name in {value.name for value in vars(module).values() if isinstance(value, SERVABLE)}: + assert served.setdefault(name, owner) == owner, f"{example} and {served[name]} both serve {name!r}" + + async def test_01_object(): async with serving("tutorial/01_object.py") as ingress: assert (await ingress.post("wallet/alice/hello")).json() == "Hello, alice!" @@ -157,12 +175,12 @@ async def test_basic_langgraph_agent(): """The agent serves, with its companion; a message needs a model's provider, and is left to a real run.""" pytest.importorskip("langgraph") async with serving("basic/langgraph_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("langgraph_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("langgraph_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_basic_create_agent(monkeypatch: pytest.MonkeyPatch): - """create_agent's model (scripted here) calls the orders service through the Restate tool, and answers.""" + """create_agent's model (scripted here) calls the shipping service through the Restate tool, and answers.""" pytest.importorskip("langchain.agents") import langchain.agents.factory # pylint: disable=import-outside-toplevel from langchain_core.language_models import BaseChatModel # pylint: disable=import-outside-toplevel @@ -182,7 +200,7 @@ def _generate(self, messages: list[BaseMessage], *args: Any, **kwargs: Any) -> C if isinstance(last, ToolMessage): reply = AIMessage(str(last.text)) else: - reply = AIMessage("", tool_calls=[{"id": "c1", "name": "orders_status", "args": {"input": "103"}}]) + reply = AIMessage("", tool_calls=[{"id": "c1", "name": "shipping_status", "args": {"input": "103"}}]) return ChatResult(generations=[ChatGeneration(message=reply)]) # The example names an OpenAI model; the scripted one stands in for it, with no API key. @@ -196,8 +214,8 @@ async def test_basic_pydantic_ai_agent(): """The agent serves, with its companion; a message needs a model's provider, and is left to a real run.""" pytest.importorskip("pydantic_ai") async with serving("basic/pydantic_ai_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("pydantic_ai_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("pydantic_ai_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_basic_adk_agent(monkeypatch: pytest.MonkeyPatch): @@ -205,16 +223,16 @@ async def test_basic_adk_agent(monkeypatch: pytest.MonkeyPatch): pytest.importorskip("google.adk") monkeypatch.setenv("RESTATE_AGENT_MODEL", "google/gemini-x") # an ADK agent needs a model async with serving("basic/adk_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("adk_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("adk_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_basic_openai_agent(): """The agent serves, with its companion; a message needs a model's provider, and is left to a real run.""" pytest.importorskip("agents") async with serving("basic/openai_agent.py") as ingress: - assert (await ingress.post("assistant/alice/messages", {})).json()["messages"] == [] - assert (await ingress.post("assistant/alice/status")).json() == {"answering": None, "waiting": []} + assert (await ingress.post("openai_assistant/alice/messages", {})).json()["messages"] == [] + assert (await ingress.post("openai_assistant/alice/status")).json() == {"answering": None, "waiting": []} async def test_intermediate_support(): @@ -368,20 +386,20 @@ def counted(order: str, amount: float) -> str: module.TOOLS["refund"] = module.tool(counted, name="refund", description="Refund an order.") async with serving("intermediate/loop.py", setup) as ingress: - chat = asyncio.create_task(ingress.post("refunds/alice/chat", "Order 103 arrived broken")) + chat = asyncio.create_task(ingress.post("refund_loop/alice/chat", "Order 103 arrived broken")) waiting: list[dict[str, Any]] = [] for _ in range(100): - if waiting := (await ingress.post("refunds/alice/pending")).json(): + if waiting := (await ingress.post("refund_loop/alice/pending")).json(): break await asyncio.sleep(0.1) assert [w["name"] for w in waiting] == ["approval"] and not refunded # it waits, and has not refunded - await ingress.post("refunds/alice/signal", {"name": "approval", "value": "approved"}) + await ingress.post("refund_loop/alice/signal", {"name": "approval", "value": "approved"}) assert (await chat).json() == "Done: Refunded $30 for order 103." assert refunded == ["103"] and crashed == [True] and unanswered == [True] # The model was asked once for each step (and again after the crash that cut its connection), not after the # crash that came once its final reply was stored. assert asked == ["user", asked[1], "c2", "c3", "c3"] and asked[1] in ("c1", "c1b") - shown = (await ingress.post("refunds/alice/messages", {})).json()["messages"] + shown = (await ingress.post("refund_loop/alice/messages", {})).json()["messages"] tools = [(p["name"], p["output"]) for p in shown[1]["parts"]] assert tools == [ ("order_status", "Order 103 was delivered yesterday."), From 73b8c3733e7463896a9cc8f3235acccae81022e1 Mon Sep 17 00:00:00 2001 From: igalshilman Date: Mon, 9 Nov 2026 12:08:40 +0100 Subject: [PATCH 3/5] Examples: drop the test of unique names --- tests/examples_agents.py | 18 ------------------ 1 file changed, 18 deletions(-) diff --git a/tests/examples_agents.py b/tests/examples_agents.py index ae676633..dda25c92 100644 --- a/tests/examples_agents.py +++ b/tests/examples_agents.py @@ -80,24 +80,6 @@ async def serving(path: str, setup: Callable[[ModuleType], None] | None = None) yield Ingress(harness.ingress_url) -async def test_no_two_examples_serve_a_name(monkeypatch: pytest.MonkeyPatch): - """A name is served by one example only. Restate keeps a key's state, its database with it, from one example to - the next: an example of another's name would load that one's conversations. The tutorial's lessons are versions - of one application, and share their names.""" - for framework in ("agents", "langgraph", "langchain.agents", "pydantic_ai", "google.adk"): - pytest.importorskip(framework) - served: dict[str, str] = {} - for path in sorted(EXAMPLES.glob("*/*.py")): - example = str(path.relative_to(EXAMPLES)) - with monkeypatch.context() as env: - if example == "basic/adk_agent.py": - env.setenv("RESTATE_AGENT_MODEL", "google/gemini-x") # an ADK agent needs a model - module = load(example) - owner = "the tutorial" if example.startswith("tutorial/") else example - for name in {value.name for value in vars(module).values() if isinstance(value, SERVABLE)}: - assert served.setdefault(name, owner) == owner, f"{example} and {served[name]} both serve {name!r}" - - async def test_01_object(): async with serving("tutorial/01_object.py") as ingress: assert (await ingress.post("wallet/alice/hello")).json() == "Hello, alice!" From dd9ab566741187cc739af7f2c3eca3fa346bfb79 Mon Sep 17 00:00:00 2001 From: igalshilman Date: Mon, 9 Nov 2026 12:13:49 +0100 Subject: [PATCH 4/5] Agents: a conversation this agent cannot read fails for good, not retried forever A stored row that the agent's framework cannot load (one stored by an agent of another framework, under the same name) failed the handler with the framework's own error, and Restate retried it for good: every attempt read the same row. Every load of a stored item now goes through load_item, which fails with a terminal error that says so. --- python/restate/agents/_conversation.py | 22 ++++++++++++++++---- python/restate/agents/_messages.py | 6 +++--- python/restate/agents/pydantic_ai/_stores.py | 5 +++-- 3 files changed, 24 insertions(+), 9 deletions(-) diff --git a/python/restate/agents/_conversation.py b/python/restate/agents/_conversation.py index 49184ca8..6fea6d82 100644 --- a/python/restate/agents/_conversation.py +++ b/python/restate/agents/_conversation.py @@ -294,6 +294,20 @@ def summary_items(items: Items, summary: str) -> list[Item]: return [items.prompt(SUMMARY + summary), items.reply("Understood.")] +def load_item(items: Items, stored: str) -> Item: + """Return the item stored as ``stored``, or fail for good: a retry would read the same row again. + + A row that ``items`` cannot read was stored by another framework's agent of the same name, say. + """ + try: + return items.load(stored) + except Exception as error: + raise TerminalError( + f"This conversation holds a message that this agent cannot read (one stored by an agent of another " + f"framework, under the same name?): {type(error).__name__}: {error}" + ) from error + + @dataclass(frozen=True) class Start: """Where a conversation starts: its first position, the last ``seq`` given, and the summary of the earliest turns.""" @@ -370,7 +384,7 @@ async def latest_snapshot(db: DatabaseReader, items: Items) -> Snapshot | None: row = await db.get(f"SELECT covers, turn, items FROM {SNAPSHOTS} ORDER BY covers DESC, id DESC LIMIT 1") if row is None: return None - held = [items.load(str(stored)) for stored in json.loads(str(row["items"]))] + held = [load_item(items, str(stored)) for stored in json.loads(str(row["items"]))] return _snapshot_of(items, int(typing.cast(int, row["covers"])), _int(row["turn"]), held) @@ -948,7 +962,7 @@ async def load( if whole: stored = await db.values(f"SELECT n, item, reply, agent FROM {STEPS} WHERE n > ? ORDER BY n", [whole[0].n]) steps = [ - Entry(int(typing.cast(int, n)), items.load(str(item)), reply=_int(reply), agent=_str(agent)) + Entry(int(typing.cast(int, n)), load_item(items, str(item)), reply=_int(reply), agent=_str(agent)) for n, item, reply, agent in stored ] entries = sorted([*whole, *steps], key=_position) @@ -976,7 +990,7 @@ async def _from_snapshot(cls, db: Database, items: Items, start: Start, last: in if whole: stored = await db.values(f"SELECT n, item, reply, agent FROM {STEPS} WHERE n > ? ORDER BY n", [whole[0].n]) steps = [ - Entry(int(typing.cast(int, n)), items.load(str(item)), reply=_int(reply), agent=_str(agent)) + Entry(int(typing.cast(int, n)), load_item(items, str(item)), reply=_int(reply), agent=_str(agent)) for n, item, reply, agent in stored ] entries = sorted([*whole, *steps], key=_position) @@ -1245,7 +1259,7 @@ async def _transcript(db: DatabaseReader, items: Items, first: int, bounded: tup at = int(typing.cast(int, n)) entry = Entry( at, - items.load(str(item)), + load_item(items, str(item)), str(prompt), _str(answer), agent=_str(agent), diff --git a/python/restate/agents/_messages.py b/python/restate/agents/_messages.py index a9ba5d73..303fdd88 100644 --- a/python/restate/agents/_messages.py +++ b/python/restate/agents/_messages.py @@ -30,7 +30,7 @@ from dataclasses import dataclass, field from restate.agents._api import ChatMessage, Messages, MessagesRequest, Part, Role, ToolPart, text_part -from restate.agents._conversation import SCHEMA, STEPS, TURNS, Items, last_of, start_of +from restate.agents._conversation import SCHEMA, STEPS, TURNS, Items, last_of, load_item, start_of from restate.sessions import DatabaseReader @@ -50,7 +50,7 @@ def render(items: Items, rows: typing.Sequence[typing.Sequence[typing.Any]]) -> """Each tool call's message, by the call's id.""" for n, seq, stored in rows: position, changed = int(n), int(seq) - said, calls, results = items.read(items.load(str(stored))) + said, calls, results = items.read(load_item(items, str(stored))) if said is not None: role, text = said messages.append(_Message(position, "user" if role == "user" else "assistant")) @@ -187,7 +187,7 @@ def _summarized(items: Items, rows: typing.Sequence[typing.Sequence[typing.Any]] """ messages: list[_Message] = [] for n, seq, stored, answer, answered_at, answered_seq in rows: - said = items.read(items.load(str(stored))).said + said = items.read(load_item(items, str(stored))).said text = said[1] if said else "" messages.append(_Message(int(n), "user", [text_part(text)] if text.strip() else [], int(seq))) if answer is not None and answered_at is not None: diff --git a/python/restate/agents/pydantic_ai/_stores.py b/python/restate/agents/pydantic_ai/_stores.py index 95008fc6..b72be899 100644 --- a/python/restate/agents/pydantic_ai/_stores.py +++ b/python/restate/agents/pydantic_ai/_stores.py @@ -42,7 +42,7 @@ from pydantic_ai_harness.planning import PlanItem, TaskStatus from pydantic_ai_harness.step_persistence import RunRecord -from restate.agents._conversation import STEPS, TURNS, Item, Items +from restate.agents._conversation import STEPS, TURNS, Item, Items, load_item from restate.agents._tools import current_call from restate.sessions import SessionContext from restate.sessions.internal import Schema, definition_of @@ -322,7 +322,8 @@ async def run_history(self, *, run_id: str) -> list[ModelMessage]: steps = await db.values(f"SELECT item FROM {STEPS} WHERE n > ? ORDER BY n", [n]) else: steps = await db.values(f"SELECT item FROM {STEPS} WHERE n > ? AND n < ? ORDER BY n", [n, end]) - loaded: list[Item] = [self.items.load(str(row["item"]))] + [self.items.load(str(item)) for [item] in steps] + stored = [str(row["item"])] + [str(item) for [item] in steps] + loaded: list[Item] = [load_item(self.items, item) for item in stored] return typing.cast(list[ModelMessage], loaded) From da14dcef2d5e01c1608c6f21a9661bfec0f7ccb8 Mon Sep 17 00:00:00 2001 From: igalshilman Date: Mon, 9 Nov 2026 12:34:03 +0100 Subject: [PATCH 5/5] Agents: a conversation this agent cannot read: the error names another version of its framework too, and a test From the review of #303: load_item cannot tell another framework's row from one stored by another version of the agent's own framework, so the error says either. A test of the terminal error: an agent of another framework, under the same name, fails for good. --- python/restate/agents/_conversation.py | 8 ++++--- tests/agents_conversation.py | 32 ++++++++++++++++++++++++++ 2 files changed, 37 insertions(+), 3 deletions(-) diff --git a/python/restate/agents/_conversation.py b/python/restate/agents/_conversation.py index 6fea6d82..e75d5258 100644 --- a/python/restate/agents/_conversation.py +++ b/python/restate/agents/_conversation.py @@ -297,14 +297,16 @@ def summary_items(items: Items, summary: str) -> list[Item]: def load_item(items: Items, stored: str) -> Item: """Return the item stored as ``stored``, or fail for good: a retry would read the same row again. - A row that ``items`` cannot read was stored by another framework's agent of the same name, say. + A row that ``items`` cannot read was stored by another framework's agent of the same name, or by another + version of this framework, say. """ try: return items.load(stored) except Exception as error: raise TerminalError( - f"This conversation holds a message that this agent cannot read (one stored by an agent of another " - f"framework, under the same name?): {type(error).__name__}: {error}" + f"This conversation holds a message that this agent cannot read (stored by an agent of another " + f"framework under the same name, or by another version of its framework?): {type(error).__name__}: " + f"{error}" ) from error diff --git a/tests/agents_conversation.py b/tests/agents_conversation.py index 80853c24..76ee2eb1 100644 --- a/tests/agents_conversation.py +++ b/tests/agents_conversation.py @@ -228,6 +228,38 @@ async def failure(text: str, invocation: Any = None) -> tuple[str, int]: assert await failure("gone", gone) == (ENDED, 500) +class _OtherFramework(Items): + """Another framework's items, which have a ``kind``: it cannot read the made-up framework's.""" + + def load(self, stored: str) -> Item: + item = json.loads(stored) + if "kind" not in item: + raise ValueError("an item has a kind") + return item + + +async def test_a_conversation_this_agent_cannot_read_fails_for_good(): + """An agent of the same name on another framework fails with a terminal error: a retry would read the same row.""" + obj = sessions.SessionObject("asker") + + async def run(conversation: Conversation) -> str: + await conversation.turn.append_reply({"role": "assistant", "text": "hello"}) + return "hello" + + @obj.handler + async def ask(ctx: sessions.SessionContext, text: str) -> str: + return await answer_once(ctx, text, ITEMS, run) + + @obj.handler + async def ask_other(ctx: sessions.SessionContext, text: str) -> str: + return await answer_once(ctx, text, _OtherFramework(), run) + + backend = Backend() + assert await invoke(obj, "ask", Host(backend, service="asker", answers=PUBLISHED), request(), "hi") == "hello" + with pytest.raises(restate.TerminalError, match="cannot read .* ValueError: an item has a kind"): + await invoke(obj, "ask_other", Host(backend, service="asker", answers=PUBLISHED), request(), "hi") + + async def test_a_dropped_reply_takes_what_followed_it_and_each_item_is_read_once(): reads: list[Item] = []