From bd0e82982f0ccb9dca69db42fdf78634fc3191b0 Mon Sep 17 00:00:00 2001 From: Nan Yu Date: Fri, 16 Oct 2026 18:50:51 +0000 Subject: [PATCH 1/3] feat(python/agent)!: bind a RendererState to A2uiRequestProcessor A parser holds no state now, so the processor creates one and reuses it, and the caller chooses how long the renderer state lives. 1. `parser` returns the one parser the processor reuses. `create_parser()` is removed. 2. `create_state()` creates an empty `RendererState` that holds the active catalogs. 3. `A2uiRequestProcessor(state=...)` and `A2uiGenerator.create_processor(..., state=...)` bind a state, so a canvas agent's `parse_response` and the new `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 state that holds other catalogs than the processor is rejected. Part of #3263. Signed-off-by: Nan Yu --- python/a2ui_a2a/CHANGELOG.md | 2 +- python/a2ui_a2a/src/a2ui/a2a/parts.py | 5 +- python/a2ui_agent/CHANGELOG.md | 10 ++ .../src/a2ui/processor/generator.py | 11 +- .../src/a2ui/processor/processor.py | 104 ++++++++++++++++-- .../parser/test_renderer_state_parsing.py | 45 ++++++++ 6 files changed, 163 insertions(+), 14 deletions(-) diff --git a/python/a2ui_a2a/CHANGELOG.md b/python/a2ui_a2a/CHANGELOG.md index b1ec98bbc0..6cf2ad5c79 100644 --- a/python/a2ui_a2a/CHANGELOG.md +++ b/python/a2ui_a2a/CHANGELOG.md @@ -12,7 +12,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. diff --git a/python/a2ui_a2a/src/a2ui/a2a/parts.py b/python/a2ui_a2a/src/a2ui/a2a/parts.py index 1e1871e810..d5e74a3a80 100644 --- a/python/a2ui_a2a/src/a2ui/a2a/parts.py +++ b/python/a2ui_a2a/src/a2ui/a2a/parts.py @@ -170,8 +170,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`. diff --git a/python/a2ui_agent/CHANGELOG.md b/python/a2ui_agent/CHANGELOG.md index 861eb0fe2e..7ec1be6250 100644 --- a/python/a2ui_agent/CHANGELOG.md +++ b/python/a2ui_agent/CHANGELOG.md @@ -61,10 +61,20 @@ raises `A2uiParseError` if a block is still open, and checks the tree of an unwrapped stream. `ResponseStream` is exported from `a2ui.parser` and `DirectJsonResponseStream` from `a2ui.inference_formats.direct_json`. +- `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`. - **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 `parser.open_stream(wrapped=...)` and call `feed(chunk)`, then `close()`. + - `A2uiRequestProcessor.create_parser()`: use the `processor.parser` + property, which returns the one parser the processor reuses, or + `processor.parse_response` and `processor.open_stream`. - The `parser` property of `DirectJsonFormat`, `ExpressFormat`, `ElementalFormat` and `AtomFormat`: use `create_parser()`. - `validate_payload(..., surface_catalog_ids=...)`: check the payload diff --git a/python/a2ui_agent/src/a2ui/processor/generator.py b/python/a2ui_agent/src/a2ui/processor/generator.py index 16b64981e4..0b55bcfa70 100644 --- a/python/a2ui_agent/src/a2ui/processor/generator.py +++ b/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, ) diff --git a/python/a2ui_agent/src/a2ui/processor/processor.py b/python/a2ui_agent/src/a2ui/processor/processor.py index 52d20785ad..ec01110ba5 100644 --- a/python/a2ui_agent/src/a2ui/processor/processor.py +++ b/python/a2ui_agent/src/a2ui/processor/processor.py @@ -22,8 +22,8 @@ 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,6 +38,8 @@ 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. @@ -45,9 +47,14 @@ def __init__( 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. 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 +73,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 +94,91 @@ 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`. - 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) + 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 _check_state(self, state: RendererState | None) -> None: + if state is None: + return + held = [(c.catalog_id, str(c.protocol_version)) for c in state.catalogs] + active = [ + (c.catalog_id, str(c.protocol_version)) for c in self._active_catalogs + ] + if held != active: + raise A2uiCatalogError( + f"The renderer state holds catalogs {held}, but the processor's" + f" active catalogs are {active}." + ) diff --git a/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py b/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py index 8da19d2177..eb4d21fdee 100644 --- a/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py +++ b/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py @@ -32,6 +32,7 @@ from a2ui.inference_formats import to_message_dicts from a2ui.inference_formats.direct_json import DirectJsonParser from a2ui.parser import A2uiPart, ResponsePart, TextPart +from a2ui.processor import A2uiRequestProcessor from a2ui.utils import RendererState, validate_payload _BASIC = BasicCatalog("0.9") @@ -214,6 +215,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]) @@ -267,6 +292,26 @@ 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_stream_that_raises_has_ended(): stream = DirectJsonParser([_BASIC]).open_stream() stream.feed("" + json.dumps([_CREATE, _update(_column("a"))])) From a67204fec5ac2ccb30469d2ece99f95c850de3f4 Mon Sep 17 00:00:00 2001 From: Nan Yu Date: Sat, 17 Oct 2026 00:36:17 +0000 Subject: [PATCH 2/3] fix(python/agent): normalize protocol_version in A2uiRequestProcessor._check_state --- python/a2ui_agent/src/a2ui/processor/processor.py | 9 +++++++-- .../tests/parser/test_renderer_state_parsing.py | 10 ++++++++++ 2 files changed, 17 insertions(+), 2 deletions(-) diff --git a/python/a2ui_agent/src/a2ui/processor/processor.py b/python/a2ui_agent/src/a2ui/processor/processor.py index ec01110ba5..3fd6172fe3 100644 --- a/python/a2ui_agent/src/a2ui/processor/processor.py +++ b/python/a2ui_agent/src/a2ui/processor/processor.py @@ -19,6 +19,7 @@ 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 @@ -173,9 +174,13 @@ def _resolve_state(self, state: RendererState | None) -> RendererState | None: def _check_state(self, state: RendererState | None) -> None: if state is None: return - held = [(c.catalog_id, str(c.protocol_version)) for c in state.catalogs] + held = [ + (c.catalog_id, to_protocol_version(c.protocol_version).value) + for c in state.catalogs + ] active = [ - (c.catalog_id, str(c.protocol_version)) for c in self._active_catalogs + (c.catalog_id, to_protocol_version(c.protocol_version).value) + for c in self._active_catalogs ] if held != active: raise A2uiCatalogError( diff --git a/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py b/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py index 51f0ba8f80..1984758657 100644 --- a/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py +++ b/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py @@ -329,6 +329,16 @@ def test_a_bound_state_for_other_catalogs_is_rejected(): 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_a_stream_that_raises_has_ended(): stream = DirectJsonParser([_BASIC]).open_stream() stream.feed("" + json.dumps([_CREATE, _update(_column("a"))])) From bf1640e11bb4d7d25372cca470bc6e3b7886abd9 Mon Sep 17 00:00:00 2001 From: Nan Yu Date: Mon, 19 Oct 2026 17:42:53 +0000 Subject: [PATCH 3/3] feat(python/agent): add A2uiRequestProcessor.accepts_state - accepts_state(state) tells whether a state holds the processor's active catalogs in order, so an agent that keeps a state per conversation can tell when to make a new one; _check_state raises when it's False. - Say that a processor bound to a state serves only that conversation. - Test that A2uiGenerator.create_processor forwards state, and rejects a state for other catalogs. Signed-off-by: Nan Yu --- python/a2ui_agent/CHANGELOG.md | 4 +- .../src/a2ui/processor/processor.py | 42 ++++++++++++------- .../parser/test_renderer_state_parsing.py | 11 +++++ .../tests/processor/test_request_processor.py | 22 ++++++++++ 4 files changed, 63 insertions(+), 16 deletions(-) diff --git a/python/a2ui_agent/CHANGELOG.md b/python/a2ui_agent/CHANGELOG.md index 5a93329bf1..a5fd6c3755 100644 --- a/python/a2ui_agent/CHANGELOG.md +++ b/python/a2ui_agent/CHANGELOG.md @@ -78,7 +78,9 @@ `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`. + 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 diff --git a/python/a2ui_agent/src/a2ui/processor/processor.py b/python/a2ui_agent/src/a2ui/processor/processor.py index 3fd6172fe3..f8b32a6a71 100644 --- a/python/a2ui_agent/src/a2ui/processor/processor.py +++ b/python/a2ui_agent/src/a2ui/processor/processor.py @@ -51,7 +51,8 @@ def __init__( 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. + 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, or if `state` @@ -171,19 +172,30 @@ def _resolve_state(self, state: RendererState | None) -> RendererState | None: 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: + if state is None or self.accepts_state(state): return - held = [ - (c.catalog_id, to_protocol_version(c.protocol_version).value) - for c in state.catalogs - ] - active = [ - (c.catalog_id, to_protocol_version(c.protocol_version).value) - for c in self._active_catalogs - ] - if held != active: - raise A2uiCatalogError( - f"The renderer state holds catalogs {held}, but the processor's" - f" active catalogs are {active}." - ) + 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 _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 + ] diff --git a/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py b/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py index 93f7c9feac..0306fcd515 100644 --- a/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py +++ b/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py @@ -339,6 +339,17 @@ def test_a_bound_state_normalizes_protocol_version_spelling(): 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("" + json.dumps([_CREATE, _update(_column("a"))])) diff --git a/python/a2ui_agent/tests/processor/test_request_processor.py b/python/a2ui_agent/tests/processor/test_request_processor.py index 44a57ed071..95a66a1330 100644 --- a/python/a2ui_agent/tests/processor/test_request_processor.py +++ b/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)