diff --git a/python/a2ui_a2a/CHANGELOG.md b/python/a2ui_a2a/CHANGELOG.md index 6e2acc4125..ca09e3ef2a 100644 --- a/python/a2ui_a2a/CHANGELOG.md +++ b/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. diff --git a/python/a2ui_a2a/src/a2ui/a2a/parts.py b/python/a2ui_a2a/src/a2ui/a2a/parts.py index fc2ae093b5..0449a4a33e 100644 --- a/python/a2ui_a2a/src/a2ui/a2a/parts.py +++ b/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`. diff --git a/python/a2ui_agent/CHANGELOG.md b/python/a2ui_agent/CHANGELOG.md index cad70acd55..3d75f4ff4d 100644 --- a/python/a2ui_agent/CHANGELOG.md +++ b/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 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..f8b32a6a71 100644 --- a/python/a2ui_agent/src/a2ui/processor/processor.py +++ b/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,6 +39,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 +48,15 @@ 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. 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 + ] 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 bdeb1c5a2e..0306fcd515 100644 --- a/python/a2ui_agent/tests/parser/test_renderer_state_parsing.py +++ b/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("" + 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)