feat(python/agent)!: bind a RendererState to A2uiRequestProcessor

python/a2ui_a2a/CHANGELOG.md

@@ -14,7 +14,7 @@
   URI such as `.../a2ui/vv0.9`.
 - **BREAKING**: `stream_response_to_parts(stream, token_stream, version=None)`
   takes the `ResponseStream` that reads the response, such as one from
-  `Parser.open_stream()`, instead of a `Parser`. It closes the
+  `A2uiRequestProcessor.open_stream()`, instead of a `Parser`. It closes the
   stream after the last token, so text held back as a possible tag is emitted
   and a block left open raises. A canvas agent opens the stream with the
   `RendererState` it keeps.

python/a2ui_a2a/src/a2ui/a2a/parts.py

@@ -176,8 +176,9 @@ async def stream_response_to_parts(
 
     Args:
         stream: The stream that reads the response, from
-            `Parser.open_stream`. The state it was opened with is the one the
-            response is checked against.
+            `A2uiRequestProcessor.open_stream` or `Parser.open_stream`. The
+            state it was opened with is the one the response is checked
+            against.
         token_stream: An async iterable of strings (tokens).
         version: The A2UI version, as for `create_a2ui_part`.
 

python/a2ui_agent/CHANGELOG.md

@@ -70,6 +70,15 @@
   mappings. A message that matches no protocol version's message schema
   raises `A2uiValidationError` there. Use `to_message_dicts` for the wire
   form.
+- `A2uiRequestProcessor.parse_response` and the new
+  `A2uiRequestProcessor.open_stream` take an optional `state`.
+  `A2uiRequestProcessor.create_state` creates an empty one, and
+  `A2uiRequestProcessor(state=...)` and
+  `A2uiGenerator.create_processor(..., state=...)` bind one so that calls
+  need not pass it. A state that holds other catalogs than the processor is
+  rejected with `A2uiCatalogError`. `A2uiRequestProcessor.accepts_state`
+  tells whether a state holds the processor's catalogs, so an agent that
+  keeps a state per conversation can tell when to make a new one.
 - **BREAKING**: A parser holds no state, and the APIs that assumed one parser
   per response are removed. To migrate:
   - `Parser.parse_chunk(chunk, wrapped)`: open a stream with

python/a2ui_agent/src/a2ui/processor/generator.py

@@ -22,7 +22,7 @@
 from a2ui.core.schema import A2uiRendererCapabilities, AgentToRendererMessage
 from a2ui.inference_format import InferenceFormatFactory
 from a2ui.inference_formats.direct_json import DirectJsonFormatFactory
-from a2ui.utils import resolve_catalogs
+from a2ui.utils import RendererState, resolve_catalogs
 
 from .catalog_config import CatalogConfig
 from .processor import A2uiRequestProcessor
@@ -72,6 +72,8 @@ def create_processor(
         self,
         renderer_capabilities: A2uiRendererCapabilities,
         inference_format_factory: InferenceFormatFactory | None = None,
+        *,
+        state: RendererState | None = None,
     ) -> A2uiRequestProcessor:
         """Creates an A2uiRequestProcessor bound to the specified renderer capabilities.
 
@@ -80,12 +82,16 @@ def create_processor(
                 a processor is negotiated for one renderer, so a request without capabilities
                 is a catalog error here, although resolve_catalogs accepts one.
             inference_format_factory: Optional override format factory for this processor.
+            state: The renderer state a canvas agent keeps for the
+                conversation, bound to the processor so that its
+                `parse_response` and `open_stream` use it.
 
         Returns:
             Pre-negotiated client-bound A2uiRequestProcessor instance.
 
         Raises:
-            A2uiCatalogError: If renderer_capabilities is None or no catalog can be resolved.
+            A2uiCatalogError: If renderer_capabilities is None, no catalog can be
+                resolved, or `state` holds other catalogs than the resolved ones.
             A2uiValidationError: If renderer_capabilities or prompt examples are invalid.
         """
         if renderer_capabilities is None:
@@ -106,4 +112,5 @@ def create_processor(
             catalogs=active_catalogs,
             examples=self.examples,
             format_factory=factory,
+            state=state,
         )

python/a2ui_agent/src/a2ui/processor/processor.py

@@ -19,11 +19,12 @@
 from collections.abc import Sequence
 
 from a2ui.core import A2uiCatalogError, CatalogApi
+from a2ui.core.common import to_protocol_version
 from a2ui.core.schema import AgentToRendererMessage
 from a2ui.inference_format import InferenceFormat, InferenceFormatFactory
 from a2ui.inference_formats.direct_json import DirectJsonFormatFactory
-from a2ui.parser import Parser, ResponsePart
-from a2ui.utils import validate_payload
+from a2ui.parser import Parser, ResponsePart, ResponseStream
+from a2ui.utils import RendererState, validate_payload
 
 __all__ = [
     "A2uiRequestProcessor",
@@ -38,16 +39,24 @@ def __init__(
         catalogs: Sequence[CatalogApi],
         examples: Sequence[Sequence[AgentToRendererMessage]] | None = None,
         format_factory: InferenceFormatFactory | None = None,
+        *,
+        state: RendererState | None = None,
     ):
         """Initializes A2uiRequestProcessor, resolving active catalogs and instantiating validator and format strategy.
 
         Args:
             catalogs: List of active Catalog instances.
             examples: Optional list of prompt example turns.
             format_factory: Format factory for instantiating format strategies.
+            state: The renderer state a canvas agent keeps for the
+                conversation. `parse_response` and `open_stream` use it when
+                a call passes none. Without one, each call checks the response
+                against a new state, as a chat agent wants. A processor bound
+                to a state serves only that conversation.
 
         Raises:
-            A2uiCatalogError: If no active catalogs are provided.
+            A2uiCatalogError: If no active catalogs are provided, or if `state`
+                holds other catalogs.
             A2uiValidationError: If any prompt example fails validation against the active catalogs.
         """
         if not catalogs:
@@ -66,6 +75,9 @@ def __init__(
             catalogs=self._active_catalogs,
             examples=self._examples,
         )
+        self._parser: Parser | None = None
+        self._check_state(state)
+        self._state = state
 
     @property
     def active_catalogs(self) -> list[CatalogApi]:
@@ -84,15 +96,106 @@ def format(self) -> InferenceFormat:
         """Returns the InferenceFormat bound to this processor."""
         return self._format
 
+    @property
+    def state(self) -> RendererState | None:
+        """The renderer state bound to this processor, if any."""
+        return self._state
+
     @property
     def prompt_snippet(self) -> str:
         """Format-specific system prompt instruction snippet."""
         return self._format.prompt_generator.generate()
 
-    def create_parser(self) -> Parser:
-        """Creates a fresh Parser instance bound to the active catalogs."""
-        return self._format.create_parser()
+    @property
+    def parser(self) -> Parser:
+        """The parser for the format, bound to the active catalogs.
+
+        A parser holds no state, so the processor creates one and reuses it.
+        """
+        if self._parser is None:
+            self._parser = self._format.create_parser()
+        return self._parser
+
+    def create_state(self) -> RendererState:
+        """Creates an empty renderer state that holds the active catalogs.
+
+        A canvas agent keeps the state for the whole conversation and passes
+        it to `parse_response` or `open_stream` on every turn.
+        """
+        return RendererState(self._active_catalogs)
+
+    def parse_response(
+        self,
+        content: str,
+        wrapped: bool = True,
+        *,
+        state: RendererState | None = None,
+    ) -> list[ResponsePart]:
+        """Parses and validates the LLM response into ResponseParts.
+
+        Args:
+            content: The response text.
+            wrapped: False if the whole response is one block, without tags.
+            state: The renderer state to check the response against and apply
+                it to. Defaults to the state bound to the processor. With
+                neither, the response is checked on its own, against a new
+                state.
+
+        Raises:
+            A2uiCatalogError: If the state holds other catalogs than the
+                processor.
+        """
+        target = self._resolve_state(state)
+        return self.parser.parse_response(content, wrapped, state=target)
+
+    def open_stream(
+        self, *, wrapped: bool = True, state: RendererState | None = None
+    ) -> ResponseStream:
+        """Opens a stream that reads one LLM response chunk by chunk.
+
+        Args:
+            wrapped: False if the whole response is one block, without tags.
+            state: The renderer state to check the response against and apply
+                it to, as for `parse_response`.
+
+        Raises:
+            A2uiCatalogError: If the state holds other catalogs than the
+                processor.
+            NotImplementedError: If the format cannot be read incrementally.
+        """
+        target = self._resolve_state(state)
+        return self.parser.open_stream(wrapped=wrapped, state=target)
+
+    def _resolve_state(self, state: RendererState | None) -> RendererState | None:
+        if state is None:
+            return self._state
+        self._check_state(state)
+        return state
+
+    def accepts_state(self, state: RendererState) -> bool:
+        """Whether the processor can check responses against a renderer state.
+
+        A state is accepted when it holds the processor's active catalogs, in
+        the same order, since the order decides which catalog a surface
+        resolves to. A state made with `create_state` is always accepted. An
+        agent that keeps a state per conversation calls this to tell whether
+        the state still fits the request's catalogs, and makes a new one with
+        `create_state` when it doesn't.
+        """
+        return _catalog_keys(state.catalogs) == _catalog_keys(self._active_catalogs)
+
+    def _check_state(self, state: RendererState | None) -> None:
+        if state is None or self.accepts_state(state):
+            return
+        raise A2uiCatalogError(
+            f"The renderer state holds catalogs {_catalog_keys(state.catalogs)},"
+            " but the processor's active catalogs are"
+            f" {_catalog_keys(self._active_catalogs)}."
+        )
+
 
-    def parse_response(self, content: str, wrapped: bool = True) -> list[ResponsePart]:
-        """Parses and validates the LLM response into ResponseParts."""
-        return self._format.create_parser().parse_response(content, wrapped=wrapped)
+def _catalog_keys(catalogs: Sequence[CatalogApi]) -> list[tuple[str, str]]:
+    """The id and normalized protocol version of each catalog, in order."""
+    return [
+        (c.catalog_id, to_protocol_version(c.protocol_version).value) for c in catalogs
+    ]

python/a2ui_agent/tests/parser/test_renderer_state_parsing.py

@@ -22,15 +22,18 @@
 import pytest
 
 from a2ui.core import (
+    A2uiCatalogError,
     A2uiIntegrityError,
     A2uiParseError,
     A2uiValidationError,
+    Catalog,
 )
 from a2ui.core.basic_catalog import BasicCatalog
 from a2ui.inference_formats import to_message_dicts
 from a2ui.inference_formats.direct_json import DirectJsonParser
 from a2ui.inference_formats.express import ExpressParser
 from a2ui.parser import A2uiPart, ResponsePart, TextPart
+from a2ui.processor import A2uiRequestProcessor
 from a2ui.utils import RendererState, validate_payload
 
 _BASIC = BasicCatalog("0.9")
@@ -229,6 +232,30 @@ def test_feed_after_close_is_an_error():
         stream.feed("more")
 
 
+def test_request_processor_parses_canvas_turns_against_its_state():
+    processor = A2uiRequestProcessor([_BASIC])
+    state = processor.create_state()
+    processor.parse_response(
+        _block(_CREATE, _update(_column("a"), _text("a"))), state=state
+    )
+
+    stream = processor.open_stream(state=state)
+    stream.feed(_block(_update(_column("a", "b"), _text("b"))))
+    stream.close()
+
+    assert _component_ids(state) == {"root", "a", "b"}
+    with pytest.raises(A2uiIntegrityError, match="Dangling reference"):
+        processor.parse_response(_block(_update(_column("missing"))), state=state)
+
+
+def test_request_processor_rejects_a_state_for_other_catalogs():
+    processor = A2uiRequestProcessor([_BASIC])
+    other = RendererState([Catalog(catalog_id="other", protocol_version="0.9")])
+
+    with pytest.raises(A2uiCatalogError, match="active catalogs"):
+        processor.parse_response(_block(_CREATE), state=other)
+
+
 def test_every_version_a_stream_emitted_rebuilds_its_state():
     parser = DirectJsonParser([_BASIC], progressive_keys={"text"})
     state = RendererState([_BASIC])
@@ -282,6 +309,47 @@ def test_a_response_may_reference_a_component_a_later_block_adds():
         parser.parse_response(first)
 
 
+def test_a_bound_state_is_used_when_a_call_passes_none():
+    state = RendererState([_BASIC])
+    processor = A2uiRequestProcessor([_BASIC], state=state)
+
+    processor.parse_response(_block(_CREATE, _update(_column("a"), _text("a"))))
+    stream = processor.open_stream()
+    stream.feed(_block(_update(_column("a", "b"), _text("b"))))
+    stream.close()
+
+    assert processor.state is state
+    assert _component_ids(state) == {"root", "a", "b"}
+
+
+def test_a_bound_state_for_other_catalogs_is_rejected():
+    other = RendererState([Catalog(catalog_id="other", protocol_version="0.9")])
+
+    with pytest.raises(A2uiCatalogError, match="active catalogs"):
+        A2uiRequestProcessor([_BASIC], state=other)
+
+
+def test_a_bound_state_normalizes_protocol_version_spelling():
+    raw = Catalog(catalog_id=_BASIC.catalog_id, protocol_version="0.9")
+    prefixed = Catalog(catalog_id=_BASIC.catalog_id, protocol_version="v0.9")
+    state = RendererState([raw])
+
+    processor = A2uiRequestProcessor([prefixed], state=state)
+
+    assert processor.state is state
+
+
+def test_accepts_state_tells_whether_a_state_fits_the_active_catalogs():
+    other = Catalog(catalog_id="other", protocol_version="0.9")
+    processor = A2uiRequestProcessor([_BASIC, other])
+
+    assert processor.accepts_state(processor.create_state())
+    assert processor.accepts_state(RendererState([_BASIC, other]))
+    # The order decides which catalog a surface resolves to.
+    assert not processor.accepts_state(RendererState([other, _BASIC]))
+    assert not processor.accepts_state(RendererState([_BASIC]))
+
+
 def test_a_stream_that_raises_has_ended():
     stream = DirectJsonParser([_BASIC]).open_stream()
     stream.feed("<a2ui-json>" + json.dumps([_CREATE, _update(_column("a"))]))

python/a2ui_agent/tests/processor/test_request_processor.py

@@ -26,6 +26,7 @@
 from a2ui.inference_formats.express import ExpressFormatFactory
 from a2ui.parser import A2uiPart, TextPart
 from a2ui.processor import A2uiGenerator, A2uiRequestProcessor, CatalogConfig
+from a2ui.utils import RendererState
 
 
 def test_request_processor_rejects_empty_catalogs():
@@ -155,3 +156,24 @@ def test_processor_validates_examples_against_active_catalogs():
     ])
     with pytest.raises(A2uiValidationError):
         A2uiRequestProcessor(catalogs=[pruned], examples=[bad_example])
+
+
+def test_generator_binds_the_state_to_the_processor():
+    basic = BasicCatalog("v1.0")
+    generator = A2uiGenerator(catalogs=[CatalogConfig(basic)])
+    capabilities = {"v1.0": {"supportedCatalogIds": [basic.catalog_id]}}
+    state = generator.create_processor(capabilities).create_state()
+
+    processor = generator.create_processor(capabilities, state=state)
+
+    assert processor.state is state
+
+
+def test_generator_rejects_a_state_for_other_catalogs():
+    basic = BasicCatalog("v1.0")
+    generator = A2uiGenerator(catalogs=[CatalogConfig(basic)])
+    capabilities = {"v1.0": {"supportedCatalogIds": [basic.catalog_id]}}
+    other = RendererState([Catalog(catalog_id="other", protocol_version="1.0")])
+
+    with pytest.raises(A2uiCatalogError, match="active catalogs"):
+        generator.create_processor(capabilities, state=other)