Agents: a conversation this agent cannot read fails for good; examples of unique names; no Pydantic AI banner

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

examples-agents/basic/adk_agent.py

@@ -6,8 +6,8 @@
     GOOGLE_API_KEY=... RESTATE_AGENT_MODEL=google/<model-id> 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/<model-id>``). 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__":

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])))

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/<model-id>`` 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__":

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__":

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/<model-id>`` 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__":

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])))

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])))

python/restate/agents/_conversation.py

@@ -294,6 +294,22 @@ 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, 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 (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
+
+
 @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 +386,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 +964,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 +992,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 +1261,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),

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:

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

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)
 
 

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] = []